BufferFlux and BufferMono do not depend on netty-codec-http anymore (#…

…2517)

Add API for providing a function for extracting a Buffer from an Object

Related to #1953

reactor-netty5-core/src/main/java/reactor/netty5/BufferFlux.java

@@ -21,8 +21,6 @@
 import io.netty5.buffer.CompositeBuffer;
 import io.netty5.util.Send;
 import io.netty5.channel.socket.DatagramPacket;
-import io.netty5.handler.codec.http.HttpContent;
-import io.netty5.handler.codec.http.websocketx.WebSocketFrame;
 import org.reactivestreams.Publisher;
 import reactor.core.CoreSubscriber;
 import reactor.core.Fuseable;
@@ -68,6 +66,18 @@ public static BufferFlux fromInbound(Publisher<?> source) {
 	 * @return a {@link BufferFlux}
 	 */
 	public static BufferFlux fromInbound(Publisher<?> source, BufferAllocator allocator) {
+		return fromInbound(source, allocator, bufferExtractorFunction);
+	}
+
+	/**
+	 * Decorate as {@link BufferFlux}
+	 *
+	 * @param source publisher to decorate
+	 * @param allocator the channel {@link BufferAllocator}
+	 * @param bufferExtractor a function that extracts a {@link Buffer} from the given {@link Object}
+	 * @return a {@link BufferFlux}
+	 */
+	public static BufferFlux fromInbound(Publisher<?> source, BufferAllocator allocator, Function<Object, Buffer> bufferExtractor) {
 		Objects.requireNonNull(allocator, "allocator");
 		return maybeFuse(Flux.from(ReactorNetty.publisherOrScalarMap(source, bufferExtractor)), allocator);
 	}
@@ -307,19 +317,13 @@ static BufferFlux maybeFuse(Flux<Buffer> source, BufferAllocator allocator) {
 	/**
 	 * A channel object to {@link Buffer} transformer
 	 */
-	static final Function<Object, Buffer> bufferExtractor = o -> {
+	static final Function<Object, Buffer> bufferExtractorFunction = o -> {
 		if (o instanceof Buffer buffer) {
 			return buffer;
 		}
 		if (o instanceof DatagramPacket envelope) {
 			return envelope.content();
 		}
-		if (o instanceof HttpContent<?> httpContent) {
-			return httpContent.payload();
-		}
-		if (o instanceof WebSocketFrame frame) {
-			return frame.binaryData();
-		}
 		if (o instanceof byte[] bytes) {
 			return preferredAllocator().copyOf(bytes);
 		}

reactor-netty5-core/src/main/java/reactor/netty5/BufferMono.java

@@ -155,7 +155,7 @@ public void subscribe(CoreSubscriber<? super Buffer> actual) {
 	}
 
 	BufferMono(Mono<?> source) {
-		super(source.map(BufferFlux.bufferExtractor));
+		super(source.map(BufferFlux.bufferExtractorFunction));
 	}
 
 	static BufferMono maybeFuse(Mono<?> source) {

reactor-netty5-http/src/main/java/reactor/netty5/http/HttpOperations.java

@@ -23,12 +23,14 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.regex.Pattern;
 
 import io.netty5.buffer.Buffer;
 import io.netty5.buffer.BufferAllocator;
 import io.netty5.handler.codec.http.headers.HttpHeaders;
+import io.netty5.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty5.util.Resource;
 import io.netty5.channel.ChannelHandler;
 import io.netty5.channel.CombinedChannelDuplexHandler;
@@ -43,6 +45,7 @@
 import io.netty5.handler.codec.http.LastHttpContent;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
+import reactor.netty5.BufferFlux;
 import reactor.netty5.Connection;
 import reactor.netty5.ConnectionObserver;
 import reactor.netty5.NettyInbound;
@@ -55,6 +58,7 @@
 import reactor.util.Loggers;
 import reactor.util.annotation.Nullable;
 
+import static io.netty5.buffer.DefaultBufferAllocators.preferredAllocator;
 import static reactor.netty5.ReactorNetty.format;
 import static reactor.netty5.ReactorNetty.toPrettyHexDump;
 
@@ -101,6 +105,11 @@ public String requestId() {
 		return asShortText();
 	}
 
+	@Override
+	public BufferFlux receive() {
+		return BufferFlux.fromInbound(receiveObject(), alloc(), bufferExtractorFunction);
+	}
+
 	@Override
 	@SuppressWarnings("unchecked")
 	public NettyOutbound send(Publisher<? extends Buffer> source) {
@@ -434,6 +443,25 @@ else if (!SCHEME_PATTERN.matcher(tempUri).matches()) {
 			}
 	);
 
+	/**
+	 * A channel object to {@link Buffer} transformer
+	 */
+	static final Function<Object, Buffer> bufferExtractorFunction = o -> {
+		if (o instanceof Buffer buffer) {
+			return buffer;
+		}
+		if (o instanceof HttpContent<?> httpContent) {
+			return httpContent.payload();
+		}
+		if (o instanceof WebSocketFrame frame) {
+			return frame.binaryData();
+		}
+		if (o instanceof byte[] bytes) {
+			return preferredAllocator().copyOf(bytes);
+		}
+		throw new IllegalArgumentException("Object " + o + " of type " + o.getClass() + " " + "cannot be converted to Buffer");
+	};
+
 	static final Logger log = Loggers.getLogger(HttpOperations.class);
 
 	static final Pattern SCHEME_PATTERN = Pattern.compile("^(https?|wss?)://.*$");

reactor-netty5-http/src/test/java/reactor/netty5/http/server/HttpServerTests.java

@@ -309,7 +309,8 @@ void httpPipelining() throws Exception {
 				                         x.addHandlerFirst(new HttpClientCodec()))
 				                   .receiveObject()
 				                   .ofType(HttpContent.class)
-				                   .as(BufferFlux::fromInbound)
+				                   .as(httpContentFlux ->
+				                       BufferFlux.fromInbound(httpContentFlux, preferredAllocator(), o -> ((HttpContent<?>) o).payload()))
 				                   .asString()
 				                   .log()
 				                   .map(Integer::parseInt)