Upgrade to RxJava 2.0.7 and fix internal uses

Author: akarnokd

build.gradle

@@ -49,7 +49,7 @@ dependencies {
     signature 'org.codehaus.mojo.signature:java16:1.1@signature'
 
     compile "org.reactivestreams:reactive-streams:1.0.0"
-    compile "io.reactivex.rxjava2:rxjava:2.0.6"
+    compile "io.reactivex.rxjava2:rxjava:2.0.7"
 
     testCompile group: 'junit', name: 'junit', version: '4.12'
 

src/main/java/hu/akarnokd/rxjava2/async/AsyncFlowable.java

@@ -27,7 +27,6 @@
 import io.reactivex.functions.*;
 import io.reactivex.internal.disposables.SequentialDisposable;
 import io.reactivex.internal.functions.Functions;
-import io.reactivex.internal.operators.flowable.FlowableSubscribeOn;
 import io.reactivex.internal.subscribers.LambdaSubscriber;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.*;
@@ -1718,7 +1717,6 @@ public static <T> Future<Object> forEachFuture(
      * @return the Future representing the entire for-each operation
      * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-foreachfuture">RxJava Wiki: forEachFuture()</a>
      */
-    @SuppressWarnings("unchecked")
     public static <T> Future<Object> forEachFuture(
             Publisher<? extends T> source,
             Consumer<? super T> onNext,
@@ -1762,7 +1760,7 @@ public void accept(Subscription s) throws Exception {
         });
         d.lazySet(ls);
 
-        RxJavaPlugins.onAssembly(new FlowableSubscribeOn<T>((Publisher<T>)source, scheduler, false)).subscribe(ls);
+        Flowable.fromPublisher(source).subscribeOn(scheduler).subscribe(ls);
 
         return f;
     }

src/main/java/hu/akarnokd/rxjava2/basetypes/Perhaps.java

@@ -25,7 +25,6 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
-import io.reactivex.internal.operators.flowable.*;
 import io.reactivex.internal.subscribers.LambdaSubscriber;
 import io.reactivex.internal.util.ExceptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -933,7 +932,7 @@ public final Perhaps<T> switchIfEmpty(Perhaps<? extends T> other) {
      * @return the new Flowable instance
      */
     public final Flowable<T> repeat() {
-        return RxJavaPlugins.onAssembly(new FlowableRepeat<T>(this, Long.MAX_VALUE));
+        return Flowable.fromPublisher(this).repeat();
     }
 
     /**
@@ -942,10 +941,7 @@ public final Flowable<T> repeat() {
      * @return the new Flowable instance
      */
     public final Flowable<T> repeat(long times) {
-        if (times < 0L) {
-            throw new IllegalArgumentException("times >= 0 required but it was " + times);
-        }
-        return RxJavaPlugins.onAssembly(new FlowableRepeat<T>(this, times));
+        return Flowable.fromPublisher(this).repeat(times);
     }
 
     /**
@@ -956,8 +952,7 @@ public final Flowable<T> repeat(long times) {
      * @return the new Flowable instance
      */
     public final Flowable<T> repeat(BooleanSupplier stop) {
-        ObjectHelper.requireNonNull(stop, "stop is null");
-        return RxJavaPlugins.onAssembly(new FlowableRepeatUntil<T>(this, stop));
+        return Flowable.fromPublisher(this).repeatUntil(stop);
     }
 
     /**
@@ -969,8 +964,7 @@ public final Flowable<T> repeat(BooleanSupplier stop) {
      * @return the new Flowable instance
      */
     public final Flowable<T> repeatWhen(Function<? super Flowable<Object>, ? extends Publisher<?>> handler) {
-        ObjectHelper.requireNonNull(handler, "handler is null");
-        return RxJavaPlugins.onAssembly(new FlowableRepeatWhen<T>(this, handler));
+        return Flowable.fromPublisher(this).repeatWhen(handler);
     }
 
     /**

src/main/java/hu/akarnokd/rxjava2/basetypes/Solo.java

@@ -25,7 +25,6 @@
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.*;
 import io.reactivex.internal.functions.*;
-import io.reactivex.internal.operators.flowable.*;
 import io.reactivex.internal.subscribers.LambdaSubscriber;
 import io.reactivex.internal.util.ExceptionHelper;
 import io.reactivex.plugins.RxJavaPlugins;
@@ -1179,7 +1178,7 @@ public final <R> Solo<R> lift(Function<Subscriber<? super R>, Subscriber<? super
      * @return the new Flowable instance
      */
     public final Flowable<T> repeat() {
-        return RxJavaPlugins.onAssembly(new FlowableRepeat<T>(this, Long.MAX_VALUE));
+        return Flowable.fromPublisher(this).repeat();
     }
 
     /**
@@ -1188,10 +1187,7 @@ public final Flowable<T> repeat() {
      * @return the new Flowable instance
      */
     public final Flowable<T> repeat(long times) {
-        if (times < 0L) {
-            throw new IllegalArgumentException("times >= 0 required but it was " + times);
-        }
-        return RxJavaPlugins.onAssembly(new FlowableRepeat<T>(this, times));
+        return Flowable.fromPublisher(this).repeat(times);
     }
 
     /**
@@ -1202,8 +1198,7 @@ public final Flowable<T> repeat(long times) {
      * @return the new Flowable instance
      */
     public final Flowable<T> repeat(BooleanSupplier stop) {
-        ObjectHelper.requireNonNull(stop, "stop is null");
-        return RxJavaPlugins.onAssembly(new FlowableRepeatUntil<T>(this, stop));
+        return Flowable.fromPublisher(this).repeatUntil(stop);
     }
 
     /**
@@ -1215,8 +1210,7 @@ public final Flowable<T> repeat(BooleanSupplier stop) {
      * @return the new Flowable instance
      */
     public final Flowable<T> repeatWhen(Function<? super Flowable<Object>, ? extends Publisher<?>> handler) {
-        ObjectHelper.requireNonNull(handler, "handler is null");
-        return RxJavaPlugins.onAssembly(new FlowableRepeatWhen<T>(this, handler));
+        return Flowable.fromPublisher(this).repeatWhen(handler);
     }
 
     /**

src/main/java/hu/akarnokd/rxjava2/parallel/ParallelConcatMap.java

@@ -66,24 +66,8 @@ public void subscribe(Subscriber<? super R>[] subscribers) {
         @SuppressWarnings("unchecked")
         final Subscriber<T>[] parents = new Subscriber[n];
 
-        // FIXME cheat until we have support from RxJava2 internals
-        Publisher<T> p = new Publisher<T>() {
-            int i;
-
-            @SuppressWarnings("unchecked")
-            @Override
-            public void subscribe(Subscriber<? super T> s) {
-                parents[i++] = (Subscriber<T>)s;
-            }
-        };
-
-        FlowableConcatMap<T, R> op = new FlowableConcatMap<T, R>(p, mapper, prefetch, errorMode);
-
         for (int i = 0; i < n; i++) {
-
-            op.subscribe(subscribers[i]);
-// FIXME needs a FlatMap subscriber
-//            parents[i] = FlowableConcatMap.createSubscriber(s, mapper, prefetch, errorMode);
+            parents[i] = FlowableConcatMap.subscribe(subscribers[i], mapper, prefetch, errorMode);
         }
 
         source.subscribe(parents);

src/main/java/hu/akarnokd/rxjava2/parallel/ParallelFlatMap.java

@@ -70,24 +70,8 @@ public void subscribe(Subscriber<? super R>[] subscribers) {
         @SuppressWarnings("unchecked")
         final Subscriber<T>[] parents = new Subscriber[n];
 
-        // FIXME cheat until we have support from RxJava2 internals
-        Publisher<T> p = new Publisher<T>() {
-            int i;
-
-            @SuppressWarnings("unchecked")
-            @Override
-            public void subscribe(Subscriber<? super T> s) {
-                parents[i++] = (Subscriber<T>)s;
-            }
-        };
-
-        FlowableFlatMap<T, R> op = new FlowableFlatMap<T, R>(p, mapper, delayError, maxConcurrency, prefetch);
-
         for (int i = 0; i < n; i++) {
-
-            op.subscribe(subscribers[i]);
-// FIXME needs a FlatMap subscriber
-//            parents[i] = FlowableFlatMap.createSubscriber(s, mapper, delayError, maxConcurrency, prefetch);
+            parents[i] = FlowableFlatMap.subscribe(subscribers[i], mapper, delayError, maxConcurrency, prefetch);
         }
 
         source.subscribe(parents);