Fix warning when sending already-available streaming body (#11665)

Fixes #11603

Author: yawkat

http-client/src/main/java/io/micronaut/http/client/netty/StreamWriter.java

@@ -16,6 +16,7 @@
 package io.micronaut.http.client.netty;
 
 import io.micronaut.core.annotation.Internal;
+import io.micronaut.http.body.stream.LazyUpstream;
 import io.micronaut.http.netty.EventLoopFlow;
 import io.micronaut.http.netty.body.ByteBufConsumer;
 import io.micronaut.http.netty.body.StreamingNettyByteBody;
@@ -61,7 +62,12 @@ void startWriting() {
         if (ctx == null) {
             throw new IllegalStateException("Not added to a channel yet");
         }
+        LazyUpstream lazyUpstream = new LazyUpstream();
+        // primary() can call other methods here immediately, so we need a replacement upstream
+        // until it returns
+        upstream = lazyUpstream;
         upstream = body.primary(this);
+        lazyUpstream.forward(upstream);
         try {
             upstream.start();
         } catch (Exception e) {

http-client/src/test/groovy/io/micronaut/http/client/RawHttpClientSpec.groovy

@@ -0,0 +1,62 @@
+package io.micronaut.http.client
+
+import io.micronaut.context.ApplicationContext
+import io.micronaut.context.annotation.Requires
+import io.micronaut.http.ByteBodyHttpResponse
+import io.micronaut.http.HttpRequest
+import io.micronaut.http.annotation.Body
+import io.micronaut.http.annotation.Controller
+import io.micronaut.http.annotation.Post
+import io.micronaut.http.body.ByteBody
+import io.micronaut.http.netty.body.NettyBodyAdapter
+import io.micronaut.runtime.server.EmbeddedServer
+import io.netty.buffer.Unpooled
+import io.netty.channel.embedded.EmbeddedChannel
+import jakarta.inject.Singleton
+import reactor.core.publisher.Mono
+import spock.lang.Issue
+import spock.lang.Specification
+
+import java.nio.charset.StandardCharsets
+
+class RawHttpClientSpec extends Specification {
+    @Issue("https://github.com/micronaut-projects/micronaut-core/issues/11603")
+    def 'immediate stream'() {
+        given:
+        def ctx = ApplicationContext.run(['spec.name': 'RawHttpClientSpec'])
+        def server = ctx.getBean(EmbeddedServer)
+        server.start()
+        def client = ctx.createBean(RawHttpClient, server.URI)
+
+        def requestBody = NettyBodyAdapter.adapt(
+                Mono.just(Unpooled.copiedBuffer("foo", StandardCharsets.UTF_8)),
+                new EmbeddedChannel().eventLoop())
+        requestBody.split(ByteBody.SplitBackpressureMode.FASTEST).buffer().get().close()
+
+        when:
+        def resp = Mono.from(client.exchange(
+                HttpRequest.POST(server.URI.toString() + "/raw/echo", null),
+                requestBody,
+                null
+        )).block()
+        then:
+        resp instanceof ByteBodyHttpResponse<?>
+        ByteBody body = resp.byteBody()
+        body.buffer().get().toString(StandardCharsets.UTF_8) == "foo"
+
+        cleanup:
+        resp.close()
+        server.close()
+        ctx.close()
+    }
+
+    @Singleton
+    @Controller("/raw")
+    @Requires(property = "spec.name", value = "RawHttpClientSpec")
+    static class Ctrl {
+        @Post("/echo")
+        String echo(@Body String body) {
+            return body
+        }
+    }
+}

http/src/main/java/io/micronaut/http/body/stream/LazyUpstream.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2017-2025 original authors
+ *
+ * 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
+ *
+ * https://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.micronaut.http.body.stream;
+
+import io.micronaut.core.annotation.Internal;
+
+/**
+ * {@link io.micronaut.http.body.stream.BufferConsumer.Upstream} implementation that stores any
+ * inputs and then {@link #forward forwards} them to another upstream later on. This helps with
+ * reentrant calls to the subscriber during a subscribe call, when the real upstream is not yet
+ * available.
+ *
+ * @since 4.8.0
+ * @author Jonas Konrad
+ */
+@Internal
+public final class LazyUpstream implements BufferConsumer.Upstream {
+    private boolean start;
+    private boolean allowDiscard;
+    private boolean disregardBackpressure;
+    private long consumed;
+
+    @Override
+    public void start() {
+        this.start = true;
+    }
+
+    @Override
+    public void onBytesConsumed(long bytesConsumed) {
+        long newConsumed = consumed + bytesConsumed;
+        if (newConsumed < 0) {
+            newConsumed = Long.MAX_VALUE;
+        }
+        consumed = newConsumed;
+    }
+
+    @Override
+    public void allowDiscard() {
+        this.allowDiscard = true;
+    }
+
+    @Override
+    public void disregardBackpressure() {
+        this.disregardBackpressure = true;
+    }
+
+    public void forward(BufferConsumer.Upstream actual) {
+        if (consumed != 0) {
+            actual.onBytesConsumed(consumed);
+        }
+        if (start) {
+            actual.start();
+        }
+        if (allowDiscard) {
+            actual.allowDiscard();
+        }
+        if (disregardBackpressure) {
+            actual.disregardBackpressure();
+        }
+    }
+}