4.x: Fix Streamable.fromStream leaking the Stream when the first hasNext fails (#8331)

StreamableFromStream.stream() called items.iterator() and hasNext()
outside of any try block. If the Stream throws there (an already consumed
Stream, or a pipeline stage failing on the first element), the exception
escaped stream() synchronously and the Stream was never closed.

Catch it like StreamableFromIterable does, close the Stream and return a
failed Streamer instead.

Author: Dev-next-gen

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

@@ -13,7 +13,7 @@
 
 package io.reactivex.rxjava4.internal.operators.streamable;
 
-import java.util.Objects;
+import java.util.*;
 import java.util.stream.Stream;
 
 import io.reactivex.rxjava4.annotations.NonNull;
@@ -27,8 +27,22 @@ public record StreamableFromStream<T>(@NonNull Stream<? extends T> items) implem
     @SuppressWarnings("unchecked")
     @Override
     public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) {
-        var iterator = Objects.requireNonNull(items.iterator(), "iterator is null");
-        if (!iterator.hasNext()) {
+        Iterator<? extends T> iterator;
+        boolean hasNext;
+        try {
+            iterator = Objects.requireNonNull(items.iterator(), "iterator is null");
+            hasNext = iterator.hasNext();
+        } catch (Throwable ex) {
+            Exceptions.throwIfFatal(ex);
+            try {
+                items.close();
+            } catch (Throwable exc) {
+                Exceptions.throwIfFatal(exc);
+                ex.addSuppressed(exc);
+            }
+            return StreamableError.createFailed(ex);
+        }
+        if (!hasNext) {
             try {
                 items.close();
             } catch (Throwable ex) {

src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFromStreamTest.java

@@ -13,16 +13,16 @@
 
 package io.reactivex.rxjava4.internal.operators.streamable;
 
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.*;
 
 import java.util.*;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.*;
 import java.util.stream.Stream;
 
 import org.junit.jupiter.api.Test;
 
-import io.reactivex.rxjava4.core.Streamable;
+import io.reactivex.rxjava4.core.*;
 import io.reactivex.rxjava4.exceptions.TestException;
 
 public class StreamableFromStreamTest extends StreamableBaseTest {
@@ -120,4 +120,30 @@ public void ensureCloses2Crash() {
 
         assertTrue(hasClosed.get(), "Close not called");
     }
+
+    @Test
+    public void firstItemCrashCloses() {
+        var hasClosed = new AtomicBoolean();
+        Streamable.fromStream(Stream.of(1, 2, 3).<Integer>map(_ -> { throw new TestException(); })
+                .onClose(() -> hasClosed.set(true)))
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertFailure(TestException.class);
+
+        assertTrue(hasClosed.get(), "Close not called");
+    }
+
+    @Test
+    public void alreadyConsumedStreamSink() throws Throwable {
+        var stream = Stream.of(1, 2, 3);
+        stream.forEach(_ -> { });
+
+        var error = new AtomicReference<Throwable>();
+        Streamable.fromStream(stream)
+        .subscribe(StreamSink.create(_ -> Streamer.NEXT_TRUE, e -> { error.set(e); return Streamer.FINISHED; }))
+        .toCompletableFuture()
+        .get(5, TimeUnit.SECONDS);
+
+        assertInstanceOf(IllegalStateException.class, error.get());
+    }
 }