Add FlowableTransformers.coalesce

akarnokd · akarnokd · commit 65e242bbab02 · 2017-07-14T14:01:42.000+02:00
diff --git a/README.md b/README.md
@@ -45,7 +45,7 @@ Maven search:
     - [debounceFirst()](#flowabletransformersdebouncefirst), [switchFlatMap()](#flowabletransformersswitchflatmap), [flatMapSync()](#flowabletransformersflatmapsync),
     - [flatMapAsync()](#flowabletransformersflatmapasync), [switchIfEmpty()](#flowabletransformersswitchifempty--switchifemptyarray),
     - [expand()](#flowabletransformersexpand), [mapAsync()](#flowabletransformersmapasync), [filterAsync()](#flowabletransformerfilterasync),
-    - [refCount()](#flowabletransformersrefcount), [zipLatest()](#flowablesziplatest)
+    - [refCount()](#flowabletransformersrefcount), [zipLatest()](#flowablesziplatest), [coalesce()](#flowabletransformerscoalesce)
   - [Custom parallel operators and transformers](#custom-parallel-operators-and-transformers)
     - [sumX()](#paralleltransformerssumx)
   - [Special Publisher implementations](#special-publisher-implementations)
@@ -1083,7 +1083,22 @@ scheduler.advanceTimeBy(200, TimeUnit.MILLISECONDS);
 ts.assertResult("[2, 4]", "[4, 5]", "[6, 6]");
 ```
 
+### FlowableTransformers.coalesce()
 
+Coalesces items from upstream into a container via a consumer and emits the container if
+there is a downstream demand, otherwise it keeps coalescing into the same container. Note
+that the operator keeps an internal unbounded buffer to collect up upstream values before
+the coalescing happens and thus a computational heavy upstream hogging the emission thread
+may lead to excessive memory usage. It is recommended to use `observeOn` in this case.
+
+```java
+Flowable.range(1, 5)
+.compose(FlowableTransformers.coalesce(ArrayList::new, (a, b) -> a.add(b))
+.test(1)
+.assertValue(Arrays.asList(1))
+.requestMore(1)
+.assertResult(Arrays.asList(1), Arrays.asList(2, 3, 4, 5));
+```
 
 ## Custom parallel operators and transformers
 
diff --git a/src/main/java/hu/akarnokd/rxjava2/operators/FlowableCoalesce.java b/src/main/java/hu/akarnokd/rxjava2/operators/FlowableCoalesce.java
@@ -0,0 +1,260 @@
+/*
+ * Copyright 2016-2017 David Karnok
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package hu.akarnokd.rxjava2.operators;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.*;
+
+import org.reactivestreams.*;
+
+import io.reactivex.*;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.BiConsumer;
+import io.reactivex.internal.fuseable.SimplePlainQueue;
+import io.reactivex.internal.queue.SpscLinkedArrayQueue;
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.internal.util.BackpressureHelper;
+
+/**
+ * Coalesces items into a container if the downstream is not ready to receive items.
+ *
+ * @param <T> the upstream element type
+ * @param <R> the container type emitted to downstream
+ *
+ * @since 0.17.3
+ */
+final class FlowableCoalesce<T, R> extends Flowable<R> implements FlowableTransformer<T, R> {
+
+    final Publisher<T> source;
+
+    final Callable<R> containerSupplier;
+
+    final BiConsumer<R, T> coalescer;
+
+    final int bufferSize;
+
+    FlowableCoalesce(Publisher<T> source, Callable<R> containerSupplier, BiConsumer<R, T> coalescer, int bufferSize) {
+        this.source = source;
+        this.containerSupplier = containerSupplier;
+        this.coalescer = coalescer;
+        this.bufferSize = bufferSize;
+    }
+
+    @Override
+    public Publisher<R> apply(Flowable<T> upstream) {
+        return new FlowableCoalesce<T, R>(upstream, containerSupplier, coalescer, bufferSize);
+    }
+
+    @Override
+    protected void subscribeActual(Subscriber<? super R> s) {
+        source.subscribe(new CoalesceSubscriber<T, R>(s, containerSupplier, coalescer, bufferSize));
+    }
+
+    static final class CoalesceSubscriber<T, R> extends AtomicInteger
+    implements FlowableSubscriber<T>, Subscription {
+
+        private static final long serialVersionUID = -6157179110480235565L;
+
+        final Subscriber<? super R> actual;
+
+        final Callable<R> containerSupplier;
+
+        final BiConsumer<R, T> coalescer;
+
+        final AtomicLong requested;
+
+        final int bufferSize;
+
+        volatile SimplePlainQueue<T> queue;
+
+        Subscription upstream;
+
+        R container;
+
+        volatile boolean done;
+
+        volatile boolean cancelled;
+
+        Throwable error;
+
+        long emitted;
+
+        CoalesceSubscriber(Subscriber<? super R> actual, Callable<R> containerSupplier,
+                BiConsumer<R, T> coalescer, int bufferSize) {
+            this.actual = actual;
+            this.containerSupplier = containerSupplier;
+            this.coalescer = coalescer;
+            this.requested = new AtomicLong();
+            this.bufferSize = bufferSize;
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            if (SubscriptionHelper.validate(upstream, s)) {
+                upstream = s;
+                actual.onSubscribe(this);
+
+                s.request(Long.MAX_VALUE);
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            if (get() == 0 && compareAndSet(0, 1)) {
+                SimplePlainQueue<T> q = queue;
+                if (q == null || q.isEmpty()) {
+                    R c = container;
+                    try {
+                        if (c == null) {
+                            c = containerSupplier.call();
+                            container = c;
+                        }
+                        coalescer.accept(c, t);
+                    } catch (Throwable ex) {
+                        Exceptions.throwIfFatal(ex);
+                        upstream.cancel();
+                        container = null;
+                        actual.onError(ex);
+                        return;
+                    }
+                    long r = requested.get();
+                    long e = emitted;
+                    if (e != r) {
+                        container = null;
+                        actual.onNext(c);
+                        emitted = e + 1;
+                    }
+                    if (decrementAndGet() == 0) {
+                        return;
+                    }
+                }
+            } else {
+                SimplePlainQueue<T> q = queue;
+                if (q == null) {
+                    q = new SpscLinkedArrayQueue<T>(bufferSize);
+                    queue = q;
+                }
+                q.offer(t);
+                if (getAndIncrement() != 0) {
+                    return;
+                }
+            }
+            drainLoop();
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            error = t;
+            done = true;
+            drain();
+        }
+
+        @Override
+        public void onComplete() {
+            done = true;
+            drain();
+        }
+
+        @Override
+        public void request(long n) {
+            if (SubscriptionHelper.validate(n)) {
+                BackpressureHelper.add(requested, n);
+                drain();
+            }
+        }
+
+        @Override
+        public void cancel() {
+            cancelled = true;
+            upstream.cancel();
+            if (getAndIncrement() == 0) {
+                container = null;
+            }
+        }
+
+        void drain() {
+            if (getAndIncrement() != 0) {
+                return;
+            }
+
+            drainLoop();
+        }
+
+        void drainLoop() {
+            int missed = 1;
+            long e = emitted;
+            R c = container;
+            Subscriber<? super R> a = actual;
+
+            for (;;) {
+                if (cancelled) {
+                    container = null;
+                    return;
+                }
+                boolean d = done;
+                SimplePlainQueue<T> q = queue;
+                boolean empty = q == null || q.isEmpty();
+
+                if (!empty) {
+                    try {
+                        if (c == null) {
+                            c = containerSupplier.call();
+                            container = c;
+                        }
+
+                        for (;;) {
+                            T v = q.poll();
+                            if (v == null) {
+                                break;
+                            }
+                            coalescer.accept(c, v);
+                        }
+                    } catch (Throwable ex) {
+                        Exceptions.throwIfFatal(ex);
+                        container = null;
+                        a.onError(ex);
+                        return;
+                    }
+                }
+
+                if (c != null && e != requested.get()) {
+                    a.onNext(c);
+                    c = null;
+                    container = null;
+                    e++;
+                }
+
+                if (d && c == null) {
+                    Throwable ex = error;
+                    container = null;
+                    if (ex != null) {
+                        a.onError(ex);
+                    } else {
+                        a.onComplete();
+                    }
+                    return;
+                }
+
+                emitted = e;
+                missed = addAndGet(-missed);
+                if (missed == 0) {
+                    break;
+                }
+            }
+        }
+    }
+}
diff --git a/src/main/java/hu/akarnokd/rxjava2/operators/FlowableTransformers.java b/src/main/java/hu/akarnokd/rxjava2/operators/FlowableTransformers.java
@@ -1123,4 +1123,36 @@ public static <T> FlowableTransformer<T, T> refCount(int subscriberCount, long t
         ObjectHelper.requireNonNull(scheduler, "scheduler is null");
         return new FlowableRefCountTimeout<T>(null, subscriberCount, timeout, unit, scheduler);
     }
+
+    /**
+     * Coalesces items from upstream into a container via a consumer and emits the container if
+     * there is a downstream demand, otherwise it keeps coalescing into the same container.
+     * @param <T> the upstream value type
+     * @param <R> the container and result type
+     * @param containerSupplier the function called and should return a fresh container to coalesce into
+     * @param coalescer the consumer receiving the current container and upstream item to handle
+     * @return the new FlowableTransformer instance
+     * @since 0.17.3
+     */
+    public static <T, R> FlowableTransformer<T, R> coalesce(Callable<R> containerSupplier, BiConsumer<R, T> coalescer) {
+        return coalesce(containerSupplier, coalescer, Flowable.bufferSize());
+    }
+
+    /**
+     * Coalesces items from upstream into a container via a consumer and emits the container if
+     * there is a downstream demand, otherwise it keeps coalescing into the same container.
+     * @param <T> the upstream value type
+     * @param <R> the container and result type
+     * @param containerSupplier the function called and should return a fresh container to coalesce into
+     * @param coalescer the consumer receiving the current container and upstream item to handle
+     * @param bufferSize the island size of the internal unbounded buffer
+     * @return the new FlowableTransformer instance
+     * @since 0.17.3
+     */
+    public static <T, R> FlowableTransformer<T, R> coalesce(Callable<R> containerSupplier, BiConsumer<R, T> coalescer, int bufferSize) {
+        ObjectHelper.requireNonNull(containerSupplier, "containerSupplier is null");
+        ObjectHelper.requireNonNull(coalescer, "coalescer is null");
+        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
+        return new FlowableCoalesce<T, R>(null, containerSupplier, coalescer, bufferSize);
+    }
 }
diff --git a/src/test/java/hu/akarnokd/rxjava2/operators/FlowableCoalesceTest.java b/src/test/java/hu/akarnokd/rxjava2/operators/FlowableCoalesceTest.java
