You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@groovy.apache.org by "paulk-asert (via GitHub)" <gi...@apache.org> on 2026/04/17 07:20:31 UTC
[PR] GROOVY-11933: Mesh async and http builder via Flow.Publisher [groovy]
paulk-asert opened a new pull request, #2470:
URL: https://github.com/apache/groovy/pull/2470
This a piece that ties together http builder and async pieces but is of most value if we proceed with GEP-18. I'll provide a separate PR with a good example showing that but under a separate Jira.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GROOVY-11933: Mesh async and http builder via Flow.Publisher [groovy]
Posted by "paulk-asert (via GitHub)" <gi...@apache.org>.
paulk-asert merged PR #2470:
URL: https://github.com/apache/groovy/pull/2470
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GROOVY-11933: Mesh async and http builder via Flow.Publisher [groovy]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2470:
URL: https://github.com/apache/groovy/pull/2470#discussion_r3101731088
##########
src/main/java/org/apache/groovy/runtime/async/FlowPublisherAdapter.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.groovy.runtime.async;
+
+import groovy.concurrent.Awaitable;
+import groovy.concurrent.AwaitableAdapter;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Adapter for {@link java.util.concurrent.Flow.Publisher}, the JDK's built-in
+ * Reactive Streams type. Enables:
+ * <ul>
+ * <li>{@code await publisher} — completes with the first {@code onNext}
+ * value, then cancels the subscription. Completes with {@code null} if
+ * the publisher signals {@code onComplete} without emitting.</li>
+ * <li>{@code for await (item in publisher)} — iterates over emitted values
+ * with bounded backpressure (see {@link #DEFAULT_BATCH_SIZE}).</li>
+ * </ul>
+ * <p>
+ * Conformance:
+ * <ul>
+ * <li>Reactive Streams §2.13: {@code onNext(null)} is treated as a protocol
+ * violation and surfaced as a {@link NullPointerException}.</li>
+ * <li>A second {@code onSubscribe} after the first is cancelled.</li>
+ * <li>Signals after a terminal {@code onError}/{@code onComplete} are ignored.</li>
+ * </ul>
+ * <p>
+ * This adapter is registered as the lowest-priority built-in (after SPI-loaded
+ * adapters) so framework-specific adapters (Reactor, RxJava) take precedence
+ * for their concrete types.
+ *
+ * @since 6.0.0
+ */
+public final class FlowPublisherAdapter implements AwaitableAdapter {
+
+ /**
+ * Default request batch size for {@code for await} iteration. Chosen as a
+ * compromise between throughput (larger = fewer {@code request()} calls)
+ * and memory (larger = bigger in-flight buffer). Override per-call by
+ * wrapping with a custom adapter if needed.
+ */
+ public static final int DEFAULT_BATCH_SIZE = 32;
+
+ @Override
+ public boolean supportsAwaitable(Class<?> type) {
+ return Flow.Publisher.class.isAssignableFrom(type);
+ }
+
+ @Override
+ public boolean supportsIterable(Class<?> type) {
+ return Flow.Publisher.class.isAssignableFrom(type);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> Awaitable<T> toAwaitable(Object source) {
+ if (!(source instanceof Flow.Publisher<?>)) {
+ throw new IllegalArgumentException("Cannot convert to Awaitable: " + source.getClass());
+ }
+ return (Awaitable<T>) publisherToAwaitable((Flow.Publisher<Object>) source);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> Iterable<T> toBlockingIterable(Object source) {
+ if (!(source instanceof Flow.Publisher<?>)) {
+ throw new IllegalArgumentException("Cannot convert to Iterable: " + source.getClass());
+ }
+ return (Iterable<T>) publisherToBlockingIterable((Flow.Publisher<Object>) source, DEFAULT_BATCH_SIZE);
+ }
+
+ // ---- single-value: first onNext wins ---------------------------------
+
+ private static <T> Awaitable<T> publisherToAwaitable(Flow.Publisher<T> publisher) {
+ CompletableFuture<T> cf = new CompletableFuture<>();
+ AtomicReference<Flow.Subscription> subRef = new AtomicReference<>();
+ publisher.subscribe(new Flow.Subscriber<T>() {
+ @Override
+ public void onSubscribe(Flow.Subscription s) {
+ if (!subRef.compareAndSet(null, s)) {
+ s.cancel(); // duplicate onSubscribe — cancel the second
+ return;
+ }
+ s.request(1);
+ }
+
+ @Override
+ public void onNext(T item) {
+ if (cf.isDone()) return;
+ if (item == null) {
+ cf.completeExceptionally(new NullPointerException(
+ "Flow.Publisher onNext received null (Reactive Streams §2.13)"));
+ } else {
+ cf.complete(item);
+ }
+ Flow.Subscription s = subRef.getAndSet(null);
+ if (s != null) s.cancel();
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ if (cf.isDone()) return;
+ cf.completeExceptionally(t);
+ subRef.set(null);
+ }
+
+ @Override
+ public void onComplete() {
+ if (cf.isDone()) return;
+ cf.complete(null); // empty publisher → null result
+ subRef.set(null);
+ }
+ });
+ return GroovyPromise.of(cf);
+ }
+
+ // ---- multi-value: blocking iterable with bounded backpressure --------
+
+ private static <T> Iterable<T> publisherToBlockingIterable(Flow.Publisher<T> publisher, int batchSize) {
+ return new PublisherBlockingIterable<>(publisher, batchSize);
+ }
+
+ /**
+ * Blocking {@link Iterable} backed by a {@link Flow.Publisher} with bounded
+ * backpressure. Implements {@link AutoCloseable} so {@code for await}
+ * cleanup (via {@code AsyncSupport.closeIterable}) cancels the subscription
+ * on early break.
+ */
+ private static final class PublisherBlockingIterable<T> implements Iterable<T>, AutoCloseable {
+ private final Flow.Publisher<T> publisher;
+ private final int batchSize;
+ private volatile QueueSubscriber<T> active;
+
+ PublisherBlockingIterable(Flow.Publisher<T> publisher, int batchSize) {
+ this.publisher = publisher;
+ this.batchSize = Math.max(1, batchSize);
+ }
+
+ @Override
+ public Iterator<T> iterator() {
+ QueueSubscriber<T> sub = new QueueSubscriber<>(batchSize);
+ this.active = sub;
+ publisher.subscribe(sub);
+ return sub;
+ }
+
+ @Override
+ public void close() {
+ QueueSubscriber<T> sub = this.active;
+ if (sub != null) sub.cancel();
+ }
+ }
+
+ /**
+ * A {@link Flow.Subscriber} that buffers signals into a blocking queue and
+ * exposes them as an {@link Iterator}. Demand is replenished in halves of
+ * the batch size to overlap consumption with production.
+ */
+ private static final class QueueSubscriber<T> implements Flow.Subscriber<T>, Iterator<T> {
+ private static final Object COMPLETE = new Object();
+ private static final Object ERROR = new Object();
+
+ private final LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<>();
+ private final int batchSize;
+ private final int replenishThreshold;
+ private final AtomicReference<Flow.Subscription> subRef = new AtomicReference<>();
+
+ private Object next;
+ private boolean hasNext;
+ private boolean terminated;
+ private Throwable error;
+ private int consumedSinceRequest;
+
+ QueueSubscriber(int batchSize) {
+ this.batchSize = batchSize;
+ this.replenishThreshold = Math.max(1, batchSize / 2);
+ }
+
+ @Override
+ public void onSubscribe(Flow.Subscription s) {
+ if (!subRef.compareAndSet(null, s)) {
+ s.cancel();
+ return;
+ }
+ s.request(batchSize);
+ }
+
+ @Override
+ public void onNext(T item) {
+ if (item == null) {
+ onError(new NullPointerException(
+ "Flow.Publisher onNext received null (Reactive Streams §2.13)"));
+ return;
+ }
+ queue.offer(item);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ this.error = t;
+ queue.offer(ERROR);
+ subRef.set(null);
+ }
+
+ @Override
+ public void onComplete() {
+ queue.offer(COMPLETE);
+ subRef.set(null);
+ }
+
+ void cancel() {
+ Flow.Subscription s = subRef.getAndSet(null);
+ if (s != null) s.cancel();
+ terminated = true;
+ queue.clear();
Review Comment:
QueueSubscriber.cancel() clears the queue but doesn't unblock a thread currently waiting in queue.take() inside hasNext(). If cancel/close is invoked from another thread while iteration is blocked, the iterator can hang indefinitely. Consider signalling termination on cancel (e.g., enqueue a terminal sentinel) so any blocked hasNext() wakes up and stops promptly.
```suggestion
queue.clear();
queue.offer(COMPLETE);
```
##########
src/test/groovy/groovy/concurrent/FlowPublisherAdapterTest.groovy:
##########
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 groovy.concurrent
+
+import org.junit.jupiter.api.Test
+
+import java.util.concurrent.Flow
+import java.util.concurrent.SubmissionPublisher
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
+
+import static groovy.test.GroovyAssert.assertScript
+import static groovy.test.GroovyAssert.shouldFail
+
+final class FlowPublisherAdapterTest {
+
+ @Test
+ void testAwaitOnSubmissionPublisherTakesFirstValue() {
+ assertScript '''
+ import java.util.concurrent.SubmissionPublisher
+ import groovy.concurrent.Awaitable
+
+ def publisher = new SubmissionPublisher<String>()
+ async {
+ publisher.submit('first')
+ publisher.submit('second')
+ publisher.close()
+ }
+ def result = await Awaitable.from(publisher)
+ assert result == 'first'
+ '''
+ }
+
+ @Test
+ void testAwaitOnEmptyPublisherCompletesWithNull() {
+ assertScript '''
+ import java.util.concurrent.SubmissionPublisher
+ import groovy.concurrent.Awaitable
+
+ def publisher = new SubmissionPublisher<String>()
+ async { publisher.close() }
+ def result = await Awaitable.from(publisher)
+ assert result == null
+ '''
+ }
+
+ @Test
+ void testAwaitOnFailingPublisherRethrows() {
+ def boom = new RuntimeException('boom')
+ Throwable thrown = shouldFail RuntimeException, {
+ def publisher = new SubmissionPublisher<String>()
+ new Thread({
+ publisher.closeExceptionally(boom)
+ }).start()
+ org.apache.groovy.runtime.async.AsyncSupport.await(groovy.concurrent.Awaitable.from(publisher))
+ }
+ assert thrown.message == 'boom'
+ }
+
+ @Test
+ void testForAwaitIteratesAllValues() {
+ assertScript '''
+ import java.util.concurrent.SubmissionPublisher
+
+ def publisher = new SubmissionPublisher<Integer>()
+ async {
+ (1..5).each { publisher.submit(it) }
+ publisher.close()
+ }
+ def results = []
+ for await (item in publisher) {
+ results << item
+ }
+ assert results == [1, 2, 3, 4, 5]
+ '''
+ }
+
+ @Test
+ void testForAwaitOnEmptyPublisher() {
+ assertScript '''
+ import java.util.concurrent.SubmissionPublisher
+
+ def publisher = new SubmissionPublisher<Integer>()
+ async { publisher.close() }
+ def results = []
+ for await (item in publisher) {
+ results << item
+ }
+ assert results == []
+ '''
+ }
+
+ @Test
+ void testForAwaitPropagatesError() {
+ def thrown = shouldFail RuntimeException, '''
+ import java.util.concurrent.SubmissionPublisher
+
+ def publisher = new SubmissionPublisher<Integer>()
+ async {
+ publisher.submit(1)
+ publisher.submit(2)
+ publisher.closeExceptionally(new RuntimeException('mid-stream'))
+ }
+ def results = []
+ for await (item in publisher) {
+ results << item
+ }
+ '''
+ assert thrown.message == 'mid-stream'
+ }
+
+ @Test
+ void testNullOnNextBecomesNpe() {
+ // Custom publisher that violates §2.13
+ def publisher = { Flow.Subscriber<Object> sub ->
+ sub.onSubscribe(new Flow.Subscription() {
+ @Override void request(long n) { sub.onNext(null) }
+ @Override void cancel() {}
+ })
+ } as Flow.Publisher<Object>
+
+ def thrown = shouldFail NullPointerException, {
+ org.apache.groovy.runtime.async.AsyncSupport.await(groovy.concurrent.Awaitable.from(publisher))
+ }
+ assert thrown.message.contains('§2.13')
+ }
+
+ @Test
+ void testEarlyBreakCancelsSubscription() {
+ // Verify the subscription is cancelled when the consumer breaks early.
+ AtomicBoolean cancelled = new AtomicBoolean()
+ AtomicInteger requested = new AtomicInteger()
+
+ def publisher = { Flow.Subscriber<Integer> sub ->
+ sub.onSubscribe(new Flow.Subscription() {
+ @Override
+ void request(long n) {
+ requested.addAndGet((int) n)
+ // Emit a few values; do not auto-complete
+ Thread.startDaemon {
+ for (int i = 0; i < (int) n; i++) {
+ sub.onNext(i)
+ }
+ }
+ }
+ @Override void cancel() { cancelled.set(true) }
+ })
+ } as Flow.Publisher<Integer>
+
+ def shell = new GroovyShell(new Binding(publisher: publisher))
+ shell.evaluate '''
+ int seen = 0
+ for await (item in publisher) {
+ seen++
+ if (seen >= 3) break
+ }
+ '''
+
+ // Give the runtime a moment to fire the close hook
+ Thread.sleep(100)
+ assert cancelled.get(), 'Subscription should have been cancelled on early break'
+ }
Review Comment:
The test relies on Thread.sleep(100) to wait for cancellation to propagate, which can be flaky under load/slow CI. Prefer a deterministic wait (e.g., a CountDownLatch triggered by cancel(), or polling with a bounded timeout) so the test doesn't depend on timing.
##########
src/main/java/org/apache/groovy/runtime/async/FlowPublisherAdapter.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.groovy.runtime.async;
+
+import groovy.concurrent.Awaitable;
+import groovy.concurrent.AwaitableAdapter;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Adapter for {@link java.util.concurrent.Flow.Publisher}, the JDK's built-in
+ * Reactive Streams type. Enables:
+ * <ul>
+ * <li>{@code await publisher} — completes with the first {@code onNext}
+ * value, then cancels the subscription. Completes with {@code null} if
+ * the publisher signals {@code onComplete} without emitting.</li>
+ * <li>{@code for await (item in publisher)} — iterates over emitted values
+ * with bounded backpressure (see {@link #DEFAULT_BATCH_SIZE}).</li>
+ * </ul>
+ * <p>
+ * Conformance:
+ * <ul>
+ * <li>Reactive Streams §2.13: {@code onNext(null)} is treated as a protocol
+ * violation and surfaced as a {@link NullPointerException}.</li>
+ * <li>A second {@code onSubscribe} after the first is cancelled.</li>
+ * <li>Signals after a terminal {@code onError}/{@code onComplete} are ignored.</li>
+ * </ul>
+ * <p>
+ * This adapter is registered as the lowest-priority built-in (after SPI-loaded
+ * adapters) so framework-specific adapters (Reactor, RxJava) take precedence
+ * for their concrete types.
+ *
+ * @since 6.0.0
+ */
+public final class FlowPublisherAdapter implements AwaitableAdapter {
+
+ /**
+ * Default request batch size for {@code for await} iteration. Chosen as a
+ * compromise between throughput (larger = fewer {@code request()} calls)
+ * and memory (larger = bigger in-flight buffer). Override per-call by
+ * wrapping with a custom adapter if needed.
+ */
+ public static final int DEFAULT_BATCH_SIZE = 32;
+
+ @Override
+ public boolean supportsAwaitable(Class<?> type) {
+ return Flow.Publisher.class.isAssignableFrom(type);
+ }
+
+ @Override
+ public boolean supportsIterable(Class<?> type) {
+ return Flow.Publisher.class.isAssignableFrom(type);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> Awaitable<T> toAwaitable(Object source) {
+ if (!(source instanceof Flow.Publisher<?>)) {
+ throw new IllegalArgumentException("Cannot convert to Awaitable: " + source.getClass());
+ }
+ return (Awaitable<T>) publisherToAwaitable((Flow.Publisher<Object>) source);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> Iterable<T> toBlockingIterable(Object source) {
+ if (!(source instanceof Flow.Publisher<?>)) {
+ throw new IllegalArgumentException("Cannot convert to Iterable: " + source.getClass());
+ }
+ return (Iterable<T>) publisherToBlockingIterable((Flow.Publisher<Object>) source, DEFAULT_BATCH_SIZE);
+ }
+
+ // ---- single-value: first onNext wins ---------------------------------
+
+ private static <T> Awaitable<T> publisherToAwaitable(Flow.Publisher<T> publisher) {
+ CompletableFuture<T> cf = new CompletableFuture<>();
+ AtomicReference<Flow.Subscription> subRef = new AtomicReference<>();
+ publisher.subscribe(new Flow.Subscriber<T>() {
+ @Override
+ public void onSubscribe(Flow.Subscription s) {
+ if (!subRef.compareAndSet(null, s)) {
+ s.cancel(); // duplicate onSubscribe — cancel the second
+ return;
+ }
+ s.request(1);
+ }
+
+ @Override
+ public void onNext(T item) {
+ if (cf.isDone()) return;
+ if (item == null) {
+ cf.completeExceptionally(new NullPointerException(
+ "Flow.Publisher onNext received null (Reactive Streams §2.13)"));
+ } else {
+ cf.complete(item);
+ }
+ Flow.Subscription s = subRef.getAndSet(null);
+ if (s != null) s.cancel();
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ if (cf.isDone()) return;
+ cf.completeExceptionally(t);
+ subRef.set(null);
+ }
+
+ @Override
+ public void onComplete() {
+ if (cf.isDone()) return;
+ cf.complete(null); // empty publisher → null result
+ subRef.set(null);
+ }
+ });
+ return GroovyPromise.of(cf);
+ }
+
+ // ---- multi-value: blocking iterable with bounded backpressure --------
+
+ private static <T> Iterable<T> publisherToBlockingIterable(Flow.Publisher<T> publisher, int batchSize) {
+ return new PublisherBlockingIterable<>(publisher, batchSize);
+ }
+
+ /**
+ * Blocking {@link Iterable} backed by a {@link Flow.Publisher} with bounded
+ * backpressure. Implements {@link AutoCloseable} so {@code for await}
+ * cleanup (via {@code AsyncSupport.closeIterable}) cancels the subscription
+ * on early break.
+ */
+ private static final class PublisherBlockingIterable<T> implements Iterable<T>, AutoCloseable {
+ private final Flow.Publisher<T> publisher;
+ private final int batchSize;
+ private volatile QueueSubscriber<T> active;
+
+ PublisherBlockingIterable(Flow.Publisher<T> publisher, int batchSize) {
+ this.publisher = publisher;
+ this.batchSize = Math.max(1, batchSize);
+ }
+
+ @Override
+ public Iterator<T> iterator() {
+ QueueSubscriber<T> sub = new QueueSubscriber<>(batchSize);
+ this.active = sub;
+ publisher.subscribe(sub);
+ return sub;
+ }
Review Comment:
PublisherBlockingIterable implements Iterable and allows iterator() to be called multiple times, but close() only cancels the most recently created subscription (active). This can leak subscriptions if the iterable is iterated more than once. Consider either preventing multiple iterators (fail fast) or tracking/cancelling all active subscribers.
##########
subprojects/groovy-http-builder/src/test/groovy/groovy/http/HttpStreamResultTest.groovy:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 groovy.http
+
+import com.sun.net.httpserver.HttpServer
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.Flow
+import java.util.concurrent.SubmissionPublisher
+
+import static groovy.test.GroovyAssert.assertScript
+
+class HttpStreamResultTest {
+
+ private HttpServer server
+ private URI rootUri
+
+ @BeforeEach
+ void setup() {
+ server = HttpServer.create(new InetSocketAddress('127.0.0.1', 0), 0)
+ server.createContext('/lines') { exchange ->
+ byte[] header = "ignored".getBytes(StandardCharsets.UTF_8)
+ // chunked transfer (size 0)
Review Comment:
setup() defines a 'header' byte[] that is never used, and the file has unused imports (e.g. SubmissionPublisher, GroovyAssert.assertScript). Cleaning these up will make the test clearer and avoid accumulating dead code.
##########
subprojects/groovy-http-builder/src/test/groovy/groovy/http/HttpStreamResultTest.groovy:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 groovy.http
+
+import com.sun.net.httpserver.HttpServer
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.Flow
+import java.util.concurrent.SubmissionPublisher
+
+import static groovy.test.GroovyAssert.assertScript
+
+class HttpStreamResultTest {
+
+ private HttpServer server
+ private URI rootUri
+
+ @BeforeEach
+ void setup() {
+ server = HttpServer.create(new InetSocketAddress('127.0.0.1', 0), 0)
+ server.createContext('/lines') { exchange ->
+ byte[] header = "ignored".getBytes(StandardCharsets.UTF_8)
+ // chunked transfer (size 0)
+ exchange.responseHeaders.add('Content-Type', 'text/plain')
+ exchange.sendResponseHeaders(200, 0)
+ exchange.responseBody.withCloseable { out ->
+ (1..5).each { i ->
+ out.write("line-${i}\n".getBytes(StandardCharsets.UTF_8))
+ out.flush()
+ Thread.sleep(10)
+ }
+ }
+ }
+ server.createContext('/bytes') { exchange ->
+ byte[] body = ('A'..'Z').collect { it.toString() }.join().getBytes(StandardCharsets.UTF_8)
+ exchange.sendResponseHeaders(200, body.length)
+ exchange.responseBody.withCloseable { it.write(body) }
+ }
+ server.start()
+ rootUri = URI.create("http://127.0.0.1:${server.address.port}/")
+ }
+
+ @AfterEach
+ void cleanup() {
+ server?.stop(0)
+ }
+
+ @Test
+ void streamsBytesAsPublisher() {
+ HttpBuilder http = HttpBuilder.http(rootUri.toString())
+ CompletableFuture<HttpStreamResult> future = http.getStreamAsync('/bytes')
+ HttpStreamResult result = future.join()
+ assert result.status() == 200
+
+ // Drain the publisher manually to a collected string
+ StringBuilder collected = new StringBuilder()
+ def latch = new java.util.concurrent.CountDownLatch(1)
+ result.bodyAsPublisher().subscribe(new Flow.Subscriber<List>() {
+ Flow.Subscription sub
+ @Override void onSubscribe(Flow.Subscription s) { sub = s; s.request(Long.MAX_VALUE) }
+ @Override void onNext(List buffers) {
+ buffers.each { collected.append(StandardCharsets.UTF_8.decode(it)) }
+ }
+ @Override void onError(Throwable t) { latch.countDown() }
+ @Override void onComplete() { latch.countDown() }
+ })
+ latch.await()
+ assert collected.toString() == ('A'..'Z').collect { it.toString() }.join()
+ }
+
+ @Test
+ void streamsLinesAsLinePublisher() {
+ HttpBuilder http = HttpBuilder.http(rootUri.toString())
+ HttpStreamResult result = http.getStreamAsync('/lines').join()
+ assert result.status() == 200
+
+ List<String> lines = []
+ def latch = new java.util.concurrent.CountDownLatch(1)
+ result.bodyAsLinePublisher().subscribe(new Flow.Subscriber<String>() {
+ @Override void onSubscribe(Flow.Subscription s) { s.request(Long.MAX_VALUE) }
+ @Override void onNext(String line) { lines << line }
+ @Override void onError(Throwable t) { latch.countDown() }
+ @Override void onComplete() { latch.countDown() }
+ })
+ latch.await()
+ assert lines == ['line-1', 'line-2', 'line-3', 'line-4', 'line-5']
Review Comment:
These tests block on CountDownLatch.await() without a timeout. If the publisher never completes (regression, CI slowness, etc.), the build can hang indefinitely. Use a bounded await (with an assertion on success) to keep the test suite fail-fast.
##########
subprojects/groovy-http-builder/src/main/groovy/groovy/http/HttpBuilder.groovy:
##########
@@ -166,6 +166,62 @@ final class HttpBuilder {
return requestAsync('PATCH', uri, spec)
}
+ /**
+ * Streams the response body without buffering. Returns an
+ * {@link HttpStreamResult} whose {@code bodyAsPublisher()} delivers
+ * {@code List<ByteBuffer>} chunks as they arrive — enabling
+ * {@code for await (chunk in result.bodyAsPublisher())} via the
+ * {@code FlowPublisherAdapter}.
+ */
+ CompletableFuture<HttpStreamResult> streamAsync(final String method,
+ final Object uri,
+ @DelegatesTo(value = RequestSpec, strategy = Closure.DELEGATE_FIRST)
+ final Closure<?> spec = null) {
+ HttpRequest httpRequest = buildStreamRequest(method, uri, spec)
+ return client.sendAsync(httpRequest, HttpResponse.BodyHandlers.ofPublisher())
+ .thenApply { HttpResponse response -> new HttpStreamResult(response) }
+ }
+
+ CompletableFuture<HttpStreamResult> getStreamAsync(final Object uri = null,
+ @DelegatesTo(value = RequestSpec, strategy = Closure.DELEGATE_FIRST)
+ final Closure<?> spec = null) {
+ return streamAsync('GET', uri, spec)
+ }
+
+ CompletableFuture<HttpStreamResult> postStreamAsync(final Object uri = null,
+ @DelegatesTo(value = RequestSpec, strategy = Closure.DELEGATE_FIRST)
+ final Closure<?> spec = null) {
+ return streamAsync('POST', uri, spec)
+ }
+
+ private HttpRequest buildStreamRequest(final String method, final Object uri, final Closure<?> spec) {
+ RequestSpec requestSpec = new RequestSpec()
+ if (spec != null) {
+ Closure<?> code = (Closure<?>) spec.clone()
+ code.resolveStrategy = Closure.DELEGATE_FIRST
+ code.delegate = requestSpec
+ code.call()
+ }
+
+ URI resolvedUri = resolveUri(uri, requestSpec.queryParameters)
+ HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(resolvedUri)
+
+ Duration timeout = requestSpec.timeout ?: defaultRequestTimeout
+ if (timeout != null) {
+ requestBuilder.timeout(timeout)
+ }
+
+ defaultHeaders.each { String name, String value ->
+ requestBuilder.header(name, value)
+ }
+ requestSpec.headers.each { String name, String value ->
+ requestBuilder.setHeader(name, value)
+ }
+
+ requestBuilder.method(method, bodyPublisher(method, requestSpec.body))
+ return requestBuilder.build()
Review Comment:
streamAsync/getStreamAsync/postStreamAsync accept the same RequestSpec closure as the non-streaming APIs, but any bodyHandler configuration in that spec (e.g. RequestSpec.asString()) is ignored because streaming always uses BodyHandlers.ofPublisher(). This is surprising for callers; consider using a dedicated spec type for streaming requests (without bodyHandler), or explicitly rejecting/ignoring bodyHandler with a clear error/documentation.
##########
subprojects/groovy-http-builder/src/main/groovy/groovy/http/HttpStreamResult.groovy:
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 groovy.http
+
+import org.apache.groovy.lang.annotation.Incubating
+
+import java.net.http.HttpHeaders
+import java.net.http.HttpResponse
+import java.nio.ByteBuffer
+import java.nio.charset.Charset
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.Flow
+import java.util.concurrent.SubmissionPublisher
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * Streaming response wrapper for {@link HttpBuilder}'s {@code streamAsync} family.
+ * <p>
+ * Unlike {@link HttpResult}, this does not buffer the response body. The body
+ * is exposed as a {@link Flow.Publisher} of byte chunks that can be consumed
+ * incrementally — typically via {@code for await (chunk in result.bodyAsPublisher())}
+ * once the {@code groovy-concurrent-java} {@code FlowPublisherAdapter} is on
+ * the classpath.
+ */
+@Incubating
+record HttpStreamResult(
+ int status,
+ HttpHeaders headers,
+ HttpResponse<Flow.Publisher<List<ByteBuffer>>> raw) {
+
+ HttpStreamResult(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {
+ this(response.statusCode(), response.headers(), response)
+ }
+
+ /**
+ * Returns the response body as a publisher of {@code List<ByteBuffer>}
+ * chunks, exactly as JDK {@code HttpResponse.BodyHandlers.ofPublisher()}
+ * provides it.
+ */
+ Flow.Publisher<List<ByteBuffer>> bodyAsPublisher() {
+ return raw.body()
+ }
+
+ /**
+ * Returns the response body as a publisher of decoded strings, one per
+ * arriving chunk. Each emitted string corresponds to one upstream HTTP
+ * frame (often a complete SSE event or a JSON line, but boundaries are
+ * not guaranteed). Use {@link #bodyAsLinePublisher} for line-aligned
+ * delivery.
+ *
+ * @param charset the charset to decode bytes with (defaults to UTF-8)
+ */
+ Flow.Publisher<String> bodyAsTextPublisher(Charset charset = StandardCharsets.UTF_8) {
+ Closure mapper = { chunk ->
+ StringBuilder sb = new StringBuilder()
+ for (ByteBuffer buf : (List<ByteBuffer>) chunk) {
+ sb.append(charset.decode(buf))
+ }
+ sb.toString()
+ }
+ return (Flow.Publisher<String>) new MappedPublisher(raw.body(), mapper)
+ }
+
+ /**
+ * Returns the response body as a publisher of newline-terminated strings.
+ * Useful for line-oriented streaming protocols (NDJSON, SSE without event
+ * framing, log tails). Trailing partial lines are emitted on completion.
+ *
+ * @param charset the charset to decode bytes with (defaults to UTF-8)
+ */
+ Flow.Publisher<String> bodyAsLinePublisher(Charset charset = StandardCharsets.UTF_8) {
+ return new LinePublisher(raw.body(), charset)
+ }
+
+ /**
+ * One-shot mapping {@link Flow.Publisher} that delegates subscription to a
+ * source publisher and applies a transformer per signalled item.
+ */
+ private static final class MappedPublisher implements Flow.Publisher<Object> {
+ private final Flow.Publisher source
+ private final Closure mapper
+
+ MappedPublisher(Flow.Publisher source, Closure mapper) {
+ this.source = source
+ this.mapper = mapper
+ }
+
+ @Override
+ void subscribe(Flow.Subscriber<? super Object> downstream) {
+ source.subscribe(new MappingSubscriber(downstream, mapper))
+ }
+ }
+
+ private static final class MappingSubscriber implements Flow.Subscriber<Object> {
+ private final Flow.Subscriber<? super Object> downstream
+ private final Closure mapper
+
+ MappingSubscriber(Flow.Subscriber<? super Object> downstream, Closure mapper) {
+ this.downstream = downstream
+ this.mapper = mapper
+ }
+
+ @Override void onSubscribe(Flow.Subscription s) { downstream.onSubscribe(s) }
+
+ @Override
+ void onNext(Object item) {
+ try {
+ downstream.onNext(mapper.call(item))
+ } catch (Throwable t) {
+ downstream.onError(t)
+ }
+ }
+
+ @Override void onError(Throwable t) { downstream.onError(t) }
+
+ @Override void onComplete() { downstream.onComplete() }
Review Comment:
MappingSubscriber calls downstream.onError(t) when the mapper throws, but it doesn't cancel the upstream subscription (and it doesn't retain a reference to it). That can leave the upstream producing and potentially continue calling onNext after the downstream has already seen a terminal signal. Consider storing the subscription and cancelling it when a mapping error occurs (and/or guarding against multiple terminal signals).
```suggestion
private final AtomicReference<Flow.Subscription> subscriptionRef = new AtomicReference<>()
private volatile boolean done
MappingSubscriber(Flow.Subscriber<? super Object> downstream, Closure mapper) {
this.downstream = downstream
this.mapper = mapper
}
@Override
void onSubscribe(Flow.Subscription s) {
subscriptionRef.set(s)
downstream.onSubscribe(s)
}
@Override
void onNext(Object item) {
if (done) return
try {
downstream.onNext(mapper.call(item))
} catch (Throwable t) {
done = true
Flow.Subscription subscription = subscriptionRef.get()
if (subscription != null) {
subscription.cancel()
}
downstream.onError(t)
}
}
@Override
void onError(Throwable t) {
if (done) return
done = true
downstream.onError(t)
}
@Override
void onComplete() {
if (done) return
done = true
downstream.onComplete()
}
```
##########
subprojects/groovy-http-builder/src/main/groovy/groovy/http/HttpStreamResult.groovy:
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 groovy.http
+
+import org.apache.groovy.lang.annotation.Incubating
+
+import java.net.http.HttpHeaders
+import java.net.http.HttpResponse
+import java.nio.ByteBuffer
+import java.nio.charset.Charset
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.Flow
+import java.util.concurrent.SubmissionPublisher
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * Streaming response wrapper for {@link HttpBuilder}'s {@code streamAsync} family.
+ * <p>
+ * Unlike {@link HttpResult}, this does not buffer the response body. The body
+ * is exposed as a {@link Flow.Publisher} of byte chunks that can be consumed
+ * incrementally — typically via {@code for await (chunk in result.bodyAsPublisher())}
+ * once the {@code groovy-concurrent-java} {@code FlowPublisherAdapter} is on
+ * the classpath.
+ */
+@Incubating
+record HttpStreamResult(
+ int status,
+ HttpHeaders headers,
+ HttpResponse<Flow.Publisher<List<ByteBuffer>>> raw) {
+
+ HttpStreamResult(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {
+ this(response.statusCode(), response.headers(), response)
+ }
+
+ /**
+ * Returns the response body as a publisher of {@code List<ByteBuffer>}
+ * chunks, exactly as JDK {@code HttpResponse.BodyHandlers.ofPublisher()}
+ * provides it.
+ */
+ Flow.Publisher<List<ByteBuffer>> bodyAsPublisher() {
+ return raw.body()
+ }
+
+ /**
+ * Returns the response body as a publisher of decoded strings, one per
+ * arriving chunk. Each emitted string corresponds to one upstream HTTP
+ * frame (often a complete SSE event or a JSON line, but boundaries are
+ * not guaranteed). Use {@link #bodyAsLinePublisher} for line-aligned
+ * delivery.
+ *
+ * @param charset the charset to decode bytes with (defaults to UTF-8)
+ */
+ Flow.Publisher<String> bodyAsTextPublisher(Charset charset = StandardCharsets.UTF_8) {
+ Closure mapper = { chunk ->
+ StringBuilder sb = new StringBuilder()
+ for (ByteBuffer buf : (List<ByteBuffer>) chunk) {
+ sb.append(charset.decode(buf))
+ }
+ sb.toString()
+ }
+ return (Flow.Publisher<String>) new MappedPublisher(raw.body(), mapper)
+ }
+
+ /**
+ * Returns the response body as a publisher of newline-terminated strings.
+ * Useful for line-oriented streaming protocols (NDJSON, SSE without event
+ * framing, log tails). Trailing partial lines are emitted on completion.
+ *
+ * @param charset the charset to decode bytes with (defaults to UTF-8)
+ */
+ Flow.Publisher<String> bodyAsLinePublisher(Charset charset = StandardCharsets.UTF_8) {
+ return new LinePublisher(raw.body(), charset)
+ }
+
+ /**
+ * One-shot mapping {@link Flow.Publisher} that delegates subscription to a
+ * source publisher and applies a transformer per signalled item.
+ */
+ private static final class MappedPublisher implements Flow.Publisher<Object> {
+ private final Flow.Publisher source
+ private final Closure mapper
+
+ MappedPublisher(Flow.Publisher source, Closure mapper) {
+ this.source = source
+ this.mapper = mapper
+ }
+
+ @Override
+ void subscribe(Flow.Subscriber<? super Object> downstream) {
+ source.subscribe(new MappingSubscriber(downstream, mapper))
+ }
+ }
+
+ private static final class MappingSubscriber implements Flow.Subscriber<Object> {
+ private final Flow.Subscriber<? super Object> downstream
+ private final Closure mapper
+
+ MappingSubscriber(Flow.Subscriber<? super Object> downstream, Closure mapper) {
+ this.downstream = downstream
+ this.mapper = mapper
+ }
+
+ @Override void onSubscribe(Flow.Subscription s) { downstream.onSubscribe(s) }
+
+ @Override
+ void onNext(Object item) {
+ try {
+ downstream.onNext(mapper.call(item))
+ } catch (Throwable t) {
+ downstream.onError(t)
+ }
+ }
+
+ @Override void onError(Throwable t) { downstream.onError(t) }
+
+ @Override void onComplete() { downstream.onComplete() }
+ }
+
+ /**
+ * Adapts a chunked byte publisher into a publisher of complete lines.
+ * Buffers across chunk boundaries so multi-chunk lines emit correctly.
+ */
+ private static final class LinePublisher implements Flow.Publisher<String> {
+ private final Flow.Publisher<List<ByteBuffer>> source
+ private final Charset charset
+
+ LinePublisher(Flow.Publisher<List<ByteBuffer>> source, Charset charset) {
+ this.source = source
+ this.charset = charset
+ }
+
+ @Override
+ void subscribe(Flow.Subscriber<? super String> downstream) {
+ StringBuilder pending = new StringBuilder()
+ AtomicReference<Flow.Subscription> subRef = new AtomicReference<>()
+ source.subscribe(new Flow.Subscriber<List<ByteBuffer>>() {
+ @Override
+ void onSubscribe(Flow.Subscription s) {
+ subRef.set(s)
+ downstream.onSubscribe(s)
Review Comment:
The AtomicReference subRef is assigned in onSubscribe but never used, and SubmissionPublisher is imported but unused. Removing unused state/imports will reduce noise (or, if you intended to use subRef for cancellation/backpressure, wire it into the implementation).
##########
subprojects/groovy-http-builder/src/main/groovy/groovy/http/HttpStreamResult.groovy:
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 groovy.http
+
+import org.apache.groovy.lang.annotation.Incubating
+
+import java.net.http.HttpHeaders
+import java.net.http.HttpResponse
+import java.nio.ByteBuffer
+import java.nio.charset.Charset
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.Flow
+import java.util.concurrent.SubmissionPublisher
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * Streaming response wrapper for {@link HttpBuilder}'s {@code streamAsync} family.
+ * <p>
+ * Unlike {@link HttpResult}, this does not buffer the response body. The body
+ * is exposed as a {@link Flow.Publisher} of byte chunks that can be consumed
+ * incrementally — typically via {@code for await (chunk in result.bodyAsPublisher())}
+ * once the {@code groovy-concurrent-java} {@code FlowPublisherAdapter} is on
+ * the classpath.
+ */
+@Incubating
+record HttpStreamResult(
+ int status,
+ HttpHeaders headers,
+ HttpResponse<Flow.Publisher<List<ByteBuffer>>> raw) {
+
+ HttpStreamResult(HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {
+ this(response.statusCode(), response.headers(), response)
+ }
+
+ /**
+ * Returns the response body as a publisher of {@code List<ByteBuffer>}
+ * chunks, exactly as JDK {@code HttpResponse.BodyHandlers.ofPublisher()}
+ * provides it.
+ */
+ Flow.Publisher<List<ByteBuffer>> bodyAsPublisher() {
+ return raw.body()
+ }
+
+ /**
+ * Returns the response body as a publisher of decoded strings, one per
+ * arriving chunk. Each emitted string corresponds to one upstream HTTP
+ * frame (often a complete SSE event or a JSON line, but boundaries are
+ * not guaranteed). Use {@link #bodyAsLinePublisher} for line-aligned
+ * delivery.
+ *
+ * @param charset the charset to decode bytes with (defaults to UTF-8)
+ */
+ Flow.Publisher<String> bodyAsTextPublisher(Charset charset = StandardCharsets.UTF_8) {
+ Closure mapper = { chunk ->
+ StringBuilder sb = new StringBuilder()
+ for (ByteBuffer buf : (List<ByteBuffer>) chunk) {
+ sb.append(charset.decode(buf))
+ }
+ sb.toString()
+ }
+ return (Flow.Publisher<String>) new MappedPublisher(raw.body(), mapper)
+ }
+
+ /**
+ * Returns the response body as a publisher of newline-terminated strings.
+ * Useful for line-oriented streaming protocols (NDJSON, SSE without event
+ * framing, log tails). Trailing partial lines are emitted on completion.
+ *
+ * @param charset the charset to decode bytes with (defaults to UTF-8)
+ */
+ Flow.Publisher<String> bodyAsLinePublisher(Charset charset = StandardCharsets.UTF_8) {
+ return new LinePublisher(raw.body(), charset)
+ }
+
+ /**
+ * One-shot mapping {@link Flow.Publisher} that delegates subscription to a
+ * source publisher and applies a transformer per signalled item.
+ */
+ private static final class MappedPublisher implements Flow.Publisher<Object> {
+ private final Flow.Publisher source
+ private final Closure mapper
+
+ MappedPublisher(Flow.Publisher source, Closure mapper) {
+ this.source = source
+ this.mapper = mapper
+ }
+
+ @Override
+ void subscribe(Flow.Subscriber<? super Object> downstream) {
+ source.subscribe(new MappingSubscriber(downstream, mapper))
+ }
+ }
+
+ private static final class MappingSubscriber implements Flow.Subscriber<Object> {
+ private final Flow.Subscriber<? super Object> downstream
+ private final Closure mapper
+
+ MappingSubscriber(Flow.Subscriber<? super Object> downstream, Closure mapper) {
+ this.downstream = downstream
+ this.mapper = mapper
+ }
+
+ @Override void onSubscribe(Flow.Subscription s) { downstream.onSubscribe(s) }
+
+ @Override
+ void onNext(Object item) {
+ try {
+ downstream.onNext(mapper.call(item))
+ } catch (Throwable t) {
+ downstream.onError(t)
+ }
+ }
+
+ @Override void onError(Throwable t) { downstream.onError(t) }
+
+ @Override void onComplete() { downstream.onComplete() }
+ }
+
+ /**
+ * Adapts a chunked byte publisher into a publisher of complete lines.
+ * Buffers across chunk boundaries so multi-chunk lines emit correctly.
+ */
+ private static final class LinePublisher implements Flow.Publisher<String> {
+ private final Flow.Publisher<List<ByteBuffer>> source
+ private final Charset charset
+
+ LinePublisher(Flow.Publisher<List<ByteBuffer>> source, Charset charset) {
+ this.source = source
+ this.charset = charset
+ }
+
+ @Override
+ void subscribe(Flow.Subscriber<? super String> downstream) {
+ StringBuilder pending = new StringBuilder()
+ AtomicReference<Flow.Subscription> subRef = new AtomicReference<>()
+ source.subscribe(new Flow.Subscriber<List<ByteBuffer>>() {
+ @Override
+ void onSubscribe(Flow.Subscription s) {
+ subRef.set(s)
+ downstream.onSubscribe(s)
+ }
+
+ @Override
+ void onNext(List<ByteBuffer> chunk) {
+ for (ByteBuffer buf : chunk) {
+ pending.append(charset.decode(buf))
+ }
+ int newline
+ while ((newline = pending.indexOf('\n')) != -1) {
+ String line = pending.substring(0, newline)
+ if (line.endsWith('\r')) line = line.substring(0, line.length() - 1)
+ pending.delete(0, newline + 1)
+ downstream.onNext(line)
+ }
Review Comment:
LinePublisher passes the upstream Subscription through to downstream and can emit multiple lines per upstream chunk in a single onNext. This can violate Reactive Streams demand/backpressure (downstream may request N but receive >N onNext calls). Consider implementing a proper mapping Subscription that tracks downstream demand and buffers/drains lines accordingly (or otherwise ensures you never call downstream.onNext more times than requested).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GROOVY-11933: Mesh async and http builder via Flow.Publisher [groovy]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #2470:
URL: https://github.com/apache/groovy/pull/2470#issuecomment-4266330920
## [Codecov](https://app.codecov.io/gh/apache/groovy/pull/2470?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `60.43956%` with `72 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 67.0837%. Comparing base ([`a82fef4`](https://app.codecov.io/gh/apache/groovy/commit/a82fef4f1fe180ed8a8e9c131dae698f57af6d01?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`8af797d`](https://app.codecov.io/gh/apache/groovy/commit/8af797d2048a214bc7bd5e9345673f672a9d5028?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/groovy/pull/2470?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...che/groovy/runtime/async/FlowPublisherAdapter.java](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&filepath=src%2Fmain%2Fjava%2Forg%2Fapache%2Fgroovy%2Fruntime%2Fasync%2FFlowPublisherAdapter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dyb292eS9ydW50aW1lL2FzeW5jL0Zsb3dQdWJsaXNoZXJBZGFwdGVyLmphdmE=) | 69.1589% | [19 Missing and 14 partials :warning: ](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rc/main/groovy/groovy/http/HttpStreamResult.groovy](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&filepath=subprojects%2Fgroovy-http-builder%2Fsrc%2Fmain%2Fgroovy%2Fgroovy%2Fhttp%2FHttpStreamResult.groovy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3VicHJvamVjdHMvZ3Jvb3Z5LWh0dHAtYnVpbGRlci9zcmMvbWFpbi9ncm9vdnkvZ3Jvb3Z5L2h0dHAvSHR0cFN0cmVhbVJlc3VsdC5ncm9vdnk=) | 46.1538% | [25 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...der/src/main/groovy/groovy/http/HttpBuilder.groovy](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&filepath=subprojects%2Fgroovy-http-builder%2Fsrc%2Fmain%2Fgroovy%2Fgroovy%2Fhttp%2FHttpBuilder.groovy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3VicHJvamVjdHMvZ3Jvb3Z5LWh0dHAtYnVpbGRlci9zcmMvbWFpbi9ncm9vdnkvZ3Jvb3Z5L2h0dHAvSHR0cEJ1aWxkZXIuZ3Jvb3Z5) | 50.0000% | [8 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #2470 +/- ##
==================================================
- Coverage 67.1025% 67.0837% -0.0188%
- Complexity 30871 30883 +12
==================================================
Files 1435 1437 +2
Lines 119620 119813 +193
Branches 21182 21210 +28
==================================================
+ Hits 80268 80375 +107
- Misses 32664 32729 +65
- Partials 6688 6709 +21
```
| [Files with missing lines](https://app.codecov.io/gh/apache/groovy/pull/2470?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...va/groovy/concurrent/AwaitableAdapterRegistry.java](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&filepath=src%2Fmain%2Fjava%2Fgroovy%2Fconcurrent%2FAwaitableAdapterRegistry.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3JjL21haW4vamF2YS9ncm9vdnkvY29uY3VycmVudC9Bd2FpdGFibGVBZGFwdGVyUmVnaXN0cnkuamF2YQ==) | `73.0159% <100.0000%> (+0.4352%)` | :arrow_up: |
| [...der/src/main/groovy/groovy/http/HttpBuilder.groovy](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&filepath=subprojects%2Fgroovy-http-builder%2Fsrc%2Fmain%2Fgroovy%2Fgroovy%2Fhttp%2FHttpBuilder.groovy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3VicHJvamVjdHMvZ3Jvb3Z5LWh0dHAtYnVpbGRlci9zcmMvbWFpbi9ncm9vdnkvZ3Jvb3Z5L2h0dHAvSHR0cEJ1aWxkZXIuZ3Jvb3Z5) | `71.5152% <50.0000%> (-3.3100%)` | :arrow_down: |
| [...rc/main/groovy/groovy/http/HttpStreamResult.groovy](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&filepath=subprojects%2Fgroovy-http-builder%2Fsrc%2Fmain%2Fgroovy%2Fgroovy%2Fhttp%2FHttpStreamResult.groovy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3VicHJvamVjdHMvZ3Jvb3Z5LWh0dHAtYnVpbGRlci9zcmMvbWFpbi9ncm9vdnkvZ3Jvb3Z5L2h0dHAvSHR0cFN0cmVhbVJlc3VsdC5ncm9vdnk=) | `46.1538% <46.1538%> (ø)` | |
| [...che/groovy/runtime/async/FlowPublisherAdapter.java](https://app.codecov.io/gh/apache/groovy/pull/2470?src=pr&el=tree&filepath=src%2Fmain%2Fjava%2Forg%2Fapache%2Fgroovy%2Fruntime%2Fasync%2FFlowPublisherAdapter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dyb292eS9ydW50aW1lL2FzeW5jL0Zsb3dQdWJsaXNoZXJBZGFwdGVyLmphdmE=) | `69.1589% <69.1589%> (ø)` | |
... and [3 files with indirect coverage changes](https://app.codecov.io/gh/apache/groovy/pull/2470/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@groovy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org