diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
index e11f9b7056..021352a79c 100644
--- a/rxjava-core/src/main/java/rx/Observable.java
+++ b/rxjava-core/src/main/java/rx/Observable.java
@@ -26,6 +26,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import rx.joins.Pattern2;
 import rx.joins.Plan0;
@@ -106,7 +108,6 @@
 import rx.operators.SafeObservableSubscription;
 import rx.operators.SafeObserver;
 import rx.plugins.RxJavaErrorHandler;
-import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
 import rx.schedulers.Schedulers;
 import rx.subjects.AsyncSubject;
@@ -137,11 +138,11 @@
 /**
  * The Observable interface that implements the Reactive Pattern.
  * <p>
- * This interface provides overloaded methods for subscribing as well as
- * delegate methods to the various operators.
+ * This interface provides overloaded methods for subscribing as well as delegate methods to the
+ * various operators.
  * <p>
- * The documentation for this interface makes use of marble diagrams. The
- * following legend explains these diagrams:
+ * The documentation for this interface makes use of marble diagrams. The following legend explains
+ * these diagrams:
  * <p>
  * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/legend.png">
  * <p>
@@ -152,40 +153,321 @@
  */
 public class Observable<T> {
 
-    private final static ConcurrentHashMap<Class, Boolean> internalClassMap = new ConcurrentHashMap<Class, Boolean>();
+    /**
+     * Executed when 'unsubscribe' is invoked on a {@link PartialSubscription}.
+     */
+    public interface OnPartialUnsubscribeFunc extends Function {
+        void onUnsubscribe();
+    }
+
+    /**
+     * Executed when 'subscribe' is invoked on a {@link PartialSubscription}.
+     */
+    public interface OnPartialSubscribeFunc<T> extends Function {
+        void onSubscribe(Observer<? super T> observer);
+    }
+
+    final static ConcurrentHashMap<Class, Boolean> internalClassMap = new ConcurrentHashMap<Class, Boolean>();
 
     /**
-     * Executed when 'subscribe' is invoked.
+     * Executed when 'getSubscription' is invoked.
      */
-    private final OnSubscribeFunc<T> onSubscribe;
+    private final OnGetSubscriptionFunc<T> onGetSubscription;
 
     /**
-     * Function interface for work to be performed when an {@link Observable}
-     * is subscribed to via {@link Observable#subscribe(Observer)}
+     * Function interface for work to be performed when an {@link Observable} is subscription is
+     * created via {@link Observable#getSubscription()}
      * 
      * @param <T>
      */
-    public static interface OnSubscribeFunc<T> extends Function {
-
-        public Subscription onSubscribe(Observer<? super T> t1);
+    public static interface OnGetSubscriptionFunc<T> extends Function {
+        public PartialSubscription<T> onGetSubscription();
+    }
 
+    /**
+     * Function interface for work to be performed when an {@link Observable} is subscribed to via
+     * {@link Observable#subscribe(Observer)}
+     * 
+     * @param <T>
+     * @deprecated use OnGetSubscriptionFunc
+     */
+    @Deprecated
+    public static interface OnSubscribeFunc<T> extends Function {
+        public Subscription onSubscribe(Observer<? super T> observer);
     }
 
     /**
-     * Observable with Function to execute when subscribed to.
+     * Observable with Function to execute when subscription is created but before work has started.
      * <p>
-     * NOTE: Use {@link #create(OnSubscribeFunc)} to create an Observable
-     * instead of this constructor unless you specifically have a need for
-     * inheritance.
+     * NOTE: Use {@link #create(OnGetSubscriptionFunc)} to create an Observable instead of this
+     * constructor unless you specifically have a need for inheritance.
      * 
-     * @param onSubscribe {@link OnSubscribeFunc} to be executed when
-     *                    {@link #subscribe(Observer)} is called
+     * @param onGetSubscription
+     *            {@link OnGetSubscriptionFunc} to be executed when {@link #getSubscription()} is
+     *            called
      */
-    protected Observable(OnSubscribeFunc<T> onSubscribe) {
-        this.onSubscribe = onSubscribe;
+    protected Observable(OnGetSubscriptionFunc<T> onGetSubscription) {
+        this.onGetSubscription = onGetSubscription;
+    }
+
+    public static final class PartialSubscription<T> implements Subscription {
+        private AtomicReference<Observer<? super T>> subscribedObserver = new AtomicReference<Observer<? super T>>(null);
+        private static final Observer UNSUBSCRIBED_OBSERVER = new Observer() {
+            @Override
+            public void onCompleted() {
+            }
+
+            @Override
+            public void onError(Throwable e) {
+            }
+
+            @Override
+            public void onNext(Object args) {
+            }
+        };
+
+        private final OnPartialSubscribeFunc<T> onPartialSubscribe;
+
+        private final OnPartialUnsubscribeFunc onPartialUnsubscribe;
+
+        public PartialSubscription(OnPartialSubscribeFunc<T> onPartialSubscribe, OnPartialUnsubscribeFunc onPartialUnsubscribe) {
+            this.onPartialSubscribe = onPartialSubscribe;
+            this.onPartialUnsubscribe = onPartialUnsubscribe;
+        }
+
+        private boolean unsubscribed = false;
+        
+        @Override
+        @SuppressWarnings("unchecked")
+        public void unsubscribe() {
+            Observer<? super T> observer = subscribedObserver.get();
+            if (observer == null) {
+                subscribedObserver.compareAndSet(null, UNSUBSCRIBED_OBSERVER);
+                observer = subscribedObserver.get();
+            }
+            try {
+                onPartialUnsubscribe.onUnsubscribe();
+            } catch (Throwable e) {
+                observer.onError(e);
+            } finally {
+                unsubscribed = true;
+            }
+            
+        }
+        
+        public boolean isUnsubscribed() {
+            return unsubscribed;
+        }
+
+        /**
+         * Whether a given {@link Function} is an internal implementation inside
+         * rx.* packages or not.
+         * <p>
+         * For why this is being used see https://github.com/Netflix/RxJava/issues/216 for
+         * discussion on "Guideline 6.4: Protect calls to user code from within an operator"
+         * 
+         * Note: If strong reasons for not depending on package names comes up then the
+         * implementation of this method can change to looking for a marker interface.
+         * 
+         * @param o
+         * @return {@code true} if the given function is an internal implementation,
+         *         and {@code false} otherwise.
+         */
+        private boolean isInternalImplementation(Object o) {
+            if (o == null) {
+                return true;
+            }
+            // prevent double-wrapping (yeah it happens)
+            if (o instanceof SafeObserver) {
+                return true;
+            }
+
+            Class<?> clazz = o.getClass();
+            if (internalClassMap.containsKey(clazz)) {
+                // don't need to do reflection
+                return internalClassMap.get(clazz);
+            } else {
+                // we treat the following package as "internal" and don't wrap it
+                Package p = o.getClass().getPackage(); // it can be null
+                Boolean isInternal = (p != null && p.getName().startsWith("rx.operators"));
+                internalClassMap.put(clazz, isInternal);
+                return isInternal;
+            }
+        }
+
+        /**
+         * Allow the {@link RxJavaErrorHandler} to receive the exception from
+         * onError.
+         * 
+         * @param e
+         */
+        private void handleError(Throwable e) {
+            // onError should be rare so we'll only fetch when needed
+            RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
+        }
+
+        public void subscribe(Observer<? super T> observer) {
+            if (!subscribedObserver.compareAndSet(null, observer)) {
+                // someone else has already subscribed to this PartialSubscription
+                if (subscribedObserver.get() == UNSUBSCRIBED_OBSERVER) {
+                    // this means that the subscription was unsubscribed before the subscribe was called.
+                    return;
+                }
+                else {
+                    throw new IllegalStateException("A PartialSubscription can only be subscribed to once. The first Oserver is "+ subscribedObserver.get() +" the second Observer is "+ observer);
+                }
+            }
+
+            // allow the hook to intercept and/or decorate
+            OnPartialSubscribeFunc<T> onPartialSubscribeFunction = onPartialSubscribe;// hook.onPartialSubscribeStart(this,
+            // onPartialSubscribe);
+            // validate and proceed
+            if (observer == null) {
+                throw new IllegalArgumentException("observer can not be null");
+            }
+            if (onPartialSubscribeFunction == null) {
+                throw new IllegalStateException("onSubscribe function can not be null.");
+                // the subscribe function can also be overridden but generally that's not the
+                // appropriate approach so I won't mention that in the exception
+            }
+            try {
+                /**
+                 * See https://github.com/Netflix/RxJava/issues/216 for discussion on
+                 * "Guideline 6.4: Protect calls to user code from within an operator"
+                 */
+                if (isInternalImplementation(observer)) {
+                    onPartialSubscribeFunction.onSubscribe(observer);
+                } else {
+                    onPartialSubscribeFunction.onSubscribe(new SafeObserver<T>(this, observer));
+                }
+            } catch (OnErrorNotImplementedException e) {
+                // special handling when onError is not implemented ... we just rethrow
+                throw e;
+            } catch (Throwable e) {
+                // if an unhandled error occurs executing the onSubscribe we will propagate it
+                try {
+                    observer.onError(e);// hook.onSubscribeError(this, e));
+                } catch (OnErrorNotImplementedException e2) {
+                    // special handling when onError is not implemented ... we just rethrow
+                    throw e2;
+                } catch (Throwable e2) {
+                    // if this happens it means the onError itself failed (perhaps an invalid
+                    // function implementation)
+                    // so we are unable to propagate the error correctly and will just throw
+                    RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
+                    // hook.onSubscribeError(this, r);
+                    throw r;
+                }
+            }
+        }
+
+        private class ObserverFunctionWrapper implements Observer<T> {
+            private final Action1<? super T> onNext;
+            private final Action1<Throwable> onError;
+            private final Action0 onCompleted;
+
+            public ObserverFunctionWrapper(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onCompleted) {
+                this.onNext = onNext;
+                this.onError = onError;
+                this.onCompleted = onCompleted;
+            }
+
+            @Override
+            public void onCompleted() {
+                if (onCompleted != null)
+                    onCompleted.call();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                handleError(e);
+                if (onError == null)
+                    throw new OnErrorNotImplementedException(e);
+                onError.call(e);
+            }
+
+            @Override
+            public void onNext(T data) {
+                if (onNext != null) {
+                    onNext.call(data);
+                }
+            }
+        }
+
+        /**
+         * Subscribe and ignore all events.
+         * 
+         * @return
+         */
+        public void subscribe() {
+            subscribe(new ObserverFunctionWrapper(null, null, null));
+        }
+
+        /**
+         * An {@link Observer} must call an Observable's {@code subscribe} method
+         * in order to receive items and notifications from the Observable.
+         * 
+         * @param onNext
+         * @return
+         */
+        public void subscribe(final Action1<? super T> onNext) {
+            if (onNext == null) {
+                throw new IllegalArgumentException("onNext can not be null");
+            }
+
+            subscribe(new ObserverFunctionWrapper(onNext, null, null));
+        }
+
+        /**
+         * An {@link Observer} must call an Observable's {@code subscribe} method in
+         * order to receive items and notifications from the Observable.
+         * 
+         * @param onNext
+         * @param onError
+         * @return
+         */
+        public void subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError) {
+            if (onNext == null) {
+                throw new IllegalArgumentException("onNext can not be null");
+            }
+            if (onError == null) {
+                throw new IllegalArgumentException("onError can not be null");
+            }
+
+            subscribe(new ObserverFunctionWrapper(onNext, onError, null));
+        }
+
+        /**
+         * An {@link Observer} must call an Observable's {@code subscribe} method in
+         * order to receive items and notifications from the Observable.
+         * 
+         * @param onNext
+         * @param onError
+         * @param onComplete
+         * @return
+         */
+        public void subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
+            if (onNext == null) {
+                throw new IllegalArgumentException("onNext can not be null");
+            }
+            if (onError == null) {
+                throw new IllegalArgumentException("onError can not be null");
+            }
+            if (onComplete == null) {
+                throw new IllegalArgumentException("onComplete can not be null");
+            }
+
+            subscribe(new ObserverFunctionWrapper(onNext, onError, onComplete));
+        }
+
+        public static <T> PartialSubscription<T> create(OnPartialSubscribeFunc<T> onPartialSubscribe, OnPartialUnsubscribeFunc onPartialUnsubscribe) {
+            return new PartialSubscription<T>(onPartialSubscribe, onPartialUnsubscribe);
+        }
     }
 
-    private final static RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();
+    public PartialSubscription<T> getSubscription() {
+        return onGetSubscription.onGetSubscription();
+    }
 
     /**
      * An {@link Observer} must call an Observable's {@code subscribe} method in
@@ -217,55 +499,13 @@ protected Observable(OnSubscribeFunc<T> onSubscribe) {
      * @throws IllegalArgumentException if the {@link Observer} provided as the
      *                                  argument to {@code subscribe()} is
      *                                  {@code null}
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe(Observer<? super T> observer) {
-        // allow the hook to intercept and/or decorate
-        OnSubscribeFunc<T> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);
-        // validate and proceed
-        if (observer == null) {
-            throw new IllegalArgumentException("observer can not be null");
-        }
-        if (onSubscribeFunction == null) {
-            throw new IllegalStateException("onSubscribe function can not be null.");
-            // the subscribe function can also be overridden but generally that's not the appropriate approach so I won't mention that in the exception
-        }
-        try {
-            /**
-             * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
-             */
-            if (isInternalImplementation(observer)) {
-                Subscription s = onSubscribeFunction.onSubscribe(observer);
-                if (s == null) {
-                    // this generally shouldn't be the case on a 'trusted' onSubscribe but in case it happens
-                    // we want to gracefully handle it the same as AtomicObservableSubscription does
-                    return hook.onSubscribeReturn(this, Subscriptions.empty());
-                } else {
-                    return hook.onSubscribeReturn(this, s);
-                }
-            } else {
-                SafeObservableSubscription subscription = new SafeObservableSubscription();
-                subscription.wrap(onSubscribeFunction.onSubscribe(new SafeObserver<T>(subscription, observer)));
-                return hook.onSubscribeReturn(this, subscription);
-            }
-        } catch (OnErrorNotImplementedException e) {
-            // special handling when onError is not implemented ... we just rethrow
-            throw e;
-        } catch (Throwable e) {
-            // if an unhandled error occurs executing the onSubscribe we will propagate it
-            try {
-                observer.onError(hook.onSubscribeError(this, e));
-            } catch (OnErrorNotImplementedException e2) {
-                // special handling when onError is not implemented ... we just rethrow
-                throw e2;
-            } catch (Throwable e2) {
-                // if this happens it means the onError itself failed (perhaps an invalid function implementation)
-                // so we are unable to propagate the error correctly and will just throw
-                RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
-                hook.onSubscribeError(this, r);
-                throw r;
-            }
-            return Subscriptions.empty();
-        }
+        PartialSubscription<T> partial = getSubscription();
+        partial.subscribe(observer);
+        return partial;
     }
 
     /**
@@ -299,86 +539,39 @@ public Subscription subscribe(Observer<? super T> observer) {
      *         finished sending them
      * @throws IllegalArgumentException if an argument to {@code subscribe()}
      *                                  is {@code null}
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe(Observer<? super T> observer, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(observer);
     }
 
-    /**
-     * Protects against errors being thrown from Observer implementations and
-     * ensures onNext/onError/onCompleted contract compliance.
-     * <p>
-     * See https://github.com/Netflix/RxJava/issues/216 for a discussion on
-     * "Guideline 6.4: Protect calls to user code from within an operator"
-     */
-    private Subscription protectivelyWrapAndSubscribe(Observer<? super T> o) {
-        SafeObservableSubscription subscription = new SafeObservableSubscription();
-        return subscription.wrap(subscribe(new SafeObserver<T>(subscription, o)));
-    }
-
     /**
      * Subscribe and ignore all events.
-     *  
-     * @return 
+     * 
+     * @return
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe() {
-        return protectivelyWrapAndSubscribe(new Observer<T>() {
-
-            @Override
-            public void onCompleted() {
-                // do nothing
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                handleError(e);
-                throw new OnErrorNotImplementedException(e);
-            }
-
-            @Override
-            public void onNext(T args) {
-                // do nothing
-            }
-
-        });
+        PartialSubscription<T> partial = getSubscription();
+        partial.subscribe();
+        return partial;
     }
-    
+
     /**
      * An {@link Observer} must call an Observable's {@code subscribe} method
      * in order to receive items and notifications from the Observable.
      * 
      * @param onNext
-     * @return 
+     * @return
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe(final Action1<? super T> onNext) {
-        if (onNext == null) {
-            throw new IllegalArgumentException("onNext can not be null");
-        }
-
-        /**
-         * Wrapping since raw functions provided by the user are being invoked.
-         * 
-         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
-         */
-        return protectivelyWrapAndSubscribe(new Observer<T>() {
-
-            @Override
-            public void onCompleted() {
-                // do nothing
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                handleError(e);
-                throw new OnErrorNotImplementedException(e);
-            }
-
-            @Override
-            public void onNext(T args) {
-                onNext.call(args);
-            }
-
-        });
+        PartialSubscription<T> partial = getSubscription();
+        partial.subscribe(onNext);
+        return partial;
     }
 
     /**
@@ -388,7 +581,9 @@ public void onNext(T args) {
      * @param onNext
      * @param scheduler
      * @return
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe(final Action1<? super T> onNext, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(onNext);
     }
@@ -400,39 +595,13 @@ public Subscription subscribe(final Action1<? super T> onNext, Scheduler schedul
      * @param onNext
      * @param onError
      * @return
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError) {
-        if (onNext == null) {
-            throw new IllegalArgumentException("onNext can not be null");
-        }
-        if (onError == null) {
-            throw new IllegalArgumentException("onError can not be null");
-        }
-
-        /**
-         * Wrapping since raw functions provided by the user are being invoked.
-         * 
-         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
-         */
-        return protectivelyWrapAndSubscribe(new Observer<T>() {
-
-            @Override
-            public void onCompleted() {
-                // do nothing
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                handleError(e);
-                onError.call(e);
-            }
-
-            @Override
-            public void onNext(T args) {
-                onNext.call(args);
-            }
-
-        });
+        PartialSubscription<T> partial = getSubscription();
+        partial.subscribe(onNext, onError);
+        return partial;
     }
 
     /**
@@ -443,7 +612,9 @@ public void onNext(T args) {
      * @param onError
      * @param scheduler
      * @return
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(onNext, onError);
     }
@@ -456,42 +627,13 @@ public Subscription subscribe(final Action1<? super T> onNext, final Action1<Thr
      * @param onError
      * @param onComplete
      * @return
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
-        if (onNext == null) {
-            throw new IllegalArgumentException("onNext can not be null");
-        }
-        if (onError == null) {
-            throw new IllegalArgumentException("onError can not be null");
-        }
-        if (onComplete == null) {
-            throw new IllegalArgumentException("onComplete can not be null");
-        }
-
-        /**
-         * Wrapping since raw functions provided by the user are being invoked.
-         * 
-         * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
-         */
-        return protectivelyWrapAndSubscribe(new Observer<T>() {
-
-            @Override
-            public void onCompleted() {
-                onComplete.call();
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                handleError(e);
-                onError.call(e);
-            }
-
-            @Override
-            public void onNext(T args) {
-                onNext.call(args);
-            }
-
-        });
+        PartialSubscription<T> partial = getSubscription();
+        partial.subscribe(onNext, onError, onComplete);
+        return partial;
     }
 
     /**
@@ -503,7 +645,9 @@ public void onNext(T args) {
      * @param onComplete
      * @param scheduler
      * @return
+     * @deprecated see {@link Observable#getSubscription()}
      */
+    @Deprecated
     public Subscription subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(onNext, onError, onComplete);
     }
@@ -512,29 +656,21 @@ public Subscription subscribe(final Action1<? super T> onNext, final Action1<Thr
      * Returns a {@link ConnectableObservable} that upon connection causes the
      * source Observable to push results into the specified subject.
      * 
-     * @param subject the {@link Subject} for the {@link ConnectableObservable}
-     *                to push source items into
-     * @param <R> result type
+     * @param subject
+     *            the {@link Subject} for the {@link ConnectableObservable} to push source items
+     *            into
+     * @param <R>
+     *            result type
      * @return a {@link ConnectableObservable} that upon connection causes the
-     *         source Observable to push results into the specified
-     *         {@link Subject}
-     * @see <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablepublish-and-observablemulticast">RxJava Wiki: Observable.publish() and Observable.multicast()</a>
+     *         source Observable to push results into the specified {@link Subject}
+     * @see <a
+     *      href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators#observablepublish-and-observablemulticast">RxJava
+     *      Wiki: Observable.publish() and Observable.multicast()</a>
      */
     public <R> ConnectableObservable<R> multicast(Subject<? super T, ? extends R> subject) {
         return OperationMulticast.multicast(this, subject);
     }
 
-    /**
-     * Allow the {@link RxJavaErrorHandler} to receive the exception from
-     * onError.
-     * 
-     * @param e
-     */
-    private void handleError(Throwable e) {
-        // onError should be rare so we'll only fetch when needed
-        RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
-    }
-
     /**
      * An Observable that never sends any information to an {@link Observer}.
      * 
@@ -544,13 +680,19 @@ private void handleError(Throwable e) {
      */
     private static class NeverObservable<T> extends Observable<T> {
         public NeverObservable() {
-            super(new OnSubscribeFunc<T>() {
-
+            super(new OnGetSubscriptionFunc<T>() {
                 @Override
-                public Subscription onSubscribe(Observer<? super T> t1) {
-                    return Subscriptions.empty();
+                public PartialSubscription<T> onGetSubscription() {
+                    return PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
+                        @Override
+                        public void onSubscribe(Observer<? super T> observer) {
+                        }
+                    }, new OnPartialUnsubscribeFunc() {
+                        @Override
+                        public void onUnsubscribe() {
+                        }
+                    });
                 }
-
             });
         }
     }
@@ -564,24 +706,22 @@ public Subscription onSubscribe(Observer<? super T> t1) {
     private static class ThrowObservable<T> extends Observable<T> {
 
         public ThrowObservable(final Throwable exception) {
-            super(new OnSubscribeFunc<T>() {
-
-                /**
-                 * Accepts an {@link Observer} and calls its
-                 * {@link Observer#onError onError} method.
-                 * 
-                 * @param observer an {@link Observer} of this Observable
-                 * @return a reference to the subscription
-                 */
+            super(new OnGetSubscriptionFunc<T>() {
                 @Override
-                public Subscription onSubscribe(Observer<? super T> observer) {
-                    observer.onError(exception);
-                    return Subscriptions.empty();
+                public PartialSubscription<T> onGetSubscription() {
+                    return PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
+                        @Override
+                        public void onSubscribe(Observer<? super T> observer) {
+                            observer.onError(exception);
+                        }
+                    }, new OnPartialUnsubscribeFunc() {
+                        @Override
+                        public void onUnsubscribe() {
+                        }
+                    });
                 }
-
             });
         }
-
     }
 
     /**
@@ -602,16 +742,40 @@ public Subscription onSubscribe(Observer<? super T> observer) {
      * See <a href="http://go.microsoft.com/fwlink/?LinkID=205219">Rx Design
      * Guidelines (PDF)</a> for detailed information.
      * 
-     * @param <T> the type of the items that this Observable emits
-     * @param func a function that accepts an {@code Observer<T>}, invokes its
-     *             {@code onNext}, {@code onError}, and {@code onCompleted}
-     *             methods as appropriate, and returns a {@link Subscription} to
-     *             allow the Observer to cancel the subscription
+     * @param <T>
+     *            the type of the items that this Observable emits
+     * @param func
+     *            a function that accepts an {@code Observer<T>}, invokes its {@code onNext},
+     *            {@code onError}, and {@code onCompleted} methods as appropriate, and returns a
+     *            {@link Subscription} to
+     *            allow the Observer to cancel the subscription
      * @return an Observable that, when an {@link Observer} subscribes to it,
      *         will execute the given function
-     * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#create">RxJava Wiki: create()</a>
+     * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#create">RxJava
+     *      Wiki: create()</a>
      */
-    public static <T> Observable<T> create(OnSubscribeFunc<T> func) {
+    @Deprecated
+    public static <T> Observable<T> create(final OnSubscribeFunc<T> func) {
+        return new Observable<T>(new OnGetSubscriptionFunc<T>() {
+            @Override
+            public PartialSubscription<T> onGetSubscription() {
+                final SafeObservableSubscription subscription = new SafeObservableSubscription();
+                return PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
+                    @Override
+                    public void onSubscribe(Observer<? super T> observer) {
+                        subscription.wrap(func.onSubscribe(observer));
+                    }
+                }, new OnPartialUnsubscribeFunc() {
+                    @Override
+                    public void onUnsubscribe() {
+                        subscription.unsubscribe();
+                    }
+                });
+            }
+        });
+    }
+
+    public static <T> Observable<T> create(OnGetSubscriptionFunc<T> func) {
         return new Observable<T>(func);
     }
 
diff --git a/rxjava-core/src/main/java/rx/Observer.java b/rxjava-core/src/main/java/rx/Observer.java
index 3d35066ba4..e8abd83425 100644
--- a/rxjava-core/src/main/java/rx/Observer.java
+++ b/rxjava-core/src/main/java/rx/Observer.java
@@ -53,5 +53,5 @@ public interface Observer<T> {
      * 
      * @param args
      */
-    public void onNext(T args);
+    public void onNext(T data);
 }
diff --git a/rxjava-core/src/main/java/rx/observables/ConnectableObservable.java b/rxjava-core/src/main/java/rx/observables/ConnectableObservable.java
index 1826905eaf..cb8d99d999 100644
--- a/rxjava-core/src/main/java/rx/observables/ConnectableObservable.java
+++ b/rxjava-core/src/main/java/rx/observables/ConnectableObservable.java
@@ -37,8 +37,8 @@
 
 public abstract class ConnectableObservable<T> extends Observable<T> {
 
-    protected ConnectableObservable(OnSubscribeFunc<T> onSubscribe) {
-        super(onSubscribe);
+    protected ConnectableObservable(OnGetSubscriptionFunc<T> onGetSubscription) {
+        super(onGetSubscription);
     }
 
     /**
diff --git a/rxjava-core/src/main/java/rx/observables/GroupedObservable.java b/rxjava-core/src/main/java/rx/observables/GroupedObservable.java
index f40e8c40af..02fb295eb2 100644
--- a/rxjava-core/src/main/java/rx/observables/GroupedObservable.java
+++ b/rxjava-core/src/main/java/rx/observables/GroupedObservable.java
@@ -31,8 +31,8 @@
 public class GroupedObservable<K, T> extends Observable<T> {
     private final K key;
 
-    public GroupedObservable(K key, OnSubscribeFunc<T> onSubscribe) {
-        super(onSubscribe);
+    public GroupedObservable(K key, OnGetSubscriptionFunc<T> onGetSubscription) {
+        super(onGetSubscription);
         this.key = key;
     }
 
diff --git a/rxjava-core/src/main/java/rx/operators/OperationCast.java b/rxjava-core/src/main/java/rx/operators/OperationCast.java
index dc54c204f6..84fd447678 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationCast.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationCast.java
@@ -16,7 +16,7 @@
 package rx.operators;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
+import rx.Observable.OnGetSubscriptionFunc;
 import rx.util.functions.Func1;
 
 /**
@@ -24,7 +24,7 @@
  */
 public class OperationCast {
 
-    public static <T, R> OnSubscribeFunc<R> cast(
+    public static <T, R> OnGetSubscriptionFunc<R> cast(
             Observable<? extends T> source, final Class<R> klass) {
         return OperationMap.map(source, new Func1<T, R>() {
             public R call(T t) {
diff --git a/rxjava-core/src/main/java/rx/operators/OperationGroupBy.java b/rxjava-core/src/main/java/rx/operators/OperationGroupBy.java
index a30b97523f..9813775000 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationGroupBy.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationGroupBy.java
@@ -162,21 +162,21 @@ private static class GroupedSubject<K, T> extends GroupedObservable<K, T> implem
 
         static <K, T> GroupedSubject<K, T> create(final K key, final GroupBy<K, T> parent) {
             final AtomicReference<Observer<? super T>> subscribedObserver = new AtomicReference<Observer<? super T>>(OperationGroupBy.<T> emptyObserver());
-            return new GroupedSubject<K, T>(key, new OnSubscribeFunc<T>() {
-
-                private final SafeObservableSubscription subscription = new SafeObservableSubscription();
+            return new GroupedSubject<K, T>(key, new OnGetSubscriptionFunc<T>() {
 
                 @Override
-                public Subscription onSubscribe(Observer<? super T> observer) {
-                    // register Observer
-                    subscribedObserver.set(observer);
-
-                    parent.subscribeKey(key);
-
-                    return subscription.wrap(new Subscription() {
+                public PartialSubscription<T> onGetSubscription() {
+                    return PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
                         @Override
-                        public void unsubscribe() {
-                            // we remove the Observer so we stop emitting further events (they will be ignored if parent continues to send)
+                        public void onSubscribe(Observer<? super T> observer) {
+                            subscribedObserver.set(observer);
+                            parent.subscribeKey(key);
+                        }
+                    }, new OnPartialUnsubscribeFunc() {
+                        @Override
+                        public void onUnsubscribe() {
+                            // we remove the Observer so we stop emitting further events (they will
+                            // be ignored if parent continues to send)
                             subscribedObserver.set(OperationGroupBy.<T> emptyObserver());
                             // now we need to notify the parent that we're unsubscribed
                             parent.unsubscribeKey(key);
@@ -188,8 +188,8 @@ public void unsubscribe() {
 
         private final AtomicReference<Observer<? super T>> subscribedObserver;
 
-        public GroupedSubject(K key, OnSubscribeFunc<T> onSubscribe, AtomicReference<Observer<? super T>> subscribedObserver) {
-            super(key, onSubscribe);
+        public GroupedSubject(K key, OnGetSubscriptionFunc<T> onGetSubscription, AtomicReference<Observer<? super T>> subscribedObserver) {
+            super(key, onGetSubscription);
             this.subscribedObserver = subscribedObserver;
         }
 
@@ -214,12 +214,12 @@ private static <T> Observer<T> emptyObserver() {
         return new Observer<T>() {
             @Override
             public void onCompleted() {
-                // do nothing            
+                // do nothing
             }
 
             @Override
             public void onError(Throwable e) {
-                // do nothing            
+                // do nothing
             }
 
             @Override
diff --git a/rxjava-core/src/main/java/rx/operators/OperationGroupByUntil.java b/rxjava-core/src/main/java/rx/operators/OperationGroupByUntil.java
index 63a271a688..864bc0d31c 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationGroupByUntil.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationGroupByUntil.java
@@ -21,7 +21,11 @@
 import java.util.Map;
 
 import rx.Observable;
+import rx.Observable.OnGetSubscriptionFunc;
+import rx.Observable.OnPartialSubscribeFunc;
+import rx.Observable.OnPartialUnsubscribeFunc;
 import rx.Observable.OnSubscribeFunc;
+import rx.Observable.PartialSubscription;
 import rx.Observer;
 import rx.Subscription;
 import rx.observables.GroupedObservable;
@@ -29,7 +33,6 @@
 import rx.subjects.Subject;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.SerialSubscription;
-import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func1;
 
 /**
@@ -197,11 +200,19 @@ public void onCompleted() {
             
         }
     }
-    protected static <T> OnSubscribeFunc<T> neverSubscribe() {
-        return new OnSubscribeFunc<T>() {
+    protected static <T> OnGetSubscriptionFunc<T> neverSubscribe() {
+        return new OnGetSubscriptionFunc<T>() {
             @Override
-            public Subscription onSubscribe(Observer<? super T> t1) {
-                return Subscriptions.empty();
+            public PartialSubscription<T> onGetSubscription() {
+                return PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
+                    @Override
+                    public void onSubscribe(Observer<? super T> observer) {
+                    }
+                }, new OnPartialUnsubscribeFunc() {
+                    @Override
+                    public void onUnsubscribe() {
+                    }
+                });
             }
         };
     }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationMap.java b/rxjava-core/src/main/java/rx/operators/OperationMap.java
index d78b5dc6ef..69327b9819 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationMap.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationMap.java
@@ -15,8 +15,15 @@
  */
 package rx.operators;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
 import rx.Observable;
+import rx.Observable.OnGetSubscriptionFunc;
+import rx.Observable.OnPartialSubscribeFunc;
+import rx.Observable.OnPartialUnsubscribeFunc;
 import rx.Observable.OnSubscribeFunc;
+import rx.Observable.PartialSubscription;
 import rx.Observer;
 import rx.Subscription;
 import rx.util.functions.Func1;
@@ -42,9 +49,10 @@ public final class OperationMap {
      *            the type of the input sequence.
      * @param <R>
      *            the type of the output sequence.
-     * @return a sequence that is the result of applying the transformation function to each item in the input sequence.
+     * @return a sequence that is the result of applying the transformation function to each item in
+     *         the input sequence.
      */
-    public static <T, R> OnSubscribeFunc<R> map(final Observable<? extends T> sequence, final Func1<? super T, ? extends R> func) {
+    public static <T, R> OnGetSubscriptionFunc<R> map(final Observable<? extends T> sequence, final Func1<? super T, ? extends R> func) {
         return mapWithIndex(sequence, new Func2<T, Integer, R>() {
             @Override
             public R call(T value, @SuppressWarnings("unused") Integer unused) {
@@ -60,26 +68,30 @@ public R call(T value, @SuppressWarnings("unused") Integer unused) {
      * @param sequence
      *            the input sequence.
      * @param func
-     *            a function to apply to each item in the sequence. The function gets the index of the emitted item
+     *            a function to apply to each item in the sequence. The function gets the index of
+     *            the emitted item
      *            as additional parameter.
      * @param <T>
      *            the type of the input sequence.
      * @param <R>
      *            the type of the output sequence.
-     * @return a sequence that is the result of applying the transformation function to each item in the input sequence.
+     * @return a sequence that is the result of applying the transformation function to each item in
+     *         the input sequence.
      */
-    public static <T, R> OnSubscribeFunc<R> mapWithIndex(final Observable<? extends T> sequence, final Func2<? super T, Integer, ? extends R> func) {
-        return new OnSubscribeFunc<R>() {
+    public static <T, R> OnGetSubscriptionFunc<R> mapWithIndex(final Observable<? extends T> sequence, final Func2<? super T, Integer, ? extends R> func) {
+        return new OnGetSubscriptionFunc<R>() {
             @Override
-            public Subscription onSubscribe(Observer<? super R> observer) {
-                return new MapObservable<T, R>(sequence, func).onSubscribe(observer);
+            public PartialSubscription<R> onGetSubscription() {
+                return new MapObservable<T, R>(sequence, func).onGetSubscription();
             }
         };
     }
 
     /**
-     * Accepts a sequence of observable sequences and a transformation function. Returns a flattened sequence that is the result of
-     * applying the transformation function to each item in the sequence of each observable sequence.
+     * Accepts a sequence of observable sequences and a transformation function. Returns a flattened
+     * sequence that is the result of
+     * applying the transformation function to each item in the sequence of each observable
+     * sequence.
      * <p>
      * The closure should return an Observable which will then be merged.
      * 
@@ -91,21 +103,23 @@ public Subscription onSubscribe(Observer<? super R> observer) {
      *            the type of the input sequence.
      * @param <R>
      *            the type of the output sequence.
-     * @return a sequence that is the result of applying the transformation function to each item in the input sequence.
+     * @return a sequence that is the result of applying the transformation function to each item in
+     *         the input sequence.
      */
     public static <T, R> OnSubscribeFunc<R> mapMany(Observable<? extends T> sequence, Func1<? super T, ? extends Observable<? extends R>> func) {
         return OperationMerge.merge(Observable.create(map(sequence, func)));
     }
 
     /**
-     * An observable sequence that is the result of applying a transformation to each item in an input sequence.
+     * An observable sequence that is the result of applying a transformation to each item in an
+     * input sequence.
      * 
      * @param <T>
      *            the type of the input sequence.
      * @param <R>
      *            the type of the output sequence.
      */
-    private static class MapObservable<T, R> implements OnSubscribeFunc<R> {
+    private static class MapObservable<T, R> implements OnGetSubscriptionFunc<R> {
         public MapObservable(Observable<? extends T> sequence, Func2<? super T, Integer, ? extends R> func) {
             this.sequence = sequence;
             this.func = func;
@@ -116,25 +130,38 @@ public MapObservable(Observable<? extends T> sequence, Func2<? super T, Integer,
         private int index;
 
         @Override
-        public Subscription onSubscribe(final Observer<? super R> observer) {
-            final SafeObservableSubscription subscription = new SafeObservableSubscription();
-            return subscription.wrap(sequence.subscribe(new SafeObserver<T>(subscription, new Observer<T>() {
-                @Override
-                public void onNext(T value) {
-                    observer.onNext(func.call(value, index));
-                    index++;
-                }
+        public PartialSubscription<R> onGetSubscription() {
+            final AtomicReference<PartialSubscription<? extends T>> subscription = new AtomicReference<PartialSubscription<? extends T>>();
+            return PartialSubscription.create(new OnPartialSubscribeFunc<R>() {
 
                 @Override
-                public void onError(Throwable ex) {
-                    observer.onError(ex);
+                public void onSubscribe(final Observer<? super R> observer) {
+                    subscription.set(sequence.getSubscription());
+                    subscription.get().subscribe(new Observer<T>() {
+                        @Override
+                        public void onNext(T value) {
+                            observer.onNext(func.call(value, index));
+                            index++;
+                        }
+
+                        @Override
+                        public void onError(Throwable ex) {
+                            observer.onError(ex);
+                        }
+
+                        @Override
+                        public void onCompleted() {
+                            observer.onCompleted();
+                        }
+                    });
                 }
+            }, new OnPartialUnsubscribeFunc() {
 
                 @Override
-                public void onCompleted() {
-                    observer.onCompleted();
+                public void onUnsubscribe() {
+                    subscription.get().unsubscribe();
                 }
-            })));
+            });
         }
     }
 }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationMulticast.java b/rxjava-core/src/main/java/rx/operators/OperationMulticast.java
index b634b2dbac..66bfabf653 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationMulticast.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationMulticast.java
@@ -35,10 +35,10 @@ private static class MulticastConnectableObservable<T, R> extends ConnectableObs
         private Subscription subscription;
 
         public MulticastConnectableObservable(Observable<? extends T> source, final Subject<? super T, ? extends R> subject) {
-            super(new OnSubscribeFunc<R>() {
+            super(new OnGetSubscriptionFunc<R>() {
                 @Override
-                public Subscription onSubscribe(Observer<? super R> observer) {
-                    return subject.subscribe(observer);
+                public PartialSubscription<R> onGetSubscription() {
+                    return (PartialSubscription<R>) subject.getSubscription();
                 }
             });
             this.source = source;
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTake.java b/rxjava-core/src/main/java/rx/operators/OperationTake.java
index 877ba4d5f3..aa16843440 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationTake.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationTake.java
@@ -15,13 +15,16 @@
  */
 package rx.operators;
 
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
+import rx.Observable.OnGetSubscriptionFunc;
+import rx.Observable.OnPartialSubscribeFunc;
+import rx.Observable.OnPartialUnsubscribeFunc;
+import rx.Observable.PartialSubscription;
 import rx.Observer;
-import rx.Subscription;
-import rx.subscriptions.Subscriptions;
 
 /**
  * Returns an Observable that emits the first <code>num</code> items emitted by the source
@@ -29,10 +32,10 @@
  * <p>
  * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/take.png">
  * <p>
- * You can choose to pay attention only to the first <code>num</code> items emitted by an
- * Observable by using the take operation. This operation returns an Observable that will invoke a
- * subscribing Observer's <code>onNext</code> function a maximum of <code>num</code> times before
- * invoking <code>onCompleted</code>.
+ * You can choose to pay attention only to the first <code>num</code> items emitted by an Observable
+ * by using the take operation. This operation returns an Observable that will invoke a subscribing
+ * Observer's <code>onNext</code> function a maximum of <code>num</code> times before invoking
+ * <code>onCompleted</code>.
  */
 public final class OperationTake {
 
@@ -41,18 +44,13 @@ public final class OperationTake {
      * 
      * @param items
      * @param num
-     * @return the specified number of contiguous values from the start of the given observable sequence
+     * @return the specified number of contiguous values from the start of the given observable
+     *         sequence
      */
-    public static <T> OnSubscribeFunc<T> take(final Observable<? extends T> items, final int num) {
-        // wrap in a Func so that if a chain is built up, then asynchronously subscribed to twice we will have 2 instances of Take<T> rather than 1 handing both, which is not thread-safe.
-        return new OnSubscribeFunc<T>() {
-
-            @Override
-            public Subscription onSubscribe(Observer<? super T> observer) {
-                return new Take<T>(items, num).onSubscribe(observer);
-            }
-
-        };
+    public static <T> OnGetSubscriptionFunc<T> take(final Observable<? extends T> items, final int num) {
+        // wrap in a Func so that if a chain is built up, then asynchronously subscribed to twice we
+        // will have 2 instances of Take<T> rather than 1 handing both, which is not thread-safe.
+        return new Take<T>(items, num);
     }
 
     /**
@@ -66,10 +64,10 @@ public Subscription onSubscribe(Observer<? super T> observer) {
      * 
      * @param <T>
      */
-    private static class Take<T> implements OnSubscribeFunc<T> {
+    private static class Take<T> implements OnGetSubscriptionFunc<T> {
         private final Observable<? extends T> items;
         private final int num;
-        private final SafeObservableSubscription subscription = new SafeObservableSubscription();
+        private final AtomicReference<PartialSubscription<? extends T>> subscription = new AtomicReference<Observable.PartialSubscription<? extends T>>();
 
         private Take(Observable<? extends T> items, int num) {
             this.items = items;
@@ -77,30 +75,26 @@ private Take(Observable<? extends T> items, int num) {
         }
 
         @Override
-        public Subscription onSubscribe(Observer<? super T> observer) {
-            if (num < 1) {
-                items.subscribe(new Observer<T>()
-                {
-                    @Override
-                    public void onCompleted()
-                    {
-                    }
-
-                    @Override
-                    public void onError(Throwable e)
-                    {
+        public PartialSubscription<T> onGetSubscription() {
+            return PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
+                @Override
+                public void onSubscribe(Observer<? super T> observer) {
+                    PartialSubscription<? extends T> partialSubscription = items.getSubscription();
+                    subscription.set(partialSubscription);
+                    if (num < 1) {
+                        // signal that we don't really want any values by unsubscribing before we've
+                        // even subscribed.
+                        subscription.get().unsubscribe();
                     }
 
-                    @Override
-                    public void onNext(T args)
-                    {
-                    }
-                }).unsubscribe();
-                observer.onCompleted();
-                return Subscriptions.empty();
-            }
-
-            return subscription.wrap(items.subscribe(new ItemObserver(observer)));
+                    partialSubscription.subscribe(new ItemObserver(observer));
+                }
+            }, new OnPartialUnsubscribeFunc() {
+                @Override
+                public void onUnsubscribe() {
+                    subscription.get().unsubscribe();
+                }
+            });
         }
 
         private class ItemObserver implements Observer<T> {
@@ -145,7 +139,7 @@ public void onNext(T args) {
                     } catch (Throwable ex) {
                         hasEmitedError = true;
                         observer.onError(ex);
-                        subscription.unsubscribe();
+                        subscription.get().unsubscribe();
                         return;
                     }
                     if (count == num) {
@@ -153,12 +147,12 @@ public void onNext(T args) {
                     }
                 }
                 if (count >= num) {
-                    // this will work if the sequence is asynchronous, it will have no effect on a synchronous observable
-                    subscription.unsubscribe();
+                    // OLD: this will work if the sequence is asynchronous, it will have no effect on a synchronous observable
+                    // NEW: with the two phase getSubscription and subscribe event synchronous observables can be interrupted.
+                    subscription.get().unsubscribe();
                 }
             }
 
         }
-
     }
 }
diff --git a/rxjava-core/src/main/java/rx/operators/OperationTimestamp.java b/rxjava-core/src/main/java/rx/operators/OperationTimestamp.java
index a1044ba748..be9ea3ffb3 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationTimestamp.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationTimestamp.java
@@ -16,8 +16,8 @@
 package rx.operators;
 
 import rx.Observable;
-import rx.Observable.OnSubscribeFunc;
 import rx.Scheduler;
+import rx.Observable.OnGetSubscriptionFunc;
 import rx.util.Timestamped;
 import rx.util.functions.Func1;
 
@@ -37,7 +37,7 @@ public final class OperationTimestamp {
      *            the type of the input sequence.
      * @return a sequence of timestamped values created by adding timestamps to each item in the input sequence.
      */
-    public static <T> OnSubscribeFunc<Timestamped<T>> timestamp(Observable<? extends T> sequence) {
+    public static <T> OnGetSubscriptionFunc<Timestamped<T>> timestamp(Observable<? extends T> sequence) {
         return OperationMap.map(sequence, new Func1<T, Timestamped<T>>() {
             @Override
             public Timestamped<T> call(T value) {
@@ -48,7 +48,7 @@ public Timestamped<T> call(T value) {
     /**
      * Timestamp the source elements based on the timing provided by the scheduler.
      */
-    public static <T> OnSubscribeFunc<Timestamped<T>> timestamp(Observable<? extends T> source, final Scheduler scheduler) {
+    public static <T> OnGetSubscriptionFunc<Timestamped<T>> timestamp(Observable<? extends T> source, final Scheduler scheduler) {
         return OperationMap.map(source, new Func1<T, Timestamped<T>>() {
             @Override
             public Timestamped<T> call(T value) {
diff --git a/rxjava-core/src/main/java/rx/operators/OperationToObservableIterable.java b/rxjava-core/src/main/java/rx/operators/OperationToObservableIterable.java
index a8a970bdd8..77df06cfda 100644
--- a/rxjava-core/src/main/java/rx/operators/OperationToObservableIterable.java
+++ b/rxjava-core/src/main/java/rx/operators/OperationToObservableIterable.java
@@ -15,7 +15,13 @@
  */
 package rx.operators;
 
-import rx.Observable.OnSubscribeFunc;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import rx.Observable.OnGetSubscriptionFunc;
+import rx.Observable.OnPartialSubscribeFunc;
+import rx.Observable.OnPartialUnsubscribeFunc;
+import rx.Observable.PartialSubscription;
 import rx.Observer;
 import rx.Subscription;
 import rx.subscriptions.Subscriptions;
@@ -30,24 +36,38 @@
  */
 public final class OperationToObservableIterable<T> {
 
-    public static <T> OnSubscribeFunc<T> toObservableIterable(Iterable<? extends T> list) {
+    public static <T> OnGetSubscriptionFunc<T> toObservableIterable(Iterable<? extends T> list) {
         return new ToObservableIterable<T>(list);
     }
 
-    private static class ToObservableIterable<T> implements OnSubscribeFunc<T> {
+    private static class ToObservableIterable<T> implements OnGetSubscriptionFunc<T> {
         public ToObservableIterable(Iterable<? extends T> list) {
             this.iterable = list;
         }
 
         public Iterable<? extends T> iterable;
 
-        public Subscription onSubscribe(Observer<? super T> observer) {
-            for (T item : iterable) {
-                observer.onNext(item);
-            }
-            observer.onCompleted();
-
-            return Subscriptions.empty();
+        @Override
+        public PartialSubscription<T> onGetSubscription() {
+            final AtomicReference<PartialSubscription<T>> subscription = new AtomicReference<PartialSubscription<T>>();
+            
+            subscription.set(PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
+                @Override
+                public void onSubscribe(Observer<? super T> observer) {
+                    for (T item : iterable) {
+                        if (!subscription.get().isUnsubscribed())
+                            observer.onNext(item);
+                    }
+                    if (!subscription.get().isUnsubscribed())
+                        observer.onCompleted();
+                }
+            }, new OnPartialUnsubscribeFunc() {
+                @Override
+                public void onUnsubscribe() {
+                }
+            }));
+            
+            return subscription.get();
         }
     }
 }
diff --git a/rxjava-core/src/main/java/rx/operators/SafeObserver.java b/rxjava-core/src/main/java/rx/operators/SafeObserver.java
index 3e6508dac9..99f16f7319 100644
--- a/rxjava-core/src/main/java/rx/operators/SafeObserver.java
+++ b/rxjava-core/src/main/java/rx/operators/SafeObserver.java
@@ -19,6 +19,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import rx.Observer;
+import rx.Subscription;
 import rx.plugins.RxJavaPlugins;
 import rx.util.CompositeException;
 import rx.util.OnErrorNotImplementedException;
@@ -59,9 +60,9 @@ public class SafeObserver<T> implements Observer<T> {
 
     private final Observer<? super T> actual;
     private final AtomicBoolean isFinished = new AtomicBoolean(false);
-    private final SafeObservableSubscription subscription;
+    private final Subscription subscription;
 
-    public SafeObserver(SafeObservableSubscription subscription, Observer<? super T> actual) {
+    public SafeObserver(Subscription subscription, Observer<? super T> actual) {
         this.subscription = subscription;
         this.actual = actual;
     }
diff --git a/rxjava-core/src/main/java/rx/subjects/AbstractSubject.java b/rxjava-core/src/main/java/rx/subjects/AbstractSubject.java
index 30db331ac2..c59c034532 100644
--- a/rxjava-core/src/main/java/rx/subjects/AbstractSubject.java
+++ b/rxjava-core/src/main/java/rx/subjects/AbstractSubject.java
@@ -24,70 +24,62 @@
 
 import rx.Notification;
 import rx.Observer;
-import rx.Subscription;
-import rx.operators.SafeObservableSubscription;
-import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action2;
 
 public abstract class AbstractSubject<T> extends Subject<T, T> {
 
-    protected AbstractSubject(rx.Observable.OnSubscribeFunc<T> onSubscribe) {
-        super(onSubscribe);
+    protected AbstractSubject(OnGetSubscriptionFunc<T> onGetSubscription) {
+        super(onGetSubscription);
     }
 
     protected static class SubjectState<T> {
-        protected final ConcurrentHashMap<Subscription, Observer<? super T>> observers = new ConcurrentHashMap<Subscription, Observer<? super T>>();
+        protected final ConcurrentHashMap<Object, Observer<? super T>> observers = new ConcurrentHashMap<Object, Observer<? super T>>();
         protected final AtomicReference<Notification<T>> currentValue = new AtomicReference<Notification<T>>();
         protected final AtomicBoolean completed = new AtomicBoolean();
         protected final ReentrantLock SUBSCRIPTION_LOCK = new ReentrantLock();
     }
 
-    protected static <T> OnSubscribeFunc<T> getOnSubscribeFunc(final SubjectState<T> state, final Action2<SubjectState<T>, Observer<? super T>> onEach) {
-        return new OnSubscribeFunc<T>() {
+    protected static <T> OnGetSubscriptionFunc<T> getOnGetSubscriptionFunc(final SubjectState<T> state, final Action2<SubjectState<T>, Observer<? super T>> onEach) {
+        return new OnGetSubscriptionFunc<T>() {
             @Override
-            public Subscription onSubscribe(Observer<? super T> observer) {
-                /*
-                 * Subscription needs to be synchronized with terminal states to ensure
-                 * race conditions are handled. When subscribing we must make sure
-                 * onComplete/onError is correctly emitted to all observers, even if it
-                 * comes in while the onComplete/onError is being propagated.
-                 */
-                state.SUBSCRIPTION_LOCK.lock();
-                try {
-                    if (state.completed.get()) {
-                        emitNotification(state.currentValue.get(), observer);
-                        if (onEach != null) {
-                            onEach.call(state, observer);
-                        }
-                        return Subscriptions.empty();
-                    } else {
-                        // the subject is not completed so we subscribe
-                        final SafeObservableSubscription subscription = new SafeObservableSubscription();
-
-                        subscription.wrap(new Subscription() {
-                            @Override
-                            public void unsubscribe() {
-                                // on unsubscribe remove it from the map of outbound observers to notify
-                                state.observers.remove(subscription);
+            public PartialSubscription<T> onGetSubscription() {
+                final Object marker = new Object();
+                return PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
+                    @Override
+                    public void onSubscribe(Observer<? super T> observer) {
+                        /*
+                         * Subscription needs to be synchronized with terminal states to ensure
+                         * race conditions are handled. When subscribing we must make sure
+                         * onComplete/onError is correctly emitted to all observers, even if it
+                         * comes in while the onComplete/onError is being propagated.
+                         */
+                        state.SUBSCRIPTION_LOCK.lock();
+                        try {
+                            if (state.completed.get()) {
+                                emitNotification(state.currentValue.get(), observer);
+                                if (onEach != null) {
+                                    onEach.call(state, observer);
+                                }
+                            } else {
+                                // on subscribe add it to the map of outbound observers to notify
+                                state.observers.put(marker, observer);
+
+                                // invoke onSubscribe logic
+                                if (onEach != null) {
+                                    onEach.call(state, observer);
+                                }
                             }
-                        });
-
-                        // on subscribe add it to the map of outbound observers to notify
-                        state.observers.put(subscription, observer);
-
-                        // invoke onSubscribe logic
-                        if (onEach != null) {
-                            onEach.call(state, observer);
+                        } finally {
+                            state.SUBSCRIPTION_LOCK.unlock();
                         }
-
-                        return subscription;
                     }
-                } finally {
-                    state.SUBSCRIPTION_LOCK.unlock();
-                }
-
+                }, new OnPartialUnsubscribeFunc() {
+                    @Override
+                    public void onUnsubscribe() {
+                        state.observers.remove(marker);
+                    }
+                });
             }
-
         };
     }
 
@@ -110,7 +102,7 @@ protected static <T> void emitNotification(Notification<T> value, Observer<? sup
      * @param state
      */
     protected static <T> void emitNotification(final SubjectState<T> state, final Action2<SubjectState<T>, Observer<? super T>> onEach) {
-        for (Subscription s : snapshotOfObservers(state)) {
+        for (Object s : snapshotOfObservers(state)) {
             Observer<? super T> o = state.observers.get(s);
             // emit notifications to this observer
             emitNotification(state.currentValue.get(), o);
@@ -133,7 +125,7 @@ protected void emitNotificationAndTerminate(final SubjectState<T> state, final A
         state.SUBSCRIPTION_LOCK.lock();
         try {
             if (state.completed.compareAndSet(false, true)) {
-                for (Subscription s : snapshotOfObservers(state)) {
+                for (Object s : snapshotOfObservers(state)) {
                     Observer<? super T> o = state.observers.get(s);
                     // emit notifications to this observer
                     emitNotification(state.currentValue.get(), o);
@@ -161,7 +153,7 @@ protected void emitNotificationAndTerminate(final SubjectState<T> state, final A
      * 
      * @return List<Observer<T>>
      */
-    private static <T> Collection<Subscription> snapshotOfObservers(final SubjectState<T> state) {
-        return new ArrayList<Subscription>(state.observers.keySet());
+    private static <T> Collection<Object> snapshotOfObservers(final SubjectState<T> state) {
+        return new ArrayList<Object>(state.observers.keySet());
     }
 }
diff --git a/rxjava-core/src/main/java/rx/subjects/AsyncSubject.java b/rxjava-core/src/main/java/rx/subjects/AsyncSubject.java
index a9c18be805..dd776ea18f 100644
--- a/rxjava-core/src/main/java/rx/subjects/AsyncSubject.java
+++ b/rxjava-core/src/main/java/rx/subjects/AsyncSubject.java
@@ -17,6 +17,7 @@
 
 import rx.Notification;
 import rx.Observer;
+import rx.Observable.OnGetSubscriptionFunc;
 import rx.util.functions.Action2;
 
 /**
@@ -57,7 +58,7 @@ public class AsyncSubject<T> extends AbstractSubject<T> {
      */
     public static <T> AsyncSubject<T> create() {
         final SubjectState<T> state = new SubjectState<T>();
-        OnSubscribeFunc<T> onSubscribe = getOnSubscribeFunc(state, new Action2<SubjectState<T>, Observer<? super T>>() {
+        OnGetSubscriptionFunc<T> onSubscribe = getOnGetSubscriptionFunc(state, new Action2<SubjectState<T>, Observer<? super T>>() {
 
             @Override
             public void call(SubjectState<T> state, Observer<? super T> o) {
@@ -76,8 +77,8 @@ public void call(SubjectState<T> state, Observer<? super T> o) {
 
     private final SubjectState<T> state;
 
-    protected AsyncSubject(OnSubscribeFunc<T> onSubscribe, SubjectState<T> state) {
-        super(onSubscribe);
+    protected AsyncSubject(OnGetSubscriptionFunc<T> onGetSubscription, SubjectState<T> state) {
+        super(onGetSubscription);
         this.state = state;
     }
 
diff --git a/rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java b/rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java
index 053a51f0cf..720ebb2dc7 100644
--- a/rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java
+++ b/rxjava-core/src/main/java/rx/subjects/BehaviorSubject.java
@@ -17,6 +17,7 @@
 
 import rx.Notification;
 import rx.Observer;
+import rx.Observable.OnGetSubscriptionFunc;
 import rx.util.functions.Action2;
 
 /**
@@ -72,7 +73,7 @@ public static <T> BehaviorSubject<T> create(T defaultValue) {
         final SubjectState<T> state = new SubjectState<T>();
         // set a default value so subscriptions will immediately receive this until a new notification is received
         state.currentValue.set(new Notification<T>(defaultValue));
-        OnSubscribeFunc<T> onSubscribe = getOnSubscribeFunc(state, new Action2<SubjectState<T>, Observer<? super T>>() {
+        OnGetSubscriptionFunc<T> onSubscribe = getOnGetSubscriptionFunc(state, new Action2<SubjectState<T>, Observer<? super T>>() {
 
             @Override
             public void call(SubjectState<T> state, Observer<? super T> o) {
@@ -88,8 +89,8 @@ public void call(SubjectState<T> state, Observer<? super T> o) {
 
     private final SubjectState<T> state;
 
-    protected BehaviorSubject(OnSubscribeFunc<T> onSubscribe, SubjectState<T> state) {
-        super(onSubscribe);
+    protected BehaviorSubject(OnGetSubscriptionFunc<T> onGetSubscription, SubjectState<T> state) {
+        super(onGetSubscription);
         this.state = state;
     }
 
diff --git a/rxjava-core/src/main/java/rx/subjects/PublishSubject.java b/rxjava-core/src/main/java/rx/subjects/PublishSubject.java
index 8cf2d75747..16c7d8ee66 100644
--- a/rxjava-core/src/main/java/rx/subjects/PublishSubject.java
+++ b/rxjava-core/src/main/java/rx/subjects/PublishSubject.java
@@ -17,6 +17,7 @@
 
 import rx.Notification;
 import rx.Observer;
+import rx.Observable.OnGetSubscriptionFunc;
 
 /**
  * Subject that, once and {@link Observer} has subscribed, publishes all subsequent events to the subscriber.
@@ -44,14 +45,14 @@
 public class PublishSubject<T> extends AbstractSubject<T> {
     public static <T> PublishSubject<T> create() {
         final SubjectState<T> state = new SubjectState<T>();
-        OnSubscribeFunc<T> onSubscribe = getOnSubscribeFunc(state, null);
-        return new PublishSubject<T>(onSubscribe, state);
+        OnGetSubscriptionFunc<T> onGetSubscription = getOnGetSubscriptionFunc(state, null);
+        return new PublishSubject<T>(onGetSubscription, state);
     }
 
     private final SubjectState<T> state;
 
-    protected PublishSubject(OnSubscribeFunc<T> onSubscribe, SubjectState<T> state) {
-        super(onSubscribe);
+    protected PublishSubject(OnGetSubscriptionFunc<T> onGetSubscription, SubjectState<T> state) {
+        super(onGetSubscription);
         this.state = state;
     }
 
diff --git a/rxjava-core/src/main/java/rx/subjects/ReplaySubject.java b/rxjava-core/src/main/java/rx/subjects/ReplaySubject.java
index 13d0cc3b57..358fd4003f 100644
--- a/rxjava-core/src/main/java/rx/subjects/ReplaySubject.java
+++ b/rxjava-core/src/main/java/rx/subjects/ReplaySubject.java
@@ -16,15 +16,12 @@
 package rx.subjects;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import rx.Notification;
 import rx.Observer;
-import rx.Subscription;
-import rx.subscriptions.Subscriptions;
-import rx.util.functions.Func1;
 
 /**
  * Subject that retains all events and will replay them to an {@link Observer} that subscribes.
@@ -33,140 +30,95 @@
  * <p>
  * Example usage:
  * <p>
- * <pre> {@code
-
- * eplaySubject<Object> subject = ReplaySubject.create();
-  subject.onNext("one");
-  subject.onNext("two");
-  subject.onNext("three");
-  subject.onCompleted();
-
-  // both of the following will get the onNext/onCompleted calls from above
-  subject.subscribe(observer1);
-  subject.subscribe(observer2);
-
-  } </pre>
+ * 
+ * <pre>
+ * {
+ *     &#064;code
+ *     eplaySubject&lt;Object&gt; subject = ReplaySubject.create();
+ *     subject.onNext(&quot;one&quot;);
+ *     subject.onNext(&quot;two&quot;);
+ *     subject.onNext(&quot;three&quot;);
+ *     subject.onCompleted();
+ * 
+ *     // both of the following will get the onNext/onCompleted calls from above
+ *     subject.subscribe(observer1);
+ *     subject.subscribe(observer2);
+ * 
+ * }
+ * </pre>
  * 
  * @param <T>
  */
 public final class ReplaySubject<T> extends Subject<T, T>
 {
-
-    private boolean isDone = false;
-    private Throwable exception = null;
-    private final Map<Subscription, Observer<? super T>> subscriptions = new HashMap<Subscription, Observer<? super T>>();
-    private final List<T> history = Collections.synchronizedList(new ArrayList<T>());
+    private final Map<Object, Observer<? super T>> subscriptions = new HashMap<Object, Observer<? super T>>();
+    private final List<Notification<T>> history = new ArrayList<Notification<T>>();
 
     public static <T> ReplaySubject<T> create() {
-        return new ReplaySubject<T>(new DelegateSubscriptionFunc<T>());
+        return new ReplaySubject<T>(new ReplayOnGetSubscription<T>());
     }
 
-    private ReplaySubject(DelegateSubscriptionFunc<T> onSubscribe) {
-        super(onSubscribe);
-        onSubscribe.wrap(new SubscriptionFunc());
+    private ReplaySubject(ReplayOnGetSubscription<T> onGetSubscription) {
+        super(onGetSubscription);
+        onGetSubscription.history = history;
+        onGetSubscription.subscriptions = subscriptions;
     }
 
-    private static final class DelegateSubscriptionFunc<T> implements OnSubscribeFunc<T>
-    {
-        private Func1<? super Observer<? super T>, ? extends Subscription> delegate = null;
-
-        public void wrap(Func1<? super Observer<? super T>, ? extends Subscription> delegate)
-        {
-            if (this.delegate != null) {
-                throw new UnsupportedOperationException("delegate already set");
-            }
-            this.delegate = delegate;
-        }
+    private static final class ReplayOnGetSubscription<T> implements OnGetSubscriptionFunc<T> {
+        private Map<Object, Observer<? super T>> subscriptions;
+        private List<Notification<T>> history;
 
         @Override
-        public Subscription onSubscribe(Observer<? super T> observer)
-        {
-            return delegate.call(observer);
-        }
-    }
-
-    private class SubscriptionFunc implements Func1<Observer<? super T>, Subscription>
-    {
-        @Override
-        public Subscription call(Observer<? super T> observer) {
-            int item = 0;
-            Subscription subscription;
-
-            for (;;) {
-                while (item < history.size()) {
-                    observer.onNext(history.get(item++));
-                }
-
-                synchronized (subscriptions) {
-                    if (item < history.size()) {
-                        continue;
-                    }
-
-                    if (exception != null) {
-                        observer.onError(exception);
-                        return Subscriptions.empty();
+        public PartialSubscription<T> onGetSubscription() {
+            final Object marker = new Object();
+            return PartialSubscription.create(new OnPartialSubscribeFunc<T>() {
+                @Override
+                public void onSubscribe(Observer<? super T> observer) {
+                    int item = 0;
+
+                    for (;;) {
+                        while (item < history.size()) {
+                            history.get(item++).accept(observer);
+                        }
+
+                        synchronized (subscriptions) {
+                            if (item < history.size()) {
+                                continue;
+                            }
+                            subscriptions.put(marker, observer);
+                            break;
+                        }
                     }
-                    if (isDone) {
-                        observer.onCompleted();
-                        return Subscriptions.empty();
-                    }
-
-                    subscription = new RepeatSubjectSubscription();
-                    subscriptions.put(subscription, observer);
-                    break;
                 }
-            }
-
-            return subscription;
+            }, new OnPartialUnsubscribeFunc() {
+                @Override
+                public void onUnsubscribe() {
+                    subscriptions.remove(marker);
+                }
+            });
         }
     }
 
-    private class RepeatSubjectSubscription implements Subscription
-    {
-        @Override
-        public void unsubscribe()
-        {
-            synchronized (subscriptions) {
-                subscriptions.remove(this);
-            }
-        }
+    @Override
+    public void onCompleted() {
+        propgate(new Notification<T>());
     }
 
     @Override
-    public void onCompleted()
-    {
-        synchronized (subscriptions) {
-            isDone = true;
-            for (Observer<? super T> observer : new ArrayList<Observer<? super T>>(subscriptions.values())) {
-                observer.onCompleted();
-            }
-            subscriptions.clear();
-        }
+    public void onError(Throwable e) {
+        propgate(new Notification<T>(e));
     }
 
     @Override
-    public void onError(Throwable e)
-    {
-        synchronized (subscriptions) {
-            if (isDone) {
-                return;
-            }
-            isDone = true;
-            exception = e;
-            for (Observer<? super T> observer : new ArrayList<Observer<? super T>>(subscriptions.values())) {
-                observer.onError(e);
-            }
-            subscriptions.clear();
-        }
+    public void onNext(T args) {
+        propgate(new Notification<T>(args));
     }
 
-    @Override
-    public void onNext(T args)
-    {
+    public void propgate(Notification<T> n) {
         synchronized (subscriptions) {
-            history.add(args);
+            history.add(n);
             for (Observer<? super T> observer : new ArrayList<Observer<? super T>>(subscriptions.values())) {
-                observer.onNext(args);
+                n.accept(observer);
             }
         }
     }
diff --git a/rxjava-core/src/main/java/rx/subjects/Subject.java b/rxjava-core/src/main/java/rx/subjects/Subject.java
index 0fb675c4ea..7a08b59d2b 100644
--- a/rxjava-core/src/main/java/rx/subjects/Subject.java
+++ b/rxjava-core/src/main/java/rx/subjects/Subject.java
@@ -19,7 +19,7 @@
 import rx.Observer;
 
 public abstract class Subject<T, R> extends Observable<R> implements Observer<T> {
-    protected Subject(OnSubscribeFunc<R> onSubscribe) {
+    protected Subject(OnGetSubscriptionFunc<R> onSubscribe) {
         super(onSubscribe);
     }
 }
diff --git a/rxjava-core/src/test/java/rx/operators/OperationToObservableIterableTest.java b/rxjava-core/src/test/java/rx/operators/OperationToObservableIterableTest.java
index 8d8be93dcb..fbb44fa2c2 100644
--- a/rxjava-core/src/test/java/rx/operators/OperationToObservableIterableTest.java
+++ b/rxjava-core/src/test/java/rx/operators/OperationToObservableIterableTest.java
@@ -15,6 +15,7 @@
  */
 package rx.operators;
 
+import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 import static rx.operators.OperationToObservableIterable.*;
@@ -26,6 +27,7 @@
 
 import rx.Observable;
 import rx.Observer;
+import rx.util.functions.Func1;
 
 public class OperationToObservableIterableTest {
 
@@ -42,4 +44,15 @@ public void testIterable() {
         verify(aObserver, Mockito.never()).onError(any(Throwable.class));
         verify(aObserver, times(1)).onCompleted();
     }
+
+    @Test
+    public void testInterruptable() {
+        Func1<Integer, Integer> expensive = mock(Func1.class);
+        when(expensive.call(any(Integer.class))).thenReturn(10);
+
+        int v = Observable.from(1, 2, 3).map(expensive).take(1).toBlockingObservable().single();
+        
+        assertEquals(10, v);
+        verify(expensive, times(1)).call(any(Integer.class));
+    }
 }