1.x: fix broken backpressure through unsubscribeOn() (#5726)

Author: akarnokd

src/main/java/rx/internal/operators/OperatorUnsubscribeOn.java

@@ -52,6 +52,10 @@ public void onNext(T t) {
                 subscriber.onNext(t);
             }
 
+            @Override
+            public void setProducer(Producer p) {
+                subscriber.setProducer(p);
+            }
         };
 
         subscriber.add(Subscriptions.create(new Action0() {

src/test/java/rx/internal/operators/OperatorUnsubscribeOnTest.java

@@ -204,4 +204,31 @@ public Thread getThread() {
         }
 
     }
+
+    @Test
+    public void backpressure() {
+        AssertableSubscriber<Integer> as = Observable.range(1, 10)
+        .unsubscribeOn(Schedulers.trampoline())
+        .test(0);
+
+        as.assertNoValues()
+        .assertNoErrors()
+        .assertNotCompleted();
+
+        as.requestMore(1);
+
+        as.assertValue(1)
+        .assertNoErrors()
+        .assertNotCompleted();
+
+        as.requestMore(3);
+
+        as.assertValues(1, 2, 3, 4)
+        .assertNoErrors()
+        .assertNotCompleted();
+
+        as.requestMore(10);
+
+        as.assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+    }
 }