implement JavaFxSubscriber.toLazyBinding()

Author: thomasnield

src/main/java/rx/observers/BindingSubscriber.java

@@ -17,6 +17,7 @@
 package rx.observers;
 
 import com.sun.javafx.binding.ExpressionHelper;
+import io.reactivex.flowables.ConnectableFlowable;
 import io.reactivex.functions.Consumer;
 import javafx.beans.InvalidationListener;
 import javafx.beans.binding.Binding;
@@ -30,11 +31,18 @@
 final class BindingSubscriber<T> implements Subscriber<T>, ObservableValue<T>, Binding<T> {
 
     private final Consumer<Throwable> onError;
+    private final ConnectableFlowable<T> flowable;
+    private boolean connected = false;
     private Subscription subscription;
     private ExpressionHelper<T> helper;
     private T value;
 
     BindingSubscriber(Consumer<Throwable> onError) {
+        this.flowable = null;
+        this.onError = onError;
+    }
+    BindingSubscriber(ConnectableFlowable<T> flowable, Consumer<Throwable> onError) {
+        this.flowable = flowable;
         this.onError = onError;
     }
 
@@ -65,6 +73,10 @@ public void onNext(T t) {
     }
     @Override
     public T getValue() {
+        if (!connected && flowable != null) {
+            flowable.connect();
+            connected = true;
+        }
         return value;
     }
     @Override

src/main/java/rx/observers/JavaFxSubscriber.java

@@ -18,7 +18,9 @@
 
 import io.reactivex.Flowable;
 import io.reactivex.Observable;
+import io.reactivex.flowables.ConnectableFlowable;
 import io.reactivex.functions.Consumer;
+import io.reactivex.observables.ConnectableObservable;
 import javafx.beans.binding.Binding;
 public enum JavaFxSubscriber {
     ;//no instances
@@ -27,16 +29,35 @@ public enum JavaFxSubscriber {
      * Turns a Flowable into an eager JavaFX Binding that subscribes immediately to the Observable. Calling the Binding's dispose() method will handle the unsubscription.
      */
     public static <T> Binding<T> toBinding(Flowable<T> obs) {
-        BindingSubscriber<T> bindingObserver = new BindingSubscriber<>(e -> {});
-        obs.subscribe(bindingObserver);
-        return bindingObserver;
+        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(e -> {});
+        obs.subscribe(bindingSubscriber);
+        return bindingSubscriber;
     }
     /**
      * Turns a Flowable into an eager JavaFX Binding that subscribes immediately to the Observable. Calling the Binding's dispose() method will handle the unsubscription.
      */
     public static <T> Binding<T> toBinding(Flowable<T> obs, Consumer<Throwable> onErrorAction ) {
-        BindingSubscriber<T> bindingObserver = new BindingSubscriber<>(onErrorAction);
-        obs.subscribe(bindingObserver);
-        return bindingObserver;
+        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(onErrorAction);
+        obs.subscribe(bindingSubscriber);
+        return bindingSubscriber;
+    }
+
+    /**
+     * Turns a Flowable into an lazy JavaFX Binding that subscribes to the Observable when its getValue() is called. Calling the Binding's dispose() method will handle the unsubscription.
+     */
+    public static <T> Binding<T> toLazyBinding(Flowable<T> flowable) {
+        ConnectableFlowable<T> published = flowable.publish();
+        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(published, e -> {});
+        published.subscribe(bindingSubscriber);
+        return bindingSubscriber;
+    }
+    /**
+     * Turns a Flowable into an eager JavaFX Binding that subscribes to the Observable when its getValue() is called. Calling the Binding's dispose() method will handle the unsubscription.
+     */
+    public static <T> Binding<T> toLazyBinding(Flowable<T> flowable, Consumer<Throwable> onErrorAction ) {
+        ConnectableFlowable<T> published = flowable.publish();
+        BindingSubscriber<T> bindingSubscriber = new BindingSubscriber<>(published,onErrorAction);
+        published.subscribe(bindingSubscriber);
+        return bindingSubscriber;
     }
 }

src/test/java/rx/subscriptions/BindingTest.java

@@ -1,5 +1,6 @@
 package rx.subscriptions;
 
+import io.reactivex.Flowable;
 import io.reactivex.Observable;
 import javafx.application.Platform;
 import javafx.beans.binding.Binding;
@@ -86,7 +87,64 @@ public void testObserverLazyBinding() {
 
             sleep(1000);
 
-            System.out.println(emissionCount.get());
+            assertTrue(emissionCount.get() == 0);
+
+            binding.getValue();
+
+            sleep(1000);
+
+            assertTrue(emissionCount.get() == 4);
+            assertTrue(binding.getValue().equals("Delta"));
+            latch.countDown();
+        });
+
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Test
+    public void testSubscriberBinding() {
+        final CountDownLatch latch = new CountDownLatch(1);
+        Platform.runLater(() -> {
+
+            final AtomicInteger emissionCount = new AtomicInteger(0);
+
+            Flowable<String> items =
+                    Flowable.just("Alpha", "Beta", "Gamma", "Delta")
+                            .doOnNext(s -> emissionCount.incrementAndGet());
+
+            Binding<String> binding = JavaFxSubscriber.toBinding(items);
+
+            assertTrue(emissionCount.get() == 4);
+            assertTrue(binding.getValue().equals("Delta"));
+            latch.countDown();
+        });
+
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Test
+    public void testSubscriberLazyBinding() {
+        final CountDownLatch latch = new CountDownLatch(1);
+        Platform.runLater(() -> {
+
+            final AtomicInteger emissionCount = new AtomicInteger(0);
+
+            Flowable<String> items =
+                    Flowable.just("Alpha", "Beta", "Gamma", "Delta")
+                            .doOnNext(s -> emissionCount.incrementAndGet());
+
+            Binding<String> binding = JavaFxSubscriber.toLazyBinding(items);
+
+            sleep(1000);
+
             assertTrue(emissionCount.get() == 0);
 
             binding.getValue();