Added Mono.using(resourceSupplier, sourceSupplier, resourceCleanup, eager)

Added Mono.using(resourceSupplier, sourceSupplier, resourceCleanup)

Author: sinwe

src/main/scala/reactor/core/scala/publisher/Mono.scala

@@ -1446,6 +1446,55 @@ object Mono {
     override def test(t: T, u: T): Boolean = isEqual(t, u)
   }, bufferSize)).map(Boolean2boolean)
 
+  /**
+    * Uses a 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>Eager resource cleanup happens just before the source termination and exceptions raised by the cleanup
+    * Consumer may override the terminal even.</li> <li>Non-eager cleanup will drop any exception.</li> </ul>
+    * <p>
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.0.6.RELEASE/src/docs/marble/using.png"
+    * alt="">
+    *
+    * @param resourceSupplier a function that is called on subscribe for preparing the resource
+    * @param sourceSupplier a [[Mono]] factory derived from the supplied resource
+    * @param resourceCleanup invoked on completion
+    * @param eager           true to clean before terminating downstream subscribers
+    * @tparam T emitted type
+    * @tparam D resource type
+    * @return new [[Mono]]
+    */
+  def using[T, D](resourceSupplier: () => D, sourceSupplier: D => _ <: Mono[_ <: T], resourceCleanup: D => Unit, eager: Boolean) =
+    Mono(JMono.using[T, D](resourceSupplier, new Function[D, JMono[_ <: T]] {
+      override def apply(t: D): JMono[_ <: T] = sourceSupplier(t).asJava()
+    }, resourceCleanup, eager))
+
+  /**
+    * Uses a 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>
+    * Eager resource cleanup happens just before the source termination and exceptions raised by the cleanup Consumer
+    * may override the terminal even.
+    * <p>
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.0.6.RELEASE/src/docs/marble/using.png"
+    * alt="">
+    *
+    * @param resourceSupplier a function that is called on subscribe to prepare the resource
+    * @param sourceSupplier a [[Mono]] factory derived from the supplied resource
+    * @param resourceCleanup invoked on completion
+    * @tparam T emitted type
+    * @tparam D resource type
+    * @return new [[Mono]]
+    */
+  def using[T, D](resourceSupplier: () => D, sourceSupplier: D => Mono[_ <: T], resourceCleanup: D => Unit) =
+    Mono(JMono.using[T, D](resourceSupplier,  new Function[D, JMono[_ <: T]] {
+      override def apply(t: D): JMono[_ <: T] = sourceSupplier(t).asJava()
+    }, resourceCleanup))
+
   /**
     * Merge given monos into a new a `Mono` that will be fulfilled when all of the given `Monos`
     * have been fulfilled. An error will cause pending results to be cancelled and immediate error emission to the