Merge pull request #14 from abersnaze/FixConversions

Fixed scheduler leak and lack of backpressure support
ReactiveX · Jun 5, 2015 · 0a2f070 · 0a2f070
2 parents 7cc5b11 + ad5ad00
commit 0a2f070
Showing 1 changed file with 11 additions and 3 deletions.
diff --git a/src/main/java/rx/observable/ListenableFutureObservable.java b/src/main/java/rx/observable/ListenableFutureObservable.java
@@ -21,6 +21,7 @@
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.functions.Action0;
+import rx.internal.operators.SingleDelayedProducer;
 import rx.Observer;
 import rx.Scheduler;
 import rx.Scheduler.Worker;
@@ -45,7 +46,11 @@ public void execute(final Runnable command) {
                 worker.schedule(new Action0() {
                     @Override
                     public void call() {
-                        command.run();
+                        try {
+                            command.run();
+                        } finally {
+                            worker.unsubscribe();
+                        }
                     }
                 });
             }
@@ -63,18 +68,21 @@ public static <T> Observable<T> from(final ListenableFuture<T> future, final Exe
         return Observable.create(new OnSubscribe<T>() {
             @Override
             public void call(final Subscriber<? super T> subscriber) {
+                final SingleDelayedProducer<T> sdp = new SingleDelayedProducer<T>(subscriber);
+                subscriber.setProducer(sdp);
+
                 future.addListener(new Runnable() {
                     @Override
                     public void run() {
                         try {
                             T t = future.get();
-                            subscriber.onNext(t);
-                            subscriber.onCompleted();
+                            sdp.set(t);
                         } catch (Exception e) {
                             subscriber.onError(e);
                         }
                     }
                 }, executor);
+
             }
         });
     }