fix use of rxjava internals

akarnokd · Oct 29, 2016 · 013227b · 013227b
1 parent d840fb7
commit 013227b
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 2 deletions.
diff --git a/src/main/java/hu/akarnokd/rxjava2/async/AsyncFlowable.java b/src/main/java/hu/akarnokd/rxjava2/async/AsyncFlowable.java
@@ -1764,7 +1764,7 @@ public void accept(Subscription s) throws Exception {
         });
         d.lazySet(ls);
 
-        RxJavaPlugins.onAssembly(new FlowableSubscribeOn<T>((Publisher<T>)source, scheduler)).subscribe(ls);
+        RxJavaPlugins.onAssembly(new FlowableSubscribeOn<T>((Publisher<T>)source, scheduler, false)).subscribe(ls);
 
         return f;
     }

diff --git a/src/main/java/hu/akarnokd/rxjava2/debug/FlowableOnAssembly.java b/src/main/java/hu/akarnokd/rxjava2/debug/FlowableOnAssembly.java
@@ -110,7 +110,13 @@ public void onError(Throwable t) {
 
         @Override
         public int requestFusion(int mode) {
-            return transitiveFusion(mode);
+            QueueSubscription<T> qs = this.qs;
+            if (qs != null) {
+                int m = qs.requestFusion(mode);
+                sourceMode = m;
+                return m;
+            }
+            return NONE;
         }
 
         @Override