1.x: AsyncSubject now supports backpressure

Author: akarnokd

src/main/java/rx/subjects/AsyncSubject.java

@@ -22,6 +22,7 @@
 import rx.exceptions.Exceptions;
 import rx.functions.Action1;
 import rx.internal.operators.NotificationLite;
+import rx.internal.producers.SingleProducer;
 import rx.subjects.SubjectSubscriptionManager.SubjectObserver;
 
 /**
@@ -68,9 +69,13 @@ public static <T> AsyncSubject<T> create() {
             public void call(SubjectObserver<T> o) {
                 Object v = state.getLatest();
                 NotificationLite<T> nl = state.nl;
-                o.accept(v, nl);
-                if (v == null || (!nl.isCompleted(v) && !nl.isError(v))) {
+                if (v == null || nl.isCompleted(v)) {
                     o.onCompleted();
+                } else
+                if (nl.isError(v)) {
+                    o.onError(nl.getError(v));
+                } else {
+                    o.actual.setProducer(new SingleProducer<T>(o.actual, nl.getValue(v)));
                 }
             }
         };
@@ -97,8 +102,7 @@ public void onCompleted() {
                 if (last == nl.completed()) {
                     bo.onCompleted();
                 } else {
-                    bo.onNext(nl.getValue(last));
-                    bo.onCompleted();
+                    bo.actual.setProducer(new SingleProducer<T>(bo.actual, nl.getValue(last)));
                 }
             }
         }

src/main/java/rx/subjects/SubjectSubscriptionManager.java

@@ -203,7 +203,7 @@ public State remove(SubjectObserver o) {
      */
     protected static final class SubjectObserver<T> implements Observer<T> {
         /** The actual Observer. */
-        final Observer<? super T> actual;
+        final Subscriber<? super T> actual;
         /** Was the emitFirst run? Guarded by this. */
         boolean first = true;
         /** Guarded by this. */
@@ -215,7 +215,7 @@ protected static final class SubjectObserver<T> implements Observer<T> {
         protected volatile boolean caughtUp;
         /** Indicate where the observer is at replaying. */
         private volatile Object index;
-        public SubjectObserver(Observer<? super T> actual) {
+        public SubjectObserver(Subscriber<? super T> actual) {
             this.actual = actual;
         }
         @Override

src/test/java/rx/subjects/AsyncSubjectTest.java

@@ -430,4 +430,47 @@ public void testAsyncSubjectValueError() {
         assertNull(async.getValue());
         assertFalse(async.hasValue());
     }
+    
+    @Test
+    public void backpressureOnline() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        
+        AsyncSubject<Integer> subject = AsyncSubject.create();
+        
+        subject.subscribe(ts);
+        
+        subject.onNext(1);
+        subject.onCompleted();
+        
+        ts.assertNoValues();
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+        
+        ts.requestMore(1);
+        
+        ts.assertValue(1);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
+    
+    @Test
+    public void backpressureOffline() {
+        TestSubscriber<Integer> ts = TestSubscriber.create(0);
+        
+        AsyncSubject<Integer> subject = AsyncSubject.create();
+        subject.onNext(1);
+        subject.onCompleted();
+        
+        subject.subscribe(ts);
+        
+        ts.assertNoValues();
+        ts.assertNoErrors();
+        ts.assertNotCompleted();
+        
+        ts.requestMore(1);
+        
+        ts.assertValue(1);
+        ts.assertCompleted();
+        ts.assertNoErrors();
+    }
 }