Added Mono.blockOption

Author: sinwe

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

@@ -142,6 +142,43 @@ class Mono[T] private(private val jMono: JMono[T]) extends Publisher[T] with Map
     */
   final def block(timeout: Duration): T = jMono.block(timeout)
 
+  /**
+    * Subscribe to this {[[Mono]] Mono} and <strong>block indefinitely</strong> until a next signal is
+    * received or the Mono completes empty. Returns an [[Option]], which can be used
+    * to replace the empty case with an Exception via [[Option.orElse(throw exception)]].
+    * In case the Mono itself errors, the original exception is thrown (wrapped in a
+    * [[RuntimeException]] if it was a checked exception).
+    *
+    * <p>
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.1.RELEASE/src/docs/marble/block.png" alt="">
+    * <p>
+    * Note that each blockOptional() will trigger a new subscription: in other words, the result
+    * might miss signal from hot publishers.
+    *
+    * @return T the result
+    */
+  final def blockOption(): Option[T] = jMono.blockOptional()
+
+  /**
+    * Subscribe to this [[Mono]] and <strong>block</strong> until a next signal is
+    * received, the Mono completes empty or a timeout expires. Returns an [[Option]]
+    * for the first two cases, which can be used to replace the empty case with an
+    * Exception via [[Option.orElse(throw exception)]].
+    * In case the Mono itself errors, the original exception is thrown (wrapped in a
+    * [[RuntimeException]] if it was a checked exception).
+    * If the provided timeout expires, a [[RuntimeException]] is thrown.
+    *
+    * <p>
+    * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.1.RELEASE/src/docs/marble/block.png" alt="">
+    * <p>
+    * Note that each block() will trigger a new subscription: in other words, the result
+    * might miss signal from hot publishers.
+    *
+    * @param timeout maximum time period to wait for before raising a [[RuntimeException]]
+    * @return T the result
+    */
+  final def blockOption(timeout: Duration): Option[T] = jMono.blockOptional(timeout)
+
   /**
     * Cast the current [[Mono]] produced type into a target produced type.
     *

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

@@ -159,4 +159,6 @@ Uncomment this when used. It is not used for now and reduce the code coverage
   implicit def scalaBiPredicate2JavaBiPredicate[T, U](scalaBiPredicate: (T, U) => Boolean): BiPredicate[T, U] = new BiPredicate[T, U] {
     override def test(t: T, u: U) = scalaBiPredicate(t, u)
   }
+
+  implicit def javaOptional2ScalaOption[T](jOptional: Optional[T]): Option[T] = if(jOptional.isPresent) Some(jOptional.get()) else None
 }

src/test/scala/reactor/core/scala/publisher/MonoTest.scala

@@ -407,6 +407,28 @@ class MonoTest extends FreeSpec with Matchers with TableDrivenPropertyChecks wit
       }
     }
 
+    ".blockOption" - {
+      "without duration" - {
+        "should block the mono to get value" in {
+          Mono.just(randomValue).blockOption() shouldBe Some(randomValue)
+        }
+        "should retun None if mono is empty" in {
+          Mono.empty.blockOption() shouldBe None
+        }
+      }
+      "with duration" - {
+        "should block the mono up to the duration" in {
+          Mono.just(randomValue).blockOption(10 seconds) shouldBe Some(randomValue)
+        }
+        "shouldBlock the mono up to the duration and return None" in {
+          StepVerifier.withVirtualTime(() => Mono.just(Mono.empty.blockOption(10 seconds)))
+            .thenAwait(10 seconds)
+            .expectNext(None)
+            .verifyComplete()
+        }
+      }
+    }
+
     ".cast should cast the underlying value" in {
       val number = Mono.just(BigDecimal("123")).cast(classOf[ScalaNumber]).block()
       number shouldBe a[ScalaNumber]