ReadStream can be adapted to a Java blocking stream.

Motivation:

The Vert.x virtual stream programming model provides async/await to interact with asynchronous futures, yet the stream approach has been left naked.

Changes:

A new ReadStream blockingStream default method is added to convert a ReadStream to a java.util.stream.Stream.

This blocking stream can be used from a Vert.x virtual thread or a non Vert.x thread.

When used from a Vert.x virtual thread, each blocking interaction with the stream suspends the virtual thread and release the context ownership to let other tasks process tasks, allowing the context execution to happen and handle pending context tasks.

Author: vietj

vertx-core/src/main/asciidoc/virtualthreads.adoc

@@ -6,7 +6,7 @@ You still write the traditional Vert.x code processing events, but you have the
 
 === Introduction
 
-The non-blocking nature of Vert.x leads to asynchronous APIs. 
+The non-blocking nature of Vert.x leads to asynchronous APIs.
 Asynchronous APIs can take various forms including callback style, promises and reactive extensions.
 
 In some cases, programming using asynchronous APIs can be more challenging than using a direct synchronous style, in particular if you have several operations that you want to do sequentially.
@@ -57,6 +57,13 @@ or on a JDK `CompletionStage`
 {@link examples.VirtualThreadExamples#awaitingFutures2}
 ----
 
+You can also transform a Vert.x `ReadStream` to a Java blocking stream:
+
+[source,java]
+----
+{@link examples.VirtualThreadExamples#blockingStream}
+----
+
 ==== Field visibility
 
 A virtual thread verticle can interact safely with fields before an `await` call. However, if you are reading a field before an `await` call and reusing the value after the call, you should keep in mind that this value might have changed.

vertx-core/src/main/java/examples/VirtualThreadExamples.java

@@ -16,6 +16,7 @@
 import io.vertx.docgen.Source;
 
 import java.util.concurrent.CompletionStage;
+import java.util.stream.Stream;
 
 @Source
 public class VirtualThreadExamples {
@@ -93,6 +94,18 @@ public void awaitingFutures2(HttpClientResponse response, CompletionStage<Buffer
     Buffer body = Future.fromCompletionStage(completionStage).await();
   }
 
+  public void blockingStream(HttpServer server) {
+    server.requestHandler(request -> {
+      Stream<Buffer> blockingStream = request.blockingStream();
+      HttpServerResponse response = request.response();
+      response.setChunked(true);
+      blockingStream
+        .map(buff -> "" + buff.length())
+        .forEach(size -> response.write(size));
+      response.end();
+    });
+  }
+
   private Future<String> getRemoteString() {
     return null;
   }

vertx-core/src/main/java/io/vertx/core/internal/streams/ReadStreamIterator.java

@@ -116,7 +116,7 @@ public boolean hasNext() {
           return true;
         }
         if (ended != null) {
-          return false;
+          return ended != END_SENTINEL;
         }
         awaitProgress();
       }

vertx-core/src/main/java/io/vertx/core/streams/ReadStream.java

@@ -19,9 +19,15 @@
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.internal.PromiseInternal;
+import io.vertx.core.internal.streams.ReadStreamIterator;
 import io.vertx.core.streams.impl.PipeImpl;
 
+import java.util.Iterator;
+import java.util.Spliterator;
+import java.util.Spliterators;
 import java.util.function.BiConsumer;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 /**
  * Represents a stream of items that can be read from.
@@ -135,6 +141,18 @@ default <R, A> Future<R> collect(java.util.stream.Collector<T , A , R> collector
     return promise.future();
   }
 
+  /**
+   * Adapt this {@code ReadStream} to a blocking sequential {@code Stream}, the return stream usage is restricted to
+   * non vertx threads or vertx virtual threads.
+   *
+   * @return a blocking stream
+   */
+  @GenIgnore
+  default Stream<T> blockingStream() {
+    Iterator<T> iterator = ReadStreamIterator.iterator(this);
+    return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, Spliterator.ORDERED), false);
+  }
+
   /**
    * Pipe this {@code ReadStream} to the {@code WriteStream}.
    * <p>

vertx-core/src/test/java/io/vertx/tests/http/VirtualThreadHttpTest.java

@@ -21,7 +21,15 @@
 import org.junit.Test;
 
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.StreamSupport;
 
 public class VirtualThreadHttpTest extends VertxTestBase {
 
@@ -87,6 +95,43 @@ public void testHttpClient2() throws Exception {
     }
   }
 
+  @Test
+  public void testHttpClient3() throws Exception {
+    Assume.assumeTrue(isVirtualThreadAvailable());
+    HttpServer server = vertx.createHttpServer();
+    int numChunks = 10;
+    List<String> expected = IntStream.range(0, numChunks).mapToObj(idx -> "chunk-" + idx).collect(Collectors.toList());
+    server.requestHandler(req -> {
+      HttpServerResponse response = req.response();
+      response.setChunked(true);
+      Deque<String> toSend = new ArrayDeque<>(expected);
+      vertx.setPeriodic(10, id -> {
+        String chunk = toSend.poll();
+        if (chunk != null) {
+          response.write(chunk);
+        } else {
+          vertx.cancelTimer(id);
+          response.end();
+        }
+      });
+    });
+    server.listen(8088, "localhost").await(10, TimeUnit.SECONDS);
+    vertx.createVirtualThreadContext().runOnContext(v -> {
+      HttpClient client = vertx.createHttpClient();
+      for (int i = 0; i < 10; ++i) {
+        HttpClientRequest req = client.request(HttpMethod.GET, 8088, "localhost", "/").await();
+        HttpClientResponse resp = req.send().await();
+        List<String> chunks = new ArrayList<>();
+        resp.blockingStream().forEach(chunk -> {
+          chunks.add(chunk.toString());
+        });
+        assertEquals(expected, chunks);
+      }
+      testComplete();
+    });
+    await();
+  }
+
   @Test
   public void testHttpClientTimeout() throws Exception {
     Assume.assumeTrue(isVirtualThreadAvailable());

vertx-core/src/test/java/io/vertx/tests/streams/IteratorTest.java

@@ -16,26 +16,23 @@
 import io.vertx.core.internal.VertxInternal;
 import io.vertx.core.internal.streams.ReadStreamIterator;
 import io.vertx.core.streams.ReadStream;
-import io.vertx.test.core.AsyncTestBase;
 import io.vertx.test.core.Repeat;
-import io.vertx.test.core.RepeatRule;
+import io.vertx.test.core.VertxTestBase;
 import io.vertx.test.fakestream.FakeStream;
-import org.junit.Rule;
 import org.junit.Assume;
 import org.junit.Test;
 
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.concurrent.CountDownLatch;
+import java.util.*;
 import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
-public class IteratorTest extends AsyncTestBase {
+public class IteratorTest extends VertxTestBase {
 
   @Test
   public void testIteratorResuming() {
@@ -87,7 +84,7 @@ public void testFail() {
       assertTrue(iterator.hasNext());
       iterator.next();
     }
-    assertFalse(iterator.hasNext());
+    assertTrue(iterator.hasNext());
     try {
       iterator.next();
       fail();
@@ -114,9 +111,6 @@ public void testHasNextSignal() throws Exception {
     }
   }
 
-  @Rule
-  public RepeatRule rule = new RepeatRule();
-
   @Repeat(times = 100)
   @Test
   public void testConcurrentReads() throws Exception {
@@ -286,13 +280,9 @@ public void run() {
 
   @Test
   public void testVirtualThread() {
-    VertxInternal vertx = (VertxInternal) Vertx.vertx();
-    try {
-      Assume.assumeTrue(vertx.isVirtualThreadAvailable());
-      doTestVirtualThread(vertx);
-    } finally {
-      vertx.close();
-    }
+    VertxInternal vertx = (VertxInternal) this.vertx;
+    Assume.assumeTrue(vertx.isVirtualThreadAvailable());
+    doTestVirtualThread(vertx);
   }
 
   private void doTestVirtualThread(VertxInternal vertx) {
@@ -312,4 +302,103 @@ private void doTestVirtualThread(VertxInternal vertx) {
     });
     await();
   }
+
+  @Test
+  public void testBlockingStreamFromVirtualThread() {
+    VertxInternal vertx = (VertxInternal) this.vertx;
+    Assume.assumeTrue(vertx.isVirtualThreadAvailable());
+    ContextInternal context = vertx.createVirtualThreadContext();
+    testBlockingStream(task -> {
+      context.runOnContext(v -> task.run());
+    });
+  }
+
+  @Test
+  public void testBlockingStreamFromVanillaThread() {
+    testBlockingStream(task -> {
+      Thread thread = new Thread(task);
+      thread.start();
+    });
+  }
+
+  @Test
+  public void testBlockingStreamFromVertxThread() {
+    VertxInternal vertx = (VertxInternal) this.vertx;
+    FakeStream<Integer> readStream = new FakeStream<>();
+    List<Throwable> errors = Collections.synchronizedList(new ArrayList<>());
+    Stream<Integer> blockingStream = readStream.blockingStream();
+    ContextInternal context = vertx.createEventLoopContext();
+    context.exceptionHandler(errors::add);
+    context.runOnContext(v -> {
+      blockingStream.forEach(elt -> {
+        fail();
+      });
+    });
+    assertWaitUntil(() -> errors.size() == 1);
+    assertEquals(IllegalStateException.class, errors.get(0).getClass());
+  }
+
+  private void testBlockingStream(Executor runner) {
+    int numItems = 4;
+    FakeStream<Integer> stream = new FakeStream<>();
+    Stream<Integer> blockingStream = stream.blockingStream();
+    List<Integer> expected = new ArrayList<>();
+    for (int i = 0;i < numItems;i++) {
+      expected.add(i);
+    }
+    List<Integer> items = Collections.synchronizedList(new ArrayList<>());
+    runner.execute(() -> items.addAll(blockingStream.collect(Collectors.toList())));
+    new Thread(() -> {
+      try {
+        for (int elt : expected) {
+          stream.write(elt);
+          Thread.sleep(10);
+        }
+        stream.end();
+      } catch (InterruptedException ignore) {
+      }
+    }).start();
+    assertWaitUntil(() -> items.equals(expected));
+  }
+
+  @Test
+  public void testBlockingStreamInterleavingFromVirtualThread() {
+    VertxInternal vertx = (VertxInternal) this.vertx;
+    Assume.assumeTrue(vertx.isVirtualThreadAvailable());
+    ContextInternal context = vertx.createVirtualThreadContext();
+    FakeStream<Integer> stream = new FakeStream<>();
+    Stream<Integer> blockingStream = stream.blockingStream();
+    int num = 32;
+    context.runOnContext(v -> {
+      AtomicInteger count = new AtomicInteger();
+      vertx.setPeriodic(5, id -> {
+        assertSame(context, ((ContextInternal) Vertx.currentContext()).unwrap());
+        int i = count.getAndIncrement();
+        if (i == num) {
+          vertx.cancelTimer(id);
+          stream.end();
+        } else {
+          stream.write(i);
+        }
+      });
+      List<Integer> collected = new ArrayList<>();
+      blockingStream.forEach(collected::add);
+      assertEquals(num, collected.size());
+      testComplete();
+    });
+    await();
+  }
+
+  @Test
+  public void testStreamFailure() {
+    RuntimeException expected = new RuntimeException();
+    FakeStream<Integer> stream = new FakeStream<>();
+    Stream<Integer> blockingStream = stream.blockingStream();
+    stream.fail(expected);
+    try {
+      blockingStream.collect(Collectors.toList());
+    } catch (Exception e) {
+      assertSame(expected, e);
+    }
+  }
 }