Implemented publishLast

talshani · Sep 24, 2013 · d4b04d8 · d4b04d8
1 parent e0f57f1
commit d4b04d8
Show file tree

Hide file tree

Showing 2 changed files with 51 additions and 0 deletions.
diff --git a/rxjava-core/src/main/java/rx/Observable.java b/rxjava-core/src/main/java/rx/Observable.java
@@ -83,6 +83,7 @@
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
+import rx.subjects.AsyncSubject;
 import rx.subjects.PublishSubject;
 import rx.subjects.ReplaySubject;
 import rx.subjects.Subject;
@@ -3634,6 +3635,14 @@ public ConnectableObservable<T> publish() {
         return OperationMulticast.multicast(this, PublishSubject.<T> create());
     }
 
+    /**
+     * Returns a {@link ConnectableObservable} that shares a single subscription that contains the last notification only.
+     * @return a {@link ConnectableObservable}
+     */
+    public ConnectableObservable<T> publishLast() {
+        return OperationMulticast.multicast(this, AsyncSubject.<T> create());
+    }
+
     /**
      * Synonymous with <code>reduce()</code>.
      * <p>

diff --git a/rxjava-core/src/test/java/rx/ObservableTests.java b/rxjava-core/src/test/java/rx/ObservableTests.java
@@ -487,6 +487,48 @@ public void call(String v) {
         }
     }
 
+    @Test
+    public void testPublishLast() throws InterruptedException {
+        final AtomicInteger count = new AtomicInteger();
+        ConnectableObservable<String> connectable = Observable.create(new OnSubscribeFunc<String>() {
+            @Override
+            public Subscription onSubscribe(final Observer<? super String> observer) {
+                count.incrementAndGet();
+                final BooleanSubscription subscription = new BooleanSubscription();
+                new Thread(new Runnable() {
+                    @Override
+                    public void run() {
+                        observer.onNext("first");
+                        observer.onNext("last");
+                        observer.onCompleted();
+                    }
+                }).start();
+                return subscription;
+            }
+        }).publishLast();
+
+        // subscribe once
+        final CountDownLatch latch = new CountDownLatch(1);
+        connectable.subscribe(new Action1<String>() {
+            @Override
+            public void call(String value) {
+                assertEquals("last", value);
+                latch.countDown();
+            }
+        });
+
+        // subscribe twice
+        connectable.subscribe(new Action1<String>() {
+            @Override
+            public void call(String _) {}
+        });
+
+        Subscription subscription = connectable.connect();
+        assertTrue(latch.await(1000, TimeUnit.MILLISECONDS));
+        assertEquals(1, count.get());
+        subscription.unsubscribe();
+    }
+
     @Test
     public void testReplay() throws InterruptedException {
         final AtomicInteger counter = new AtomicInteger();