Upgrade to RxJava 2.0.4, +bufferSplit, minor fixes

Author: akarnokd

build.gradle

@@ -49,11 +49,11 @@ 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.3"
+    compile "io.reactivex.rxjava2:rxjava:2.0.4"
 
     testCompile group: 'junit', name: 'junit', version: '4.12'
 
-    testCompile 'org.mockito:mockito-core:2.1.0'
+    testCompile 'org.mockito:mockito-core:2.5.0'
 }
 
 task sourcesJar(type: Jar, dependsOn: classes) {

gradle.properties

@@ -1 +1 @@
-version=0.14.2
+version=0.14.3

src/main/java/hu/akarnokd/rxjava2/operators/FlowableBufferPredicate.java

@@ -40,19 +40,28 @@
  */
 final class FlowableBufferPredicate<T, C extends Collection<? super T>> extends Flowable<C> implements FlowableTransformer<T, C> {
 
+    enum Mode {
+        /** The item triggering the new buffer will be part of the new buffer. */
+        BEFORE,
+        /** The item triggering the new buffer will be part of the old buffer. */
+        AFTER,
+        /** The item won't be part of any buffers. */
+        SPLIT
+    }
+    
     final Publisher<T> source;
 
     final Predicate<? super T> predicate;
 
-    final boolean cutAfter;
+    final Mode mode;
 
     final Callable<C> bufferSupplier;
 
-    FlowableBufferPredicate(Publisher<T> source, Predicate<? super T> predicate, boolean cutAfter,
+    FlowableBufferPredicate(Publisher<T> source, Predicate<? super T> predicate, Mode mode,
             Callable<C> bufferSupplier) {
         this.source = source;
         this.predicate = predicate;
-        this.cutAfter = cutAfter;
+        this.mode = mode;
         this.bufferSupplier = bufferSupplier;
     }
 
@@ -68,12 +77,12 @@ protected void subscribeActual(Subscriber<? super C> s) {
             return;
         }
 
-        source.subscribe(new BufferPredicateSubscriber<T, C>(s, buffer, predicate, cutAfter, bufferSupplier));
+        source.subscribe(new BufferPredicateSubscriber<T, C>(s, buffer, predicate, mode, bufferSupplier));
     }
 
     @Override
     public Publisher<C> apply(Flowable<T> upstream) {
-        return new FlowableBufferPredicate<T, C>(upstream, predicate, cutAfter, bufferSupplier);
+        return new FlowableBufferPredicate<T, C>(upstream, predicate, mode, bufferSupplier);
     }
 
     static final class BufferPredicateSubscriber<T, C extends Collection<? super T>>
@@ -83,7 +92,7 @@ static final class BufferPredicateSubscriber<T, C extends Collection<? super T>>
 
         final Predicate<? super T> predicate;
 
-        final boolean cutAfter;
+        final Mode mode;
 
         final Callable<C> bufferSupplier;
 
@@ -95,11 +104,11 @@ static final class BufferPredicateSubscriber<T, C extends Collection<? super T>>
 
         BufferPredicateSubscriber(Subscriber<? super C> actual,
                 C buffer,
-                Predicate<? super T> predicate, boolean cutAfter,
+                Predicate<? super T> predicate, Mode mode,
                 Callable<C> bufferSupplier) {
             this.actual = actual;
             this.predicate = predicate;
-            this.cutAfter = cutAfter;
+            this.mode = mode;
             this.buffer = buffer;
             this.bufferSupplier = bufferSupplier;
         }
@@ -136,7 +145,8 @@ public boolean tryOnNext(T t) {
                     return true;
                 }
 
-                if (cutAfter) {
+                switch (mode) {
+                case AFTER: {
                     buf.add(t);
                     if (b) {
                         actual.onNext(buf);
@@ -155,7 +165,9 @@ public boolean tryOnNext(T t) {
                         count++;
                         return false;
                     }
-                } else {
+                    break;
+                }
+                case BEFORE: {
                     if (b) {
                         buf.add(t);
                         count++;
@@ -175,6 +187,27 @@ public boolean tryOnNext(T t) {
                         buffer = buf;
                         count = 1;
                     }
+                    break;
+                }
+                default:
+                    if (b) {
+                        actual.onNext(buf);
+
+                        try {
+                            buffer = bufferSupplier.call();
+                        } catch (Throwable ex) {
+                            Exceptions.throwIfFatal(ex);
+                            s.cancel();
+                            onError(ex);
+                            return true;
+                        }
+
+                        count = 0;
+                    } else {
+                        buf.add(t);
+                        count++;
+                        return false;
+                    }
                 }
             }
             return true;

src/main/java/hu/akarnokd/rxjava2/operators/FlowableTransformers.java

@@ -148,14 +148,14 @@ public static <T> FlowableTransformer<T, List<T>> bufferWhile(Predicate<? super
     @SchedulerSupport(SchedulerSupport.NONE)
     @BackpressureSupport(BackpressureKind.FULL)
     public static <T, C extends Collection<? super T>> FlowableTransformer<T, C> bufferWhile(Predicate<? super T> predicate, Callable<C> bufferSupplier) {
-        return new FlowableBufferPredicate<T, C>(null, predicate, false, bufferSupplier);
+        return new FlowableBufferPredicate<T, C>(null, predicate, FlowableBufferPredicate.Mode.BEFORE, bufferSupplier);
     }
 
     /**
      * Buffers elements into a List until the given predicate returns true at which
      * point a new empty buffer is started.
      * @param <T> the source value type
-     * @param predicate the predicate receiving the current itam and if returns true,
+     * @param predicate the predicate receiving the current item and if returns true,
      *                  the current buffer is emitted and a fresh empty buffer is created
      * @return the new FlowableTransformer instance
      *
@@ -173,7 +173,7 @@ public static <T> FlowableTransformer<T, List<T>> bufferUntil(Predicate<? super
      * point a new empty custom collection is started.
      * @param <T> the source value type
      * @param <C> the collection type
-     * @param predicate the predicate receiving the current itam and if returns true,
+     * @param predicate the predicate receiving the current item and if returns true,
      *                  the current collection is emitted and a fresh empty collection is created
      * @param bufferSupplier the callable that returns a fresh collection
      * @return the new Flowable instance
@@ -183,7 +183,42 @@ public static <T> FlowableTransformer<T, List<T>> bufferUntil(Predicate<? super
     @SchedulerSupport(SchedulerSupport.NONE)
     @BackpressureSupport(BackpressureKind.FULL)
     public static <T, C extends Collection<? super T>> FlowableTransformer<T, C> bufferUntil(Predicate<? super T> predicate, Callable<C> bufferSupplier) {
-        return new FlowableBufferPredicate<T, C>(null, predicate, true, bufferSupplier);
+        return new FlowableBufferPredicate<T, C>(null, predicate, FlowableBufferPredicate.Mode.AFTER, bufferSupplier);
+    }
+
+    /**
+     * Buffers elements into a List until the given predicate returns true at which
+     * point a new empty buffer is started; the particular item will be dropped.
+     * @param <T> the source value type
+     * @param predicate the predicate receiving the current item and if returns true,
+     *                  the current buffer is emitted and a fresh empty buffer is created
+     * @return the new FlowableTransformer instance
+     *
+     * @since 0.14.3
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @BackpressureSupport(BackpressureKind.FULL)
+    public static <T> FlowableTransformer<T, List<T>> bufferSplit(Predicate<? super T> predicate) {
+        return bufferSplit(predicate, Functions.<T>createArrayList(16));
+    }
+
+
+    /**
+     * Buffers elements into a custom collection until the given predicate returns true at which
+     * point a new empty custom collection is started; the particular item will be dropped.
+     * @param <T> the source value type
+     * @param <C> the collection type
+     * @param predicate the predicate receiving the current item and if returns true,
+     *                  the current collection is emitted and a fresh empty collection is created
+     * @param bufferSupplier the callable that returns a fresh collection
+     * @return the new Flowable instance
+     *
+     * @since 0.14.3
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @BackpressureSupport(BackpressureKind.FULL)
+    public static <T, C extends Collection<? super T>> FlowableTransformer<T, C> bufferSplit(Predicate<? super T> predicate, Callable<C> bufferSupplier) {
+        return new FlowableBufferPredicate<T, C>(null, predicate, FlowableBufferPredicate.Mode.SPLIT, bufferSupplier);
     }
 
     /**

src/main/java/hu/akarnokd/rxjava2/operators/Flowables.java

@@ -271,7 +271,7 @@ public static <T extends Comparable<? super T>> Flowable<T> orderedMerge(Iterabl
     @SchedulerSupport(SchedulerSupport.NONE)
     public static <T> Flowable<T> repeat(T item) {
         ObjectHelper.requireNonNull(item, "item is null");
-        return new FlowableRepeatScalar<T>(item);
+        return RxJavaPlugins.onAssembly(new FlowableRepeatScalar<T>(item));
     }
 
     /**
@@ -286,6 +286,6 @@ public static <T> Flowable<T> repeat(T item) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public static <T> Flowable<T> repeatCallable(Callable<T> callable) {
         ObjectHelper.requireNonNull(callable, "callable is null");
-        return new FlowableRepeatCallable<T>(callable);
+        return RxJavaPlugins.onAssembly(new FlowableRepeatCallable<T>(callable));
     }
 }

src/test/java/hu/akarnokd/rxjava2/operators/FlowableBufferPredicateTest.java

@@ -332,6 +332,29 @@ public List<Integer> call() throws Exception {
         .assertFailure(IllegalArgumentException.class, Arrays.asList(1, 2));
     }
 
+    @SuppressWarnings("unchecked")
+    @Test
+    public void bufferSupplierCrash3() {
+        Flowable.just(1, 2, -1, 3, 4, 5, -1, -1, 6)
+        .compose(FlowableTransformers.bufferSplit(new Predicate<Integer>() {
+            @Override
+            public boolean test(Integer v) throws Exception {
+                return v == -1;
+            }
+        }, new Callable<List<Integer>>() {
+            int c;
+            @Override
+            public List<Integer> call() throws Exception {
+                if (c++ == 1) {
+                    throw new IllegalArgumentException();
+                }
+                return new ArrayList<Integer>();
+            }
+        }))
+        .test()
+        .assertFailure(IllegalArgumentException.class, Arrays.asList(1, 2));
+    }
+
     @SuppressWarnings("unchecked")
     @Test
     public void doubleError() {
@@ -359,4 +382,70 @@ public boolean test(Integer v) throws Exception {
             RxJavaPlugins.reset();
         }
     }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void splitNormal() {
+        Flowable.just(1, 2, -1, 3, 4, 5, -1, -1, 6)
+        .compose(FlowableTransformers.bufferSplit(new Predicate<Integer>() {
+            @Override
+            public boolean test(Integer v) throws Exception {
+                return v == -1;
+            }
+        }))
+        .test()
+        .assertResult(
+                Arrays.asList(1, 2),
+                Arrays.asList(3, 4, 5),
+                Arrays.<Integer>asList(),
+                Arrays.asList(6)
+        );
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void splitNormalHidden() {
+        Flowable.just(1, 2, -1, 3, 4, 5, -1, -1, 6).hide()
+        .compose(FlowableTransformers.bufferSplit(new Predicate<Integer>() {
+            @Override
+            public boolean test(Integer v) throws Exception {
+                return v == -1;
+            }
+        }))
+        .test()
+        .assertResult(
+                Arrays.asList(1, 2),
+                Arrays.asList(3, 4, 5),
+                Arrays.<Integer>asList(),
+                Arrays.asList(6)
+        );
+    }
+
+    @SuppressWarnings("unchecked")
+    @Test
+    public void splitNormalBackpressured() {
+        Flowable.just(1, 2, -1, 3, 4, 5, -1, -1, 6)
+        .compose(FlowableTransformers.bufferSplit(new Predicate<Integer>() {
+            @Override
+            public boolean test(Integer v) throws Exception {
+                return v == -1;
+            }
+        }))
+        .test(0)
+        .assertNoValues()
+        .requestMore(1)
+        .assertValue(Arrays.asList(1, 2))
+        .requestMore(2)
+        .assertValues(Arrays.asList(1, 2),
+                Arrays.asList(3, 4, 5),
+                Arrays.<Integer>asList())
+        .requestMore(1)
+        .assertResult(
+                Arrays.asList(1, 2),
+                Arrays.asList(3, 4, 5),
+                Arrays.<Integer>asList(),
+                Arrays.asList(6)
+        );
+    }
+
 }