Merge pull request #10 from akarnokd/UpgradeTo221

Upgrade to RxJava 2.2.1, fixes due to internal changes in Rx

Author: akaita

rxjava2debug/build.gradle

@@ -48,7 +48,7 @@ apply plugin: 'osgi'
 dependencies {
     signature 'org.codehaus.mojo.signature:java16:1.1@signature'
 
-    compile "io.reactivex.rxjava2:rxjava:2.1.0"
+    compile "io.reactivex.rxjava2:rxjava:2.2.1"
 
     testCompile group: 'junit', name: 'junit', version: '4.12'
 

rxjava2debug/src/main/java/hu/akarnokd/rxjava2/debug/FlowableOnAssembly.java

@@ -58,12 +58,12 @@ static final class OnAssemblySubscriber<T> extends BasicFuseableSubscriber<T, T>
 
         @Override
         public void onNext(T t) {
-            actual.onNext(t);
+            downstream.onNext(t);
         }
 
         @Override
         public void onError(Throwable t) {
-            actual.onError(assembled.appendLast(t));
+            downstream.onError(assembled.appendLast(t));
         }
 
         @Override
@@ -95,17 +95,17 @@ static final class OnAssemblyConditionalSubscriber<T> extends BasicFuseableCondi
 
         @Override
         public void onNext(T t) {
-            actual.onNext(t);
+            downstream.onNext(t);
         }
 
         @Override
         public boolean tryOnNext(T t) {
-            return actual.tryOnNext(t);
+            return downstream.tryOnNext(t);
         }
 
         @Override
         public void onError(Throwable t) {
-            actual.onError(assembled.appendLast(t));
+            downstream.onError(assembled.appendLast(t));
         }
 
         @Override

rxjava2debug/src/main/java/hu/akarnokd/rxjava2/debug/ObservableOnAssembly.java

@@ -52,17 +52,17 @@ static final class OnAssemblyObserver<T> extends BasicFuseableObserver<T, T> {
 
         @Override
         public void onNext(T t) {
-            actual.onNext(t);
+            downstream.onNext(t);
         }
 
         @Override
         public void onError(Throwable t) {
-            actual.onError(assembled.appendLast(t));
+            downstream.onError(assembled.appendLast(t));
         }
 
         @Override
         public int requestFusion(int mode) {
-            QueueDisposable<T> qs = this.qs;
+            QueueDisposable<T> qs = this.qd;
             if (qs != null) {
                 int m = qs.requestFusion(mode);
                 sourceMode = m;
@@ -73,7 +73,7 @@ public int requestFusion(int mode) {
 
         @Override
         public T poll() throws Exception {
-            return qs.poll();
+            return qd.poll();
         }
     }
 }