DataBufferUtils and BodyInserters: allow consumers of OutputStream to throw IOException

spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java

@@ -439,22 +439,22 @@ static void closeChannel(@Nullable Channel channel) {
 	 * @since 6.1
 	 */
 	public static Publisher<DataBuffer> outputStreamPublisher(
-			Consumer<OutputStream> consumer, DataBufferFactory bufferFactory, Executor executor) {
+			OutputStreamHandler consumer, DataBufferFactory bufferFactory, Executor executor) {
 
 		return new OutputStreamPublisher<>(
-				consumer::accept, new DataBufferMapper(bufferFactory), executor, null);
+				consumer, new DataBufferMapper(bufferFactory), executor, null);
 	}
 
 	/**
-	 * Variant of {@link #outputStreamPublisher(Consumer, DataBufferFactory, Executor)}
+	 * Variant of {@link #outputStreamPublisher(OutputStreamHandler, DataBufferFactory, Executor)}
 	 * providing control over the chunk sizes to be produced by the publisher.
 	 * @since 6.1
 	 */
 	public static Publisher<DataBuffer> outputStreamPublisher(
-			Consumer<OutputStream> consumer, DataBufferFactory bufferFactory, Executor executor, int chunkSize) {
+			OutputStreamHandler consumer, DataBufferFactory bufferFactory, Executor executor, int chunkSize) {
 
 		return new OutputStreamPublisher<>(
-				consumer::accept, new DataBufferMapper(bufferFactory), executor, chunkSize);
+				consumer, new DataBufferMapper(bufferFactory), executor, chunkSize);
 	}
 
 	/**

spring-core/src/main/java/org/springframework/core/io/buffer/OutputStreamHandler.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2002-2024 the original author or 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 org.springframework.core.io.buffer;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Contract to provide callback access to the {@link OutputStream}.
+ * @author Rossen Stoyanchev
+ * @author Federico Pettinari
+ */
+@FunctionalInterface
+public interface OutputStreamHandler {
+
+	void handle(OutputStream outputStream) throws IOException;
+
+}

spring-core/src/main/java/org/springframework/core/io/buffer/OutputStreamPublisher.java

@@ -99,17 +99,6 @@ public void subscribe(Subscriber<? super T> subscriber) {
 	}
 
 
-	/**
-	 * Contract to provide callback access to the {@link OutputStream}.
-	 */
-	@FunctionalInterface
-	public interface OutputStreamHandler {
-
-		void handle(OutputStream outputStream) throws Exception;
-
-	}
-
-
 	/**
 	 * Maps from bytes to byte buffers.
 	 * @param <T> the type of byte buffer to map to

spring-webflux/src/main/java/org/springframework/web/reactive/function/BodyInserters.java

@@ -19,7 +19,6 @@
 import java.io.OutputStream;
 import java.util.List;
 import java.util.concurrent.Executor;
-import java.util.function.Consumer;
 
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
@@ -32,6 +31,7 @@
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.OutputStreamHandler;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.MediaType;
 import org.springframework.http.ReactiveHttpOutputMessage;
@@ -397,10 +397,10 @@ public static <T extends Publisher<DataBuffer>> BodyInserter<T, ReactiveHttpOutp
 	 * separate thread
 	 * @return an inserter that writes what is written to the output stream
 	 * @since 6.1
-	 * @see DataBufferUtils#outputStreamPublisher(Consumer, DataBufferFactory, Executor)
+	 * @see DataBufferUtils#outputStreamPublisher(OutputStreamHandler, DataBufferFactory, Executor)
 	 */
 	public static <T extends Publisher<DataBuffer>> BodyInserter<T, ReactiveHttpOutputMessage> fromOutputStream(
-			Consumer<OutputStream> outputStreamConsumer, Executor executor) {
+			OutputStreamHandler outputStreamConsumer, Executor executor) {
 
 		Assert.notNull(outputStreamConsumer, "OutputStreamConsumer must not be null");
 		Assert.notNull(executor, "Executor must not be null");
@@ -418,10 +418,10 @@ public static <T extends Publisher<DataBuffer>> BodyInserter<T, ReactiveHttpOutp
 	 * @param chunkSize minimum size of the buffer produced by the publisher
 	 * @return an inserter that writes what is written to the output stream
 	 * @since 6.1
-	 * @see DataBufferUtils#outputStreamPublisher(Consumer, DataBufferFactory, Executor, int)
+	 * @see DataBufferUtils#outputStreamPublisher(OutputStreamHandler, DataBufferFactory, Executor, int)
 	 */
 	public static <T extends Publisher<DataBuffer>> BodyInserter<T, ReactiveHttpOutputMessage> fromOutputStream(
-			Consumer<OutputStream> outputStreamConsumer, Executor executor, int chunkSize) {
+			OutputStreamHandler outputStreamConsumer, Executor executor, int chunkSize) {
 
 		Assert.notNull(outputStreamConsumer, "OutputStreamConsumer must not be null");
 		Assert.notNull(executor, "Executor must not be null");