4.x: Streamable + takeUntil + groupBy + refactor + helpers

gradle/javadoc_cleanup.gradle

@@ -6,8 +6,8 @@ task javadocCleanup(dependsOn: "javadoc") doLast {
     fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/Maybe.html'))
     fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/Completable.html'))
 
-    fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/flowables/ConnectableFlowable.html'))
-    fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/observables/ConnectableObservable.html'))
+    fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/ConnectableFlowable.html'))
+    fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/ConnectableObservable.html'))
 
     fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/subjects/ReplaySubject.html'))
     fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/processors/ReplayProcessor.html'))

src/jmh/java/io/reactivex/rxjava4/parallel/ParallelPerf.java

@@ -22,7 +22,6 @@
 
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.core.config.StandardConcurrentBufferedConfig;
-import io.reactivex.rxjava4.flowables.GroupedFlowable;
 import io.reactivex.rxjava4.functions.Function;
 import io.reactivex.rxjava4.schedulers.Schedulers;
 

src/main/java/io/reactivex/rxjava4/flowables/ConnectableFlowable.java → src/main/java/io/reactivex/rxjava4/core/ConnectableFlowable.java

@@ -11,15 +11,13 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.rxjava4.flowables;
+package io.reactivex.rxjava4.core;
 
 import java.util.Objects;
+import java.util.concurrent.Flow.*;
 import java.util.concurrent.TimeUnit;
 
-import static java.util.concurrent.Flow.*;
-
 import io.reactivex.rxjava4.annotations.*;
-import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.disposables.Disposable;
 import io.reactivex.rxjava4.functions.Consumer;
 import io.reactivex.rxjava4.internal.functions.*;

src/main/java/io/reactivex/rxjava4/observables/ConnectableObservable.java → src/main/java/io/reactivex/rxjava4/core/ConnectableObservable.java

@@ -11,13 +11,12 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.rxjava4.observables;
+package io.reactivex.rxjava4.core;
 
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
 import io.reactivex.rxjava4.annotations.*;
-import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.disposables.Disposable;
 import io.reactivex.rxjava4.functions.Consumer;
 import io.reactivex.rxjava4.internal.functions.*;

src/main/java/io/reactivex/rxjava4/core/Flowable.java

@@ -22,7 +22,6 @@
 import io.reactivex.rxjava4.core.config.*;
 import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.exceptions.*;
-import io.reactivex.rxjava4.flowables.*;
 import io.reactivex.rxjava4.functions.*;
 import io.reactivex.rxjava4.internal.functions.*;
 import io.reactivex.rxjava4.internal.jdk8.*;

src/main/java/io/reactivex/rxjava4/flowables/GroupedFlowable.java → src/main/java/io/reactivex/rxjava4/core/GroupedFlowable.java

@@ -11,10 +11,9 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.rxjava4.flowables;
+package io.reactivex.rxjava4.core;
 
 import io.reactivex.rxjava4.annotations.Nullable;
-import io.reactivex.rxjava4.core.Flowable;
 
 /**
  * A {@link Flowable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}.

src/main/java/io/reactivex/rxjava4/observables/GroupedObservable.java → src/main/java/io/reactivex/rxjava4/core/GroupedObservable.java

@@ -11,10 +11,9 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivex.rxjava4.observables;
+package io.reactivex.rxjava4.core;
 
 import io.reactivex.rxjava4.annotations.Nullable;
-import io.reactivex.rxjava4.core.Observable;
 
 /**
  * An {@link Observable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}.

src/main/java/io/reactivex/rxjava4/core/GroupedStreamable.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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 io.reactivex.rxjava4.core;
+
+import io.reactivex.rxjava4.annotations.*;
+
+/**
+ * An {@link Streamable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}.
+ * @param <K>
+ *            the type of the key, can be null
+ * @param <T>
+ *            the type of the items emitted by the {@code GroupedStreamable}
+ * @see Streamable#groupBy(io.reactivex.rxjava4.functions.Function)
+ * @see <a href="http://reactivex.io/documentation/operators/groupby.html">ReactiveX documentation: GroupBy</a>
+ * @since 4.0.0
+ */
+public abstract class GroupedStreamable<@Nullable K, @NonNull T> implements Streamable<T> {
+
+    final K key;
+
+    /**
+     * Constructs a GroupedStreamable with the given key.
+     * @param key the key
+     */
+    protected GroupedStreamable(@Nullable K key) {
+        this.key = key;
+    }
+
+    /**
+     * Returns the key that identifies the group of items emitted by this {@code GroupedStreamable}.
+     *
+     * @return the key that the items emitted by this {@code GroupedStreamable} were grouped by
+     */
+    @Nullable
+    public K getKey() {
+        return key;
+    }
+
+}

src/main/java/io/reactivex/rxjava4/core/Observable.java

@@ -32,7 +32,6 @@
 import io.reactivex.rxjava4.internal.operators.observable.*;
 import io.reactivex.rxjava4.internal.operators.single.SingleToObservable;
 import io.reactivex.rxjava4.internal.util.*;
-import io.reactivex.rxjava4.observables.*;
 import io.reactivex.rxjava4.observers.*;
 import io.reactivex.rxjava4.operators.ScalarSupplier;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;

src/main/java/io/reactivex/rxjava4/core/Streamable.java

@@ -572,6 +572,20 @@ default <R> Streamable<R> flatMap(@NonNull Function<? super T, ? extends Streama
         return RxJavaPlugins.onAssembly(new StreamableFlatMap<>(this, mapper, config.maxConcurrency()));
     }
 
+    /**
+     * Maps each upstream item into a {@code GroupedStreamable} group, emits those groups and keeps
+     * relaying the upstream items into those groups.
+     * @param <K> the key type, {@code null}s allowed
+     * @param keySelector the function that receives the upstream item and returns a key that determines
+     *                    which group the item will go into
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code keySelector} is {@code null}
+     */
+    default <@Nullable K> Streamable<GroupedStreamable<K, T>> groupBy(Function<? super T, ? extends K> keySelector) {
+        Objects.requireNonNull(keySelector, "keySelector is null");
+        return RxJavaPlugins.onAssembly(new StreamableGroupBy<>(this, keySelector));
+    }
+
     /**
      * Hides the identity of this {@code Streamable} and its {@link Streamer}.
      * <p>
@@ -688,6 +702,21 @@ default Streamable<T> takeWhile(@NonNull Predicate<? super T> predicate) {
         return RxJavaPlugins.onAssembly(new StreamableTakeWhile<>(this, predicate));
     }
 
+    /**
+     * Relays items from this {@code Streamable} until the other {@code Streamable} signals
+     * an item or completes.
+     * @param <U> the element type of the other {@code Streamable}
+     * @param other the {@code Streamable} expected to signal when to stop taking items from this {@code Streamable}
+     * @return the new {@code Streamable} instance
+     * @throws NullPointerException if {@code other} is {@code null}
+     */
+    @CheckReturnValue
+    @NonNull
+    default <U> Streamable<T> takeUntil(@NonNull Streamable<U> other) {
+        Objects.requireNonNull(other, "other is null");
+        return RxJavaPlugins.onAssembly(new StreamableTakeUntil<>(this, other));
+    }
+
     /**
      * Calls the specified converter function during assembly time and returns its resulting value.
      * <p>

src/main/java/io/reactivex/rxjava4/core/Streamer.java

@@ -39,7 +39,7 @@ public interface Streamer<@NonNull T> {
      * <ul>
      * <li>`true` indicates there is an item available for consumption via {@link #current()}
      * <li>`false` indicates there are no more items available
-     *< li>`Throwable` indicates there was an upstream error
+     * <li>`Throwable` indicates there was an upstream error
      * </ul>
      */
     @NonNull
@@ -101,17 +101,17 @@ default void awaitFinish() {
      * Use this constant in {@link #next()} to indicate
      * the next value is available, synchronously.
      */
-    CompletionStage<Boolean> NEXT_TRUE = CompletableFuture.completedStage(true);
+    CompletableFuture<Boolean> NEXT_TRUE = CompletableFuture.completedFuture(true);
 
     /**
      * Use this constant in {@link #next()} to indicate
      * no more values will be available, synchronously.
      */
-    CompletionStage<Boolean> NEXT_FALSE = CompletableFuture.completedStage(false);
+    CompletableFuture<Boolean> NEXT_FALSE = CompletableFuture.completedFuture(false);
 
     /**
      * Use this constant in {@link #finish()} to indicate
      * the cleanup was done synchronously.
      */
-    CompletionStage<Void> FINISHED = CompletableFuture.completedStage(null);
+    CompletableFuture<Void> FINISHED = CompletableFuture.completedFuture(null);
 }

src/main/java/io/reactivex/rxjava4/core/package-info.java

@@ -48,6 +48,14 @@
  * consumers to subscribe to them and receive events.</p>
  * <p>Usage examples can be found on the {@link io.reactivex.rxjava4.core.Flowable}/{@link io.reactivex.rxjava4.core.Observable}
  * and {@link java.util.concurrent.Flow.Subscriber} classes.</p>
+ * <p>
+ * Classes supporting the Flowable base reactive class:
+ * {@link io.reactivex.rxjava4.core.ConnectableFlowable} and
+ * {@link io.reactivex.rxjava4.core.GroupedFlowable}.
+ * <p>
+ * Classes supporting the Observable base reactive class:
+ * {@link io.reactivex.rxjava4.core.ConnectableObservable} and
+ * {@link io.reactivex.rxjava4.core.GroupedObservable}.
  */
 package io.reactivex.rxjava4.core;
 

src/main/java/io/reactivex/rxjava4/exceptions/ThrowableWrapper.java

@@ -14,6 +14,7 @@
 package io.reactivex.rxjava4.exceptions;
 
 import java.io.Serial;
+import java.util.concurrent.CompletionException;
 
 /**
  * A runtime exception to sneak around checked exceptions.
@@ -22,7 +23,7 @@
  * at the right place.
  * @since 4.0.0
  */
-public final class ThrowableWrapper extends RuntimeException {
+public final class ThrowableWrapper extends CompletionException {
 
     @Serial
     private static final long serialVersionUID = -5280780582536857320L;

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableAutoConnect.java

@@ -13,13 +13,11 @@
 
 package io.reactivex.rxjava4.internal.operators.flowable;
 
+import java.util.concurrent.Flow.Subscriber;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import static java.util.concurrent.Flow.*;
-
-import io.reactivex.rxjava4.core.Flowable;
+import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.disposables.Disposable;
-import io.reactivex.rxjava4.flowables.ConnectableFlowable;
 import io.reactivex.rxjava4.functions.Consumer;
 
 /**

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableGroupBy.java

@@ -16,14 +16,12 @@
 import java.io.Serial;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.Flow.*;
 import java.util.concurrent.atomic.*;
 
-import static java.util.concurrent.Flow.*;
-
 import io.reactivex.rxjava4.annotations.Nullable;
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.exceptions.*;
-import io.reactivex.rxjava4.flowables.GroupedFlowable;
 import io.reactivex.rxjava4.functions.*;
 import io.reactivex.rxjava4.internal.subscriptions.*;
 import io.reactivex.rxjava4.internal.util.*;

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableInternalHelper.java

@@ -14,14 +14,12 @@
 package io.reactivex.rxjava4.internal.operators.flowable;
 
 import java.util.Objects;
+import java.util.concurrent.Flow.*;
 import java.util.concurrent.TimeUnit;
 
-import static java.util.concurrent.Flow.*;
-
 import io.reactivex.rxjava4.core.*;
-import io.reactivex.rxjava4.flowables.ConnectableFlowable;
 import io.reactivex.rxjava4.functions.*;
-import io.reactivex.rxjava4.internal.functions.*;
+import io.reactivex.rxjava4.internal.functions.Functions;
 
 /**
  * Helper utility class to support Flowable with inner classes.

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowablePublish.java

@@ -14,21 +14,17 @@
 package io.reactivex.rxjava4.internal.operators.flowable;
 
 import java.io.Serial;
+import java.util.concurrent.Flow.*;
 import java.util.concurrent.atomic.*;
 
-import static java.util.concurrent.Flow.*;
-
-import io.reactivex.rxjava4.core.FlowableSubscriber;
+import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.disposables.Disposable;
 import io.reactivex.rxjava4.exceptions.*;
-import io.reactivex.rxjava4.flowables.ConnectableFlowable;
 import io.reactivex.rxjava4.functions.Consumer;
-import io.reactivex.rxjava4.internal.fuseable.*;
+import io.reactivex.rxjava4.internal.fuseable.HasUpstreamPublisher;
 import io.reactivex.rxjava4.internal.subscriptions.SubscriptionHelper;
 import io.reactivex.rxjava4.internal.util.*;
-import io.reactivex.rxjava4.operators.QueueSubscription;
-import io.reactivex.rxjava4.operators.SimpleQueue;
-import io.reactivex.rxjava4.operators.SpscArrayQueue;
+import io.reactivex.rxjava4.operators.*;
 import io.reactivex.rxjava4.plugins.RxJavaPlugins;
 
 /**

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableRefCount.java

@@ -14,14 +14,12 @@
 package io.reactivex.rxjava4.internal.operators.flowable;
 
 import java.io.Serial;
+import java.util.concurrent.Flow.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.*;
 
-import static java.util.concurrent.Flow.*;
-
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.disposables.Disposable;
-import io.reactivex.rxjava4.flowables.ConnectableFlowable;
 import io.reactivex.rxjava4.functions.Consumer;
 import io.reactivex.rxjava4.internal.disposables.*;
 import io.reactivex.rxjava4.internal.subscriptions.SubscriptionHelper;

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableReplay.java

@@ -15,15 +15,13 @@
 
 import java.io.Serial;
 import java.util.*;
+import java.util.concurrent.Flow.*;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.*;
 
-import static java.util.concurrent.Flow.*;
-
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.disposables.Disposable;
 import io.reactivex.rxjava4.exceptions.Exceptions;
-import io.reactivex.rxjava4.flowables.ConnectableFlowable;
 import io.reactivex.rxjava4.functions.*;
 import io.reactivex.rxjava4.internal.fuseable.HasUpstreamPublisher;
 import io.reactivex.rxjava4.internal.subscribers.SubscriberResourceWrapper;

src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableAutoConnect.java

@@ -18,7 +18,6 @@
 import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.disposables.Disposable;
 import io.reactivex.rxjava4.functions.Consumer;
-import io.reactivex.rxjava4.observables.ConnectableObservable;
 
 /**
  * Wraps a ConnectableObservable and calls its connect() method once

src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableGroupBy.java

@@ -18,13 +18,12 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.*;
 
-import io.reactivex.rxjava4.core.ObservableSource;
+import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.core.Observer;
 import io.reactivex.rxjava4.disposables.Disposable;
 import io.reactivex.rxjava4.exceptions.Exceptions;
 import io.reactivex.rxjava4.functions.Function;
 import io.reactivex.rxjava4.internal.disposables.*;
-import io.reactivex.rxjava4.observables.GroupedObservable;
 import io.reactivex.rxjava4.operators.SpscLinkedArrayQueue;
 
 public final class ObservableGroupBy<T, K, V> extends AbstractObservableWithUpstream<T, GroupedObservable<K, V>> {