4.x: Introduce Streamable / Streamer after IAsyncEnumerable /

IAsyncEnumreator

Author: akarnokd

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

@@ -0,0 +1,44 @@
+/*
+ * 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.NonNull;
+import io.reactivex.rxjava4.disposables.*;
+
+/**
+ * The {@code IAsyncEnumerable} of the Java world.
+ * Runs best with Virtual Threads.
+ * TODO proper docs
+ * @param <T> the element type of the stream.
+ * @since 4.0.0
+ */
+public abstract class Streamable<@NonNull T> {
+
+    /**
+     * Realizes the stream and returns an interface that let's one consume it.
+     * @param cancellation where to register and listen for cancellation calls.
+     * @return the Streamer instance to consume.
+     */
+    @NonNull
+    public abstract Streamer<T> stream(@NonNull DisposableContainer cancellation);
+
+    /**
+     * Realizes the stream and returns an interface that let's one consume it.
+     * @return the Streamer instance to consume.
+     */
+    @NonNull
+    public final Streamer<T> stream() {
+        return stream(new CompositeDisposable()); // FIXME, use a practically no-op disposable container instead
+    }
+}

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

@@ -0,0 +1,52 @@
+/*
+ * 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 java.util.NoSuchElementException;
+import java.util.concurrent.CompletionStage;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+
+/**
+ * A realized stream which can then be consumed asynchronously in steps.
+ * Think of it as the {@IAsyncEnumerator} of the Java world. Runs best on Virtual Threads.
+ * @param <T> the element type.
+ * TODO proper docs
+ * @since 4.0.0
+ */
+public interface Streamer<@NonNull T> {
+
+    /**
+     * Determine if there are more elements available from the source.
+     * @return eventually true or false, indicating availability or termination
+     */
+    @NonNull
+    CompletionStage<Boolean> next();
+
+    /**
+     * Returns the current element if {@link #next()} yielded {@code true}.
+     * Can be called multiple times between {@link #next()} calls.
+     * @return the current element
+     * @throws NoSuchElementException before the very first {@link #next()} or after {@link #next()} returned {@code false}
+     */
+    @NonNull
+    T current();
+
+    /**
+     * Called when the stream ends or gets cancelled. Should be always invoked.
+     * @return the stage you can await to cleanups to happen
+     */
+    @NonNull
+    CompletionStage<Void> close();
+}

src/main/java/io/reactivex/rxjava4/disposables/Disposable.java

@@ -13,19 +13,19 @@
 
 package io.reactivex.rxjava4.disposables;
 
+import java.util.Objects;
+import java.util.concurrent.Flow.Subscription;
+import java.util.concurrent.Future;
+
 import io.reactivex.rxjava4.annotations.NonNull;
 import io.reactivex.rxjava4.functions.Action;
 import io.reactivex.rxjava4.internal.disposables.EmptyDisposable;
 import io.reactivex.rxjava4.internal.functions.Functions;
-import static java.util.concurrent.Flow.*;
-
-import java.util.Objects;
-import java.util.concurrent.Future;
 
 /**
  * Represents a disposable resource.
  */
-public interface Disposable {
+public interface Disposable extends AutoCloseable {
     /**
      * Dispose the resource, the operation should be idempotent.
      */
@@ -37,6 +37,13 @@ public interface Disposable {
      */
     boolean isDisposed();
 
+    /**
+     * Dispose the resource, the operation should be idempotent.
+     */
+    default void close() {
+        dispose();
+    }
+
     /**
      * Construct a {@code Disposable} by wrapping a {@link Runnable} that is
      * executed exactly once when the {@code Disposable} is disposed.

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

@@ -91,6 +91,7 @@ protected void subscribeActual(Subscriber<? super T> s) {
         }
     }
 
+    @SuppressWarnings("resource")
     void cancel(RefConnection rc) {
         SequentialDisposable sd;
         synchronized (this) {

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

@@ -68,6 +68,7 @@ public void subscribeActual(Observer<? super T> observer) {
             return;
         }
 
+        @SuppressWarnings("resource")
         AmbCoordinator<T> ac = new AmbCoordinator<>(observer, count);
         ac.subscribe(sources);
     }

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

@@ -72,6 +72,7 @@ public void subscribeActual(Observer<? super R> observer) {
             return;
         }
 
+        @SuppressWarnings("resource")
         LatestCoordinator<T, R> lc = new LatestCoordinator<>(observer, combiner, count, bufferSize, delayError);
         lc.subscribe(sources);
     }

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

@@ -88,6 +88,7 @@ protected void subscribeActual(Observer<? super T> observer) {
         }
     }
 
+    @SuppressWarnings("resource")
     void cancel(RefConnection rc) {
         SequentialDisposable sd;
         synchronized (this) {

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

@@ -68,6 +68,7 @@ public void subscribeActual(Observer<? super R> observer) {
             return;
         }
 
+        @SuppressWarnings("resource")
         ZipCoordinator<T, R> zc = new ZipCoordinator<>(observer, zipper, count, delayError);
         zc.subscribe(sources, bufferSize);
     }

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java

@@ -0,0 +1,18 @@
+/*
+ * 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.internal.operators.streamable;
+
+public class StreamableEmpty<T> {
+
+}

src/main/java/io/reactivex/rxjava4/internal/schedulers/ExecutorScheduler.java

@@ -166,6 +166,7 @@ public Disposable schedule(@NonNull Runnable run) {
                 task = interruptibleTask;
                 disposable = interruptibleTask;
             } else {
+                @SuppressWarnings("resource")
                 BooleanRunnable runnableTask = new BooleanRunnable(decoratedRun);
 
                 task = runnableTask;