diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
index 5d98df26b9..61e3fe8af0 100644
--- a/rxjava-core/src/main/java/rx/Observable.java
+++ b/rxjava-core/src/main/java/rx/Observable.java
@@ -44,6 +44,7 @@
 import rx.operators.OperationDebounce;
 import rx.operators.OperationDefaultIfEmpty;
 import rx.operators.OperationDefer;
+import rx.operators.OperationDeferFuture;
 import rx.operators.OperationDelay;
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationDistinct;
@@ -53,6 +54,7 @@
 import rx.operators.OperationFilter;
 import rx.operators.OperationFinally;
 import rx.operators.OperationFirstOrDefault;
+import rx.operators.OperationFromFuture;
 import rx.operators.OperationGroupBy;
 import rx.operators.OperationGroupByUntil;
 import rx.operators.OperationGroupJoin;
@@ -82,6 +84,7 @@
 import rx.operators.OperationSkipLast;
 import rx.operators.OperationSkipUntil;
 import rx.operators.OperationSkipWhile;
+import rx.operators.OperationStartFuture;
 import rx.operators.OperationSubscribeOn;
 import rx.operators.OperationSum;
 import rx.operators.OperationSwitch;
@@ -114,6 +117,7 @@
 import rx.subjects.PublishSubject;
 import rx.subjects.ReplaySubject;
 import rx.subjects.Subject;
+import rx.subscriptions.BooleanSubscription;
 import rx.subscriptions.Subscriptions;
 import rx.util.OnErrorNotImplementedException;
 import rx.util.Range;
@@ -6818,6 +6822,189 @@ public <TKey, TValue, TDuration> Observable<GroupedObservable<TKey, TValue>> gro
         return create(new OperationGroupByUntil<T, TKey, TValue, TDuration>(this, keySelector, valueSelector, durationSelector));
     }
 
+    /**
+     * Invokes the asynchronous function immediately, surfacing the result through an observable sequence.
+     * <p>
+     * <em>Important note</em> subscribing to the resulting observable blocks until
+     * the future completes.
+     * @param <T> the result type
+     * @param functionAsync the asynchronous function to run
+     * @return an observable which surfaces the result of the future.
+     * @see #startFuture(rx.util.functions.Func0, rx.Scheduler) 
+     */
+    public static <T> Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync) {
+        return OperationStartFuture.startFuture(functionAsync);
+    }
+    
+    /**
+     * Invokes the asynchronous function, surfacing the result through an observable
+     * sequence and shares a BooleanSubscription between all subscribers.
+     * <p>
+     * <em>Important note</em> subscribing to the resulting observable blocks until
+     * the future completes.
+     * @param <T> the result type
+     * @param functionAsync the asynchronous function to run
+     * @return an observable which surfaces the result of the future and shares
+     *         a common subscription between subscribers
+     * @see #startCancellableFuture(rx.util.functions.Func1, rx.Scheduler) 
+     */
+    public static <T> Observable<T> startCancellableFuture(Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync) {
+        return OperationStartFuture.startFuture(functionAsync);
+    }
+    
+    /**
+     * Invokes the asynchronous function immediately, surfacing the result through an observable sequence
+     * and waits on the specified scheduler.
+     * @param <T> the result type
+     * @param functionAsync the asynchronous function to run
+     * @param scheduler the scheduler where the completion of the Future is awaited
+     * @return an observable which surfaces the result of the future.
+     */
+    public static <T> Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync,
+            Scheduler scheduler) {
+        return OperationStartFuture.startFuture(functionAsync, scheduler);
+    }
+    
+    /**
+     * Invokes the asynchronous function, surfacing the result through an observable
+     * sequence and shares a BooleanSubscription between all subscribers
+     * and waits on the specified scheduler.
+     * @param <T> the result type
+     * @param functionAsync the asynchronous function to run
+     * @param scheduler the scheduler where the completion of the Future is awaited
+     * @return an observable which surfaces the result of the future and shares
+     *         a common subscription between subscribers
+     */
+    public static <T> Observable<T> startCancellableFuture(
+            Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync,
+            Scheduler scheduler) {
+        return OperationStartFuture.startFuture(functionAsync, scheduler);
+    }
+    
+    /**
+     * Returns an observable sequence that starts the specified asynchronous 
+     * factory function whenever a new observer subscribes.
+     * <p>
+     * <em>Important note</em> subscribing to the resulting observable blocks until
+     * the future completes.
+     * @param <T> the result type
+     * @param observableFactoryAsync the asynchronous function to start for each observer
+     * @return the observable sequence containing values produced by the asynchronous observer
+     * produced by the factory
+     * @see #deferFuture(rx.util.functions.Func0, rx.Scheduler) 
+     */
+    public static <T> Observable<T> deferFuture(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) {
+        return OperationDeferFuture.deferFuture(observableFactoryAsync);
+    }
+    
+    /**
+     * Returns an observable sequence that starts the specified asynchronous 
+     * factory function whenever a new observer subscribes and shares a boolean
+     * subscription between all subscribers.
+     * <p>
+     * <em>Important note</em> subscribing to the resulting observable blocks until
+     * the future completes.
+     * @param <T> the result type
+     * @param observableFactoryAsync the asynchronous function to start for each observer
+     * @return the observable sequence containing values produced by the asynchronous observer
+     * produced by the factory
+     * @see #deferCancellableFuture(rx.util.functions.Func1, rx.Scheduler) 
+     */
+    public static <T> Observable<T> deferCancellableFuture(Func1<? super BooleanSubscription, ? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) {
+        return OperationDeferFuture.deferFuture(observableFactoryAsync);
+    }
+    
+    /**
+     * Returns an observable sequence that starts the specified asynchronous 
+     * factory function whenever a new observer subscribes.
+     * @param <T> the result type
+     * @param observableFactoryAsync the asynchronous function to start for each observer
+     * @param scheduler the scheduler where the completion of the Future is awaited
+     * @return the observable sequence containing values produced by the asynchronous observer
+     * produced by the factory
+     */
+    public static <T> Observable<T> deferFuture(
+            Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync,
+            Scheduler scheduler) {
+        return OperationDeferFuture.deferFuture(observableFactoryAsync, scheduler);
+    }
+    
+    /**
+     * Returns an observable sequence that starts the specified asynchronous 
+     * factory function whenever a new observer subscribes and shares a boolean
+     * subscription between all subscribers.
+     * @param <T> the result type
+     * @param observableFactoryAsync the asynchronous function to start for each observer
+     * @param scheduler the scheduler where the completion of the Future is awaited
+     * @return the observable sequence containing values produced by the asynchronous observer
+     * produced by the factory
+     */
+    public static <T> Observable<T> deferCancellableFuture(
+            Func1<? super BooleanSubscription, ? extends Future<? extends Observable<? extends T>>> observableFactoryAsync,
+            Scheduler scheduler) {
+        return OperationDeferFuture.deferFuture(observableFactoryAsync, scheduler);
+    }
+    
+    /**
+     * Converts to asynchronous function into an observable sequence where 
+     * each subscription to the resulting sequence causes the function to be started.
+     * <p>
+     * <em>Important note</em> subscribing to the resulting observable blocks until
+     * the future completes.
+     * @param <T> the result type
+     * @param functionAsync the function to call when an Observer subscribes
+     * @return an Observable which returns the single result or exception of the future
+     * @see #fromFuture(rx.util.functions.Func0, rx.Scheduler) 
+     */
+    public static <T> Observable<T> fromFuture(Func0<? extends Future<? extends T>> functionAsync) {
+        return create(OperationFromFuture.fromFuture(functionAsync));
+    }
+    /**
+     * Converts to asynchronous function into an observable sequence where 
+     * each subscription to the resulting sequence causes the function to be started
+     * and run on the given scheduler.
+     * @param <T> the result type
+     * @param functionAsync the function to call when an Observer subscribes
+     * @param scheduler the scheduler where the Future returned by the functionAsync is started
+     * @return an Observable which returns the single result or exception of the future
+     */
+    public static <T> Observable<T> fromFuture(Func0<? extends Future<? extends T>> functionAsync, Scheduler scheduler) {
+        return create(OperationFromFuture.fromFuture(functionAsync, scheduler));
+    }
+    /**
+     * Converts to asynchronous function into an observable sequence
+     * where each subscription to the resulting sequence causes the function to be started and
+     * the passed-in BooleanSubscription is tied to the Observable sequence's subscription.
+     * <p>
+     * <em>Important note</em> subscribing to the resulting observable blocks until
+     * the future completes.
+     * @param <T> the result type
+     * @param functionAsync the function to call when an Observer subscribes which receives
+     *                      a BooleanSubscription that is tied to the subscription of the returned Observable
+     * @return an Observable which returns the single result or exception of the future
+     * @see #fromCancellableFuture(rx.util.functions.Func1, rx.Scheduler) 
+     */
+    public static <T> Observable<T> fromCancellableFuture(Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync) {
+        return create(OperationFromFuture.fromFuture(functionAsync));
+    }
+    
+    /**
+     * Converts to asynchronous function into an observable sequence
+     * where each subscription to the resulting sequence causes the function to be started and
+     * run on the given scheduler, and
+     * the passed-in BooleanSubscription is tied to the Observable sequence's subscription.
+     * @param <T> the result type
+     * @param functionAsync the function to call when an Observer subscribes which receives
+     *                      a BooleanSubscription that is tied to the subscription of the returned Observable
+     * @param scheduler the scheduler where the Future returned by the functionAsync is started
+     * @return an Observable which returns the single result or exception of the future
+     */
+    public static <T> Observable<T> fromCancellableFuture(
+            Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync,
+            Scheduler scheduler) {
+        return create(OperationFromFuture.fromFuture(functionAsync, scheduler));
+    }
+    
     /**
      * Invokes the specified function asynchronously and returns an Observable
      * that emits the result.
diff --git a/rxjava-core/src/main/java/rx/operators/OperationDeferFuture.java b/rxjava-core/src/main/java/rx/operators/OperationDeferFuture.java
new file mode 100644
index 0000000000..5673c734c0
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/operators/OperationDeferFuture.java
@@ -0,0 +1,147 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.Future;
+import rx.Observable;
+import rx.Scheduler;
+import rx.subscriptions.BooleanSubscription;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+
+/**
+ * Defer the execution of a factory method which produces an observable sequence.
+ */
+public final class OperationDeferFuture {
+    /** Utility class. */
+    private OperationDeferFuture() { throw new IllegalStateException("No instances!"); }
+    
+    /**
+     * Returns an observable sequence that starts the specified asynchronous 
+     * factory function whenever a new observer subscribes.
+     * @param <T> the result type
+     * @param observableFactoryAsync the asynchronous function to start for each observer
+     * @return the observable sequence containing values produced by the asynchronous observer
+     * produced by the factory
+     */
+    public static <T> Observable<T> deferFuture(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) {
+        return Observable.defer(new DeferFutureFunc0<T>(observableFactoryAsync));
+    }
+    /** The function called by the defer operator. */
+    private static final class DeferFutureFunc0<T> implements Func0<Observable<T>> {
+        final Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync;
+
+        public DeferFutureFunc0(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) {
+            this.observableFactoryAsync = observableFactoryAsync;
+        }
+        
+        @Override
+        public Observable<T> call() {
+            return Observable.merge(OperationStartFuture.startFuture(observableFactoryAsync));
+        }
+        
+    }
+    /**
+     * Returns an observable sequence that starts the specified asynchronous 
+     * factory function whenever a new observer subscribes and shares a boolean
+     * subscription between all subscribers.
+     * @param <T> the result type
+     * @param observableFactoryAsync the asynchronous function to start for each observer
+     * @return the observable sequence containing values produced by the asynchronous observer
+     * produced by the factory
+     */
+    public static <T> Observable<T> deferFuture(Func1<? super BooleanSubscription, ? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) {
+        return Observable.defer(new DeferFutureFunc1<T>(observableFactoryAsync));
+    }
+    /** The function called by the defer operator with boolean subscription. */
+    private static final class DeferFutureFunc1<T> implements Func0<Observable<T>> {
+        final Func1<? super BooleanSubscription, ? extends Future<? extends Observable<? extends T>>> observableFactoryAsync;
+
+        public DeferFutureFunc1(Func1<? super BooleanSubscription, ? extends Future<? extends Observable<? extends T>>> observableFactoryAsync) {
+            this.observableFactoryAsync = observableFactoryAsync;
+        }
+        
+        @Override
+        public Observable<T> call() {
+            return Observable.merge(OperationStartFuture.startFuture(observableFactoryAsync));
+        }
+        
+    }
+    
+    /**
+     * Returns an observable sequence that starts the specified asynchronous 
+     * factory function whenever a new observer subscribes.
+     * @param <T> the result type
+     * @param observableFactoryAsync the asynchronous function to start for each observer
+     * @param scheduler the scheduler where the completion of the Future is awaited
+     * @return the observable sequence containing values produced by the asynchronous observer
+     * produced by the factory
+     */
+    public static <T> Observable<T> deferFuture(
+            Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync,
+            Scheduler scheduler) {
+        return Observable.defer(new DeferFutureFunc0Scheduled<T>(observableFactoryAsync, scheduler));
+    }
+    /** The function called by the defer operator. */
+    private static final class DeferFutureFunc0Scheduled<T> implements Func0<Observable<T>> {
+        final Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync;
+        final Scheduler scheduler;
+
+        public DeferFutureFunc0Scheduled(Func0<? extends Future<? extends Observable<? extends T>>> observableFactoryAsync,
+                Scheduler scheduler) {
+            this.observableFactoryAsync = observableFactoryAsync;
+            this.scheduler = scheduler;
+        }
+        
+        @Override
+        public Observable<T> call() {
+            return Observable.merge(OperationStartFuture.startFuture(observableFactoryAsync, scheduler));
+        }
+        
+    }
+    /**
+     * Returns an observable sequence that starts the specified asynchronous 
+     * factory function whenever a new observer subscribes and shares a boolean
+     * subscription between all subscribers.
+     * @param <T> the result type
+     * @param observableFactoryAsync the asynchronous function to start for each observer
+     * @param scheduler the scheduler where the completion of the Future is awaited
+     * @return the observable sequence containing values produced by the asynchronous observer
+     * produced by the factory
+     */
+    public static <T> Observable<T> deferFuture(
+            Func1<? super BooleanSubscription, ? extends Future<? extends Observable<? extends T>>> observableFactoryAsync,
+            Scheduler scheduler) {
+        return Observable.defer(new DeferFutureFunc1Scheduled<T>(observableFactoryAsync, scheduler));
+    }
+    /** The function called by the defer operator with boolean subscription. */
+    private static final class DeferFutureFunc1Scheduled<T> implements Func0<Observable<T>> {
+        final Func1<? super BooleanSubscription, ? extends Future<? extends Observable<? extends T>>> observableFactoryAsync;
+        final Scheduler scheduler;
+
+        public DeferFutureFunc1Scheduled(Func1<? super BooleanSubscription, ? extends Future<? extends Observable<? extends T>>> observableFactoryAsync,
+                Scheduler scheduler) {
+            this.observableFactoryAsync = observableFactoryAsync;
+            this.scheduler = scheduler;
+        }
+        
+        @Override
+        public Observable<T> call() {
+            return Observable.merge(OperationStartFuture.startFuture(observableFactoryAsync, scheduler));
+        }
+        
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/operators/OperationFromFuture.java b/rxjava-core/src/main/java/rx/operators/OperationFromFuture.java
new file mode 100644
index 0000000000..2424f9f1cb
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/operators/OperationFromFuture.java
@@ -0,0 +1,126 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.Future;
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Scheduler;
+import rx.subscriptions.BooleanSubscription;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+
+/**
+ * Converts to asynchronous function into an observable sequence where 
+ * each subscription to the resulting sequence causes the function to be started.
+ */
+public final class OperationFromFuture {
+    /** Utility class. */
+    private OperationFromFuture() { throw new IllegalStateException("No instances!"); }
+    /**
+     * Converts to asynchronous function into an observable sequence where 
+     * each subscription to the resulting sequence causes the function to be started.
+     */
+    public static <T> OnSubscribeFunc<T> fromFuture(Func0<? extends Future<? extends T>> functionAsync) {
+        return OperationDefer.defer(new FromFuture<T>(functionAsync));
+    }
+    /**
+     * Converts to asynchronous function into an observable sequence where 
+     * each subscription to the resulting sequence causes the function to be started
+     * and run on the given scheduler.
+     */
+    public static <T> OnSubscribeFunc<T> fromFuture(Func0<? extends Future<? extends T>> functionAsync, Scheduler scheduler) {
+        return OperationDefer.defer(new FromFutureScheduled<T>(functionAsync, scheduler));
+    }
+    /**
+     * Converts to asynchronous function into an observable sequence
+     * where each subscription to the resulting sequence causes the function to be started and
+     * the passed-in BooleanSubscription is tied to the Observable sequence's subscription.
+     */
+    public static <T> OnSubscribeFunc<T> fromFuture(Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync) {
+        return OperationDefer.defer(new FromFutureToken<T>(functionAsync));
+    }
+    
+    /**
+     * Converts to asynchronous function into an observable sequence
+     * where each subscription to the resulting sequence causes the function to be started and
+     * run on the given scheduler, and
+     * the passed-in BooleanSubscription is tied to the Observable sequence's subscription.
+     */
+    public static <T> OnSubscribeFunc<T> fromFuture(
+            Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync,
+            Scheduler scheduler) {
+        return OperationDefer.defer(new FromFutureTokenScheduled<T>(functionAsync, scheduler));
+    }
+    
+    /** From a future started by OperationStartFuture. */
+    private static final class FromFuture<T> implements Func0<Observable<T>> {
+        final Func0<? extends Future<? extends T>> functionAsync;
+
+        public FromFuture(Func0<? extends Future<? extends T>> functionAsync) {
+            this.functionAsync = functionAsync;
+        }
+
+        @Override
+        public Observable<T> call() {
+            return OperationStartFuture.startFuture(functionAsync);
+        }
+    }
+    /** From a future started by OperationStartFuture. */
+    private static final class FromFutureScheduled<T> implements Func0<Observable<T>> {
+        final Func0<? extends Future<? extends T>> functionAsync;
+        final Scheduler scheduler;
+
+        public FromFutureScheduled(Func0<? extends Future<? extends T>> functionAsync, Scheduler scheduler) {
+            this.functionAsync = functionAsync;
+            this.scheduler = scheduler;
+        }
+
+        @Override
+        public Observable<T> call() {
+            return OperationStartFuture.startFuture(functionAsync, scheduler);
+        }
+    }
+    
+    /** From a future started by OperationStartFuture. */
+    private static final class FromFutureToken<T> implements Func0<Observable<T>> {
+        final Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync;
+
+        public FromFutureToken(Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync) {
+            this.functionAsync = functionAsync;
+        }
+
+        @Override
+        public Observable<T> call() {
+            return OperationStartFuture.startFuture(functionAsync);
+        }
+    }
+    /** From a future started by OperationStartFuture. */
+    private static final class FromFutureTokenScheduled<T> implements Func0<Observable<T>> {
+        final Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync;
+        final Scheduler scheduler;
+
+        public FromFutureTokenScheduled(Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync, Scheduler scheduler) {
+            this.functionAsync = functionAsync;
+            this.scheduler = scheduler;
+        }
+
+        @Override
+        public Observable<T> call() {
+            return OperationStartFuture.startFuture(functionAsync, scheduler);
+        }
+    }
+}
diff --git a/rxjava-core/src/main/java/rx/operators/OperationStartFuture.java b/rxjava-core/src/main/java/rx/operators/OperationStartFuture.java
new file mode 100644
index 0000000000..f9eef5cb40
--- /dev/null
+++ b/rxjava-core/src/main/java/rx/operators/OperationStartFuture.java
@@ -0,0 +1,134 @@
+ /**
+  * Copyright 2013 Netflix, Inc.
+  *
+  * Licensed under the Apache License, Version 2.0 (the "License");
+  * you may not use this file except in compliance with the License.
+  * You may obtain a copy of the License at
+  *
+  * http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS,
+  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  * See the License for the specific language governing permissions and
+  * limitations under the License.
+  */
+package rx.operators;
+
+import java.util.concurrent.Future;
+import rx.Observable;
+import rx.Observable.OnSubscribeFunc;
+import rx.Observer;
+import rx.Scheduler;
+import rx.Subscription;
+import rx.subscriptions.BooleanSubscription;
+import rx.subscriptions.CompositeSubscription;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+
+/**
+ * Start an asynchronous Future immediately and observe its result through
+ * an observable.
+ */
+public final class OperationStartFuture {
+    /** Utility class. */
+    private OperationStartFuture() { throw new IllegalStateException("No instances!"); }
+    /**
+     * Invokes the asynchronous function, surfacing the result through an observable sequence.
+     * <p>
+     * <em>Important note</em> subscribing to the resulting observable blocks until
+     * the future completes.
+     * @param <T> the result type
+     * @param functionAsync the asynchronous function to run
+     * @return the observable
+     */
+    public static <T> Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync) {
+        Future<? extends T> task;
+        try {
+            task = functionAsync.call();
+        } catch (Throwable t) {
+            return Observable.error(t);
+        }
+        return Observable.from(task);
+    }
+    /**
+     * Invokes the asynchronous function, surfacing the result through an observable sequence
+     * running on the given scheduler.
+     * @param <T> the result type
+     * @param functionAsync the asynchronous function to run
+     * @param scheduler the scheduler where the completion of the Future is awaited
+     * @return the observable
+     */
+    public static <T> Observable<T> startFuture(Func0<? extends Future<? extends T>> functionAsync,
+            Scheduler scheduler) {
+        Future<? extends T> task;
+        try {
+            task = functionAsync.call();
+        } catch (Throwable t) {
+            return Observable.error(t);
+        }
+        return Observable.from(task, scheduler);
+    }
+    /**
+     * Invokes the asynchronous function, surfacing the result through an observable
+     * sequence and shares a BooleanSubscription between all subscribers.
+     * <em>Important note</em> subscribing to the resulting observable blocks until
+     * the future completes.
+     * @param <T> the result type
+     * @param functionAsync the asynchronous function to run
+     * @return  the observable
+     */
+    public static <T> Observable<T> startFuture(Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync) {
+        final BooleanSubscription token = new BooleanSubscription();
+        Future<? extends T> task;
+        try {
+            task = functionAsync.call(token);
+        } catch (Throwable t) {
+            return Observable.error(t);
+        }
+        final Observable<T> result = Observable.from(task);
+        return Observable.create(new StartAsyncSubscribe<T>(token, result));
+    }
+    
+    /**
+     * Invokes the asynchronous function, surfacing the result through an observable
+     * sequence and shares a BooleanSubscription between all subscribers.
+     * @param <T> the result type
+     * @param functionAsync the asynchronous function to run
+     * @param scheduler the scheduler where the completion of the Future is awaited
+     * @return  the observable
+     */
+    public static <T> Observable<T> startFuture(
+            Func1<? super BooleanSubscription, ? extends Future<? extends T>> functionAsync,
+            Scheduler scheduler) {
+        final BooleanSubscription token = new BooleanSubscription();
+        Future<? extends T> task;
+        try {
+            task = functionAsync.call(token);
+        } catch (Throwable t) {
+            return Observable.error(t);
+        }
+        final Observable<T> result = Observable.from(task, scheduler);
+        return Observable.create(new StartAsyncSubscribe<T>(token, result));
+    }
+    /** 
+     * The subsription function that combines the token and the subscription to the
+     * source.
+     * @param <T> the value type
+     */
+    private static final class StartAsyncSubscribe<T> implements OnSubscribeFunc<T> {
+        final Subscription token;
+        final Observable<? extends T> source;
+
+        public StartAsyncSubscribe(Subscription token, Observable<? extends T> source) {
+            this.token = token;
+            this.source = source;
+        }
+
+        @Override
+        public Subscription onSubscribe(Observer<? super T> t1) {
+            Subscription s = source.subscribe(t1);
+            return new CompositeSubscription(token, s);
+        }
+    }   
+}
diff --git a/rxjava-core/src/test/java/rx/operators/OperationDeferFutureTest.java b/rxjava-core/src/test/java/rx/operators/OperationDeferFutureTest.java
new file mode 100644
index 0000000000..8660b06033
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/operators/OperationDeferFutureTest.java
@@ -0,0 +1,104 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import static org.junit.Assert.fail;
+import org.junit.Test;
+import org.mockito.InOrder;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import rx.Observable;
+import rx.Observer;
+import rx.schedulers.Schedulers;
+import rx.util.functions.Func0;
+
+public class OperationDeferFutureTest {
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testSimple() throws InterruptedException {
+        final ExecutorService exec = Executors.newCachedThreadPool();
+        try {
+            final CountDownLatch ready = new CountDownLatch(1);
+
+            Func0<Future<Observable<Integer>>> func = new Func0<Future<Observable<Integer>>>() {
+                @Override
+                public Future<Observable<Integer>> call() {
+                    return exec.submit(new Callable<Observable<Integer>>() {
+                        @Override
+                        public Observable<Integer> call() throws Exception {
+                            if (!ready.await(1000, TimeUnit.MILLISECONDS)) {
+                                throw new IllegalStateException("Not started in time");
+                            }
+                            return Observable.from(1);
+                        }
+                    });
+                }
+            };
+            
+            Observable<Integer> result = Observable.deferFuture(func, Schedulers.threadPoolForComputation());
+
+            final Observer<Integer> observer = mock(Observer.class);
+
+            final CountDownLatch done = new CountDownLatch(1);
+
+            result.subscribe(new OperationStartFutureTest.MockHelper<Integer>(observer, done));
+
+            ready.countDown();
+
+            if (!done.await(1500, TimeUnit.MILLISECONDS)) {
+                fail("Not completed in time!");
+            }
+
+            InOrder inOrder = inOrder(observer);
+
+            inOrder.verify(observer).onNext(1);
+            inOrder.verify(observer).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+        } finally {
+            exec.shutdown();
+        }
+    }
+    
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testSimpleFactoryThrows() {
+        Func0<Future<Observable<Integer>>> func = new Func0<Future<Observable<Integer>>>() {
+
+            @Override
+            public Future<Observable<Integer>> call() {
+                throw new OperationStartFutureTest.CustomException();
+            }
+        };
+        
+        Observable<Integer> result = Observable.deferFuture(func);
+        
+        final Observer<Object> observer = mock(Observer.class);
+        result.subscribe(observer);
+        
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+        verify(observer).onError(any(OperationStartFutureTest.CustomException.class));
+    }
+}
diff --git a/rxjava-core/src/test/java/rx/operators/OperationFromFutureTest.java b/rxjava-core/src/test/java/rx/operators/OperationFromFutureTest.java
new file mode 100644
index 0000000000..8c7eef2186
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/operators/OperationFromFutureTest.java
@@ -0,0 +1,122 @@
+ /**
+  * Copyright 2013 Netflix, Inc.
+  *
+  * Licensed under the Apache License, Version 2.0 (the "License");
+  * you may not use this file except in compliance with the License.
+  * You may obtain a copy of the License at
+  *
+  * http://www.apache.org/licenses/LICENSE-2.0
+  *
+  * Unless required by applicable law or agreed to in writing, software
+  * distributed under the License is distributed on an "AS IS" BASIS,
+  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  * See the License for the specific language governing permissions and
+  * limitations under the License.
+  */
+package rx.operators;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import org.junit.Test;
+import org.mockito.InOrder;
+import static org.mockito.Mockito.*;
+import rx.Observable;
+import rx.Observer;
+import rx.operators.OperationStartFutureTest.CustomException;
+import rx.schedulers.Schedulers;
+import rx.util.functions.Func0;
+
+public class OperationFromFutureTest {
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testSimple() throws InterruptedException {
+        final ExecutorService exec = Executors.newCachedThreadPool();
+        
+        try {
+            final AtomicInteger runCount = new AtomicInteger(0);
+            
+            Func0<Future<Integer>> func = new Func0<Future<Integer>>() {
+                @Override
+                public Future<Integer> call() {
+                    return exec.submit(new Callable<Integer>() {
+                        @Override
+                        public Integer call() throws Exception {
+                            return runCount.incrementAndGet();
+                        }
+                    });
+                }
+            };
+            
+            Observable<Integer> result = Observable.fromFuture(func, Schedulers.threadPoolForComputation());
+            
+            for (int i = 1; i <= 10; i++) {
+                Observer<Object> o = mock(Observer.class);
+                InOrder inOrder = inOrder(o);
+                
+                final CountDownLatch done = new CountDownLatch(1);
+                
+                result.subscribe(new OperationStartFutureTest.MockHelper<Integer>(o, done));
+                
+                if (!done.await(1500, TimeUnit.MILLISECONDS)) {
+                    fail("Not completed in time!");
+                }
+                
+                inOrder.verify(o).onNext(i);
+                inOrder.verify(o).onCompleted();
+                verify(o, never()).onError(any(Throwable.class));
+                
+                assertEquals(i, runCount.get());
+            }
+            
+        } finally {
+            exec.shutdown();
+        }
+    }
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testSimpleThrows() throws InterruptedException {
+        final ExecutorService exec = Executors.newCachedThreadPool();
+        
+        try {
+            Func0<Future<Integer>> func = new Func0<Future<Integer>>() {
+                @Override
+                public Future<Integer> call() {
+                    return exec.submit(new Callable<Integer>() {
+                        @Override
+                        public Integer call() throws Exception {
+                            throw new CustomException();
+                        }
+                    });
+                }
+            };
+            
+            Observable<Integer> result = Observable.fromFuture(func, Schedulers.threadPoolForComputation());
+            
+            for (int i = 1; i <= 10; i++) {
+                Observer<Object> o = mock(Observer.class);
+                
+                final CountDownLatch done = new CountDownLatch(1);
+                
+                result.subscribe(new OperationStartFutureTest.MockHelper<Integer>(o, done));
+                
+                if (!done.await(1500, TimeUnit.MILLISECONDS)) {
+                    fail("Not completed in time!");
+                }
+                
+                verify(o, times(1)).onError(any(CustomException.class));
+                verify(o, never()).onNext(any());
+                verify(o, never()).onCompleted();
+            }
+            
+        } finally {
+            exec.shutdown();
+        }
+    }
+}
diff --git a/rxjava-core/src/test/java/rx/operators/OperationStartFutureTest.java b/rxjava-core/src/test/java/rx/operators/OperationStartFutureTest.java
new file mode 100644
index 0000000000..2df51a7b3f
--- /dev/null
+++ b/rxjava-core/src/test/java/rx/operators/OperationStartFutureTest.java
@@ -0,0 +1,278 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import static org.junit.Assert.fail;
+import org.junit.Test;
+import org.mockito.InOrder;
+import static org.mockito.Mockito.*;
+import rx.Observable;
+import rx.Observer;
+import rx.Subscription;
+import rx.schedulers.Schedulers;
+import rx.subscriptions.BooleanSubscription;
+import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+
+public class OperationStartFutureTest {
+    /** Custom exception to distinguish from any other RuntimeException. */
+    static class CustomException extends RuntimeException {}
+    /** 
+     * Forwards the events to the underlying observer and counts down the latch
+     * on terminal conditions.
+     * @param <T> 
+     */
+    static class MockHelper<T> implements Observer<T> {
+        final Observer<? super T> observer;
+        final CountDownLatch latch;
+
+        public MockHelper(Observer<? super T> observer, CountDownLatch latch) {
+            this.observer = observer;
+            this.latch = latch;
+        }
+
+        @Override
+        public void onNext(T args) {
+            try {
+                observer.onNext(args);
+            } catch (Throwable t) {
+                onError(t);
+            }
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            try {
+                observer.onError(e);
+            } finally {
+                latch.countDown();
+            }
+        }
+
+
+        @Override
+        public void onCompleted() {
+            try {
+                observer.onCompleted();
+            } finally {
+                latch.countDown();
+            }
+        }
+        
+    }
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testSimple() throws InterruptedException {
+        final ExecutorService exec = Executors.newCachedThreadPool();
+        try {
+            final CountDownLatch ready = new CountDownLatch(1);
+
+            Func0<Future<Integer>> func = new Func0<Future<Integer>>() {
+
+                @Override
+                public Future<Integer> call() {
+                    return exec.submit(new Callable<Integer>() {
+                        @Override
+                        public Integer call() throws Exception {
+                            if (!ready.await(1000, TimeUnit.MILLISECONDS)) {
+                                throw new IllegalStateException("Not started in time");
+                            }
+                            return 1;
+                        }
+                    });
+                }
+            };
+
+            Observable<Integer> result = Observable.startFuture(func, Schedulers.threadPoolForComputation());
+
+            final Observer<Integer> observer = mock(Observer.class);
+
+            final CountDownLatch done = new CountDownLatch(1);
+
+            result.subscribe(new MockHelper<Integer>(observer, done));
+
+            ready.countDown();
+
+            if (!done.await(1500, TimeUnit.MILLISECONDS)) {
+                fail("Not completed in time!");
+            }
+
+            InOrder inOrder = inOrder(observer);
+
+            inOrder.verify(observer).onNext(1);
+            inOrder.verify(observer).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+        } finally {        
+            exec.shutdown();
+        }
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testSimpleFactoryThrows() {
+        Func0<Future<Integer>> func = new Func0<Future<Integer>>() {
+
+            @Override
+            public Future<Integer> call() {
+                throw new CustomException();
+            }
+        };
+        
+        Observable<Integer> result = Observable.startFuture(func);
+        
+        final Observer<Object> observer = mock(Observer.class);
+        result.subscribe(observer);
+        
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+        verify(observer).onError(any(CustomException.class));
+    }
+    
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testSimpleCancellableFactoryThrows() {
+        Func1<BooleanSubscription, Future<Integer>> func = new Func1<BooleanSubscription, Future<Integer>>() {
+
+            @Override
+            public Future<Integer> call(BooleanSubscription token) {
+                throw new CustomException();
+            }
+        };
+        
+        Observable<Integer> result = Observable.startCancellableFuture(func);
+        
+        final Observer<Object> observer = mock(Observer.class);
+        result.subscribe(observer);
+        
+        verify(observer, never()).onNext(any());
+        verify(observer, never()).onCompleted();
+        verify(observer).onError(any(CustomException.class));
+    }
+    
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testStartCancellable() throws InterruptedException {
+        final ExecutorService exec = Executors.newCachedThreadPool();
+        try {
+            final CountDownLatch ready = new CountDownLatch(1);
+
+            Func1<BooleanSubscription, Future<Integer>> func = new Func1<BooleanSubscription, Future<Integer>>() {
+
+                @Override
+                public Future<Integer> call(BooleanSubscription token) {
+                    return exec.submit(new Callable<Integer>() {
+                        @Override
+                        public Integer call() throws Exception {
+                            if (!ready.await(1000, TimeUnit.MILLISECONDS)) {
+                                throw new IllegalStateException("Not started in time");
+                            }
+                            return 1;
+                        }
+                    });
+                }
+            };
+
+            Observable<Integer> result = Observable.startCancellableFuture(func, Schedulers.threadPoolForComputation());
+
+            final Observer<Integer> observer = mock(Observer.class);
+
+            final CountDownLatch done = new CountDownLatch(1);
+
+            result.subscribe(new MockHelper<Integer>(observer, done));
+
+            ready.countDown();
+
+            if (!done.await(1500, TimeUnit.MILLISECONDS)) {
+                fail("Not completed in time!");
+            }
+
+            InOrder inOrder = inOrder(observer);
+
+            inOrder.verify(observer).onNext(1);
+            inOrder.verify(observer).onCompleted();
+            verify(observer, never()).onError(any(Throwable.class));
+        } finally {        
+            exec.shutdown();
+        }
+    }
+    
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testStartCancellableAndCancel() throws InterruptedException {
+        final ExecutorService exec = Executors.newCachedThreadPool();
+        try {
+            final CountDownLatch ready = new CountDownLatch(1);
+            final CountDownLatch ready2 = new CountDownLatch(1);
+
+            Func1<BooleanSubscription, Future<Integer>> func = new Func1<BooleanSubscription, Future<Integer>>() {
+
+                @Override
+                public Future<Integer> call(final BooleanSubscription token) {
+                    return exec.submit(new Callable<Integer>() {
+                        @Override
+                        public Integer call() throws Exception {
+                            ready.countDown();
+                            
+                            if (!ready2.await(1000, TimeUnit.MILLISECONDS)) {
+                                throw new IllegalStateException("Not started in time");
+                            }
+                            if (token.isUnsubscribed()) {
+                                throw new CancellationException();
+                            }
+                            
+                            return 1;
+                        }
+                    });
+                }
+            };
+
+            Observable<Integer> result = Observable.startCancellableFuture(func, Schedulers.threadPoolForComputation());
+
+            final Observer<Object> observer = mock(Observer.class);
+
+            final CountDownLatch done = new CountDownLatch(1);
+
+            Subscription s = result.subscribe(new MockHelper<Object>(observer, done));
+
+            if (!ready.await(1000, TimeUnit.MILLISECONDS)) {
+                fail("Not entered the call() in time!");
+            }
+            
+            s.unsubscribe();
+            
+            // resume call
+            ready2.countDown();
+
+            if (!done.await(1500, TimeUnit.MILLISECONDS)) {
+                fail("Not completed in time!");
+            }
+
+            verify(observer, never()).onNext(any());
+            verify(observer, never()).onCompleted();
+            verify(observer, times(1)).onError(any(CancellationException.class));
+            
+        } finally {        
+            exec.shutdown();
+        }
+    }
+}