Add AutoCloseable shourtcut on Flux#using, Mono#using (#3704)

Make `Flux, Mono#using` to close `AutoCloseable` resource,
so that users don't have to pass resourceCleanup consumer.

Fixes #3333.

Author: injae-kim

reactor-core/build.gradle

@@ -256,6 +256,10 @@ task japicmp(type: JapicmpTask) {
 	classExcludes = [
 	]
 	methodExcludes = [
+		"reactor.core.publisher.Flux#using(java.util.concurrent.Callable, java.util.function.Function)",
+		"reactor.core.publisher.Flux#using(java.util.concurrent.Callable, java.util.function.Function, boolean)",
+		"reactor.core.publisher.Mono#using(java.util.concurrent.Callable, java.util.function.Function)",
+		"reactor.core.publisher.Mono#using(java.util.concurrent.Callable, java.util.function.Function, boolean)"
 	]
 }
 

reactor-core/src/main/java/reactor/core/Exceptions.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2024 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@
 import java.util.Objects;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import java.util.function.Consumer;
 
 import reactor.core.publisher.Flux;
 import reactor.util.Logger;
@@ -34,6 +35,7 @@
  * Global Reactor Core Exception handling and utils to operate on.
  *
  * @author Stephane Maldini
+ * @author Injae Kim
  * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
  */
 public abstract class Exceptions {
@@ -861,4 +863,16 @@ static final class StaticThrowable extends Error {
 		}
 	}
 
+	/**
+	 * A general-purpose {@link Consumer} that closes {@link AutoCloseable} resource.
+	 * If exception is thrown during closing the resource, it will be propagated by {@link Exceptions#propagate(Throwable)}.
+	 */
+	public static final Consumer<? super AutoCloseable> AUTO_CLOSE = resource -> {
+		try {
+			resource.close();
+		} catch (Throwable t) {
+			throw Exceptions.propagate(t);
+		}
+	};
+
 }

reactor-core/src/main/java/reactor/core/publisher/Flux.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2024 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -119,6 +119,7 @@
  * @author Stephane Maldini
  * @author David Karnok
  * @author Simon Baslé
+ * @author Injae Kim
  *
  * @see Mono
  */
@@ -2132,6 +2133,64 @@ public static <T, D> Flux<T> using(Callable<? extends D> resourceSupplier, Funct
 				eager));
 	}
 
+	/**
+	 * Uses an {@link AutoCloseable} resource, generated by a supplier for each individual Subscriber,
+	 * while streaming the values from a Publisher derived from the same resource and makes sure
+	 * the resource is released if the sequence terminates or the Subscriber cancels.
+	 * <p>
+	 * Eager {@link AutoCloseable} resource cleanup happens just before the source termination and exceptions raised
+	 * by the cleanup Consumer may override the terminal event.
+	 * <p>
+	 * <img class="marble" src="doc-files/marbles/usingForFlux.svg" alt="">
+	 * <p>
+	 * For an asynchronous version of the cleanup, with distinct path for onComplete, onError
+	 * and cancel terminations, see {@link #usingWhen(Publisher, Function, Function, BiFunction, Function)}.
+	 *
+	 * @param resourceSupplier a {@link Callable} that is called on subscribe to generate the resource
+	 * @param sourceSupplier a factory to derive a {@link Publisher} from the supplied resource
+	 * @param <T> emitted type
+	 * @param <D> resource type
+	 *
+	 * @return a new {@link Flux} built around a disposable resource
+	 * @see #usingWhen(Publisher, Function, Function, BiFunction, Function)
+	 * @see #usingWhen(Publisher, Function, Function)
+	 */
+	public static <T, D extends AutoCloseable> Flux<T> using(Callable<? extends D> resourceSupplier,
+			Function<? super D, ? extends Publisher<? extends T>> sourceSupplier) {
+		return using(resourceSupplier, sourceSupplier, true);
+	}
+
+	/**
+	 * Uses an {@link AutoCloseable} resource, generated by a supplier for each individual Subscriber,
+	 * while streaming the values from a Publisher derived from the same resource and makes sure
+	 * the resource is released if the sequence terminates or the Subscriber cancels.
+	 * <p>
+	 * <ul>
+	 *     <li>Eager {@link AutoCloseable} resource cleanup happens just before the source termination and exceptions raised
+	 *     by the cleanup Consumer may override the terminal event.</li>
+	 *     <li>Non-eager cleanup will drop any exception.</li>
+	 * </ul>
+	 * <p>
+	 * <img class="marble" src="doc-files/marbles/usingForFlux.svg" alt="">
+	 * <p>
+	 * For an asynchronous version of the cleanup, with distinct path for onComplete, onError
+	 * and cancel terminations, see {@link #usingWhen(Publisher, Function, Function, BiFunction, Function)}.
+	 *
+	 * @param resourceSupplier a {@link Callable} that is called on subscribe to generate the resource
+	 * @param sourceSupplier a factory to derive a {@link Publisher} from the supplied resource
+	 * @param eager true to clean before terminating downstream subscribers
+	 * @param <T> emitted type
+	 * @param <D> resource type
+	 *
+	 * @return a new {@link Flux} built around a disposable resource
+	 * @see #usingWhen(Publisher, Function, Function, BiFunction, Function)
+	 * @see #usingWhen(Publisher, Function, Function)
+	 */
+	public static <T, D extends AutoCloseable> Flux<T> using(Callable<? extends D> resourceSupplier,
+			Function<? super D, ? extends Publisher<? extends T>> sourceSupplier, boolean eager) {
+		return using(resourceSupplier, sourceSupplier, Exceptions.AUTO_CLOSE, eager);
+	}
+
 	/**
 	 * Uses a resource, generated by a {@link Publisher} for each individual {@link Subscriber},
 	 * while streaming the values from a {@link Publisher} derived from the same resource.

reactor-core/src/main/java/reactor/core/publisher/Mono.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2024 VMware Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -115,6 +115,7 @@
  * @author Stephane Maldini
  * @author David Karnok
  * @author Simon Baslé
+ * @author Injae Kim
  * @see Flux
  */
 public abstract class Mono<T> implements CorePublisher<T> {
@@ -912,6 +913,56 @@ public static <T, D> Mono<T> using(Callable<? extends D> resourceSupplier,
 		return using(resourceSupplier, sourceSupplier, resourceCleanup, true);
 	}
 
+	/**
+	 * Uses an {@link AutoCloseable} resource, generated by a supplier for each individual Subscriber,
+	 * while streaming the value from a Mono derived from the same resource and makes sure
+	 * the resource is released if the sequence terminates or the Subscriber cancels.
+	 * <p>
+	 * Unlike in {@link Flux#using(Callable, Function, Consumer) Flux}, in the case of a valued {@link Mono} the cleanup
+	 * happens just before passing the value to downstream. In all cases, exceptions raised by the cleanup
+	 * {@link Consumer} may override the terminal event, discarding the element if the derived {@link Mono} was valued.
+	 * <p>
+	 * <img class="marble" src="doc-files/marbles/usingForMono.svg" alt="">
+	 *
+	 * @param resourceSupplier a {@link Callable} that is called on subscribe to create the resource
+	 * @param sourceSupplier a {@link Mono} factory to create the Mono depending on the created resource
+	 * @param <T> emitted type
+	 * @param <D> resource type
+	 *
+	 * @return new {@link Mono}
+	 */
+	public static <T, D extends AutoCloseable> Mono<T> using(Callable<? extends D> resourceSupplier,
+			Function<? super D, ? extends Mono<? extends T>> sourceSupplier) {
+		return using(resourceSupplier, sourceSupplier, true);
+	}
+
+	/**
+	 * Uses an {@link AutoCloseable} resource, generated by a supplier for each individual Subscriber,
+	 * while streaming the value from a Mono derived from the same resource and makes sure
+	 * the resource is released if the sequence terminates or the Subscriber cancels.
+	 * <p>
+	 * <ul>
+	 *     <li>For eager cleanup, Unlike in {@link Flux#using(Callable, Function, Consumer) Flux},
+	 *     in the case of a valued {@link Mono} the cleanup happens just before passing the value to downstream.
+	 *     In all cases, exceptions raised by the cleanup {@link Consumer} may override the terminal event,
+	 *     discarding the element if the derived {@link Mono} was valued.</li>
+	 *     <li>Non-eager cleanup will drop any exception.</li>
+	 * </ul>
+	 * <p>
+	 * <img class="marble" src="doc-files/marbles/usingForMono.svg" alt="">
+	 *
+	 * @param resourceSupplier a {@link Callable} that is called on subscribe to create the resource
+	 * @param sourceSupplier a {@link Mono} factory to create the Mono depending on the created resource
+	 * @param eager set to true to clean before any signal (including onNext) is passed downstream
+	 * @param <T> emitted type
+	 * @param <D> resource type
+	 *
+	 * @return new {@link Mono}
+	 */
+	public static <T, D extends AutoCloseable> Mono<T> using(Callable<? extends D> resourceSupplier,
+    		Function<? super D, ? extends Mono<? extends T>> sourceSupplier, boolean eager) {
+		return using(resourceSupplier, sourceSupplier, Exceptions.AUTO_CLOSE, eager);
+	}
 
 	/**
 	 * Uses a resource, generated by a {@link Publisher} for each individual {@link Subscriber},