optimization: Uni.await()

- optimize the case where the item/failure is known and
subscription/synchronization is not needed

Author: mkouba

implementation/src/main/java/io/smallrye/mutiny/groups/UniAwait.java

@@ -4,12 +4,16 @@
 
 import java.time.Duration;
 import java.util.Optional;
+import java.util.concurrent.CompletionException;
 
 import io.smallrye.common.annotation.CheckReturnValue;
 import io.smallrye.mutiny.Context;
 import io.smallrye.mutiny.TimeoutException;
 import io.smallrye.mutiny.Uni;
+import io.smallrye.mutiny.infrastructure.Infrastructure;
 import io.smallrye.mutiny.operators.uni.UniBlockingAwait;
+import io.smallrye.mutiny.operators.uni.builders.UniCreateFromKnownFailure;
+import io.smallrye.mutiny.operators.uni.builders.UniCreateFromKnownItem;
 
 /**
  * Waits and returns the item emitted by the {@link Uni}. If the {@link Uni} receives a failure, the failure is thrown.
@@ -62,6 +66,13 @@ public T indefinitely() {
      * @return the item from the {@link Uni}, potentially {@code null}
      */
     public T atMost(Duration duration) {
+        // Optimize the case where the item/failure is known
+        // and subscription/synchronization is not needed
+        if (upstream instanceof UniCreateFromKnownItem<T> known) {
+            return awaitKnownItem(known, duration);
+        } else if (upstream instanceof UniCreateFromKnownFailure<T> known) {
+            awaitKnownFailure(known, duration);
+        }
         return UniBlockingAwait.await(upstream, duration, context);
     }
 
@@ -76,4 +87,32 @@ public UniAwaitOptional<T> asOptional() {
         return new UniAwaitOptional<>(upstream, context);
     }
 
+    private T awaitKnownItem(UniCreateFromKnownItem<T> known, Duration duration) {
+        validateDuration(duration);
+        // Blocking should not matter in this case but we retain the check for backward compatibility
+        if (!Infrastructure.canCallerThreadBeBlocked()) {
+            throw UniBlockingAwait.currentThreadCannotBeBlocked();
+        }
+        return known.getItem();
+    }
+
+    private void awaitKnownFailure(UniCreateFromKnownFailure<T> known, Duration duration) {
+        validateDuration(duration);
+        // Blocking should not matter in this case but we retain the check for backward compatibility
+        if (!Infrastructure.canCallerThreadBeBlocked()) {
+            throw UniBlockingAwait.currentThreadCannotBeBlocked();
+        }
+        Throwable throwable = known.getFailure();
+        if (throwable instanceof RuntimeException) {
+            throw (RuntimeException) throwable;
+        }
+        throw new CompletionException(throwable);
+    }
+
+    private void validateDuration(Duration duration) {
+        if (duration != null && (duration.isZero() || duration.isNegative())) {
+            throw new IllegalArgumentException("`duration` must be greater than zero");
+        }
+    }
+
 }

implementation/src/main/java/io/smallrye/mutiny/operators/uni/UniBlockingAwait.java

@@ -27,7 +27,7 @@ public static <T> T await(Uni<T> upstream, Duration duration, Context context) {
         validate(duration);
 
         if (!Infrastructure.canCallerThreadBeBlocked()) {
-            throw new IllegalStateException("The current thread cannot be blocked: " + Thread.currentThread().getName());
+            throw currentThreadCannotBeBlocked();
         }
 
         CountDownLatch latch = new CountDownLatch(1);
@@ -87,6 +87,10 @@ public void onFailure(Throwable failure) {
         }
     }
 
+    public static IllegalStateException currentThreadCannotBeBlocked() {
+        return new IllegalStateException("The current thread cannot be blocked: " + Thread.currentThread().getName());
+    }
+
     private static void validate(Duration duration) {
         if (duration == null) {
             return;

implementation/src/main/java/io/smallrye/mutiny/operators/uni/builders/UniCreateFromKnownFailure.java

@@ -23,6 +23,10 @@ public void subscribe(UniSubscriber<? super T> subscriber) {
         new KnownFailureSubscription(subscriber).forward();
     }
 
+    public Throwable getFailure() {
+        return failure;
+    }
+
     private class KnownFailureSubscription implements UniSubscription {
 
         private final UniSubscriber<? super T> subscriber;

implementation/src/main/java/io/smallrye/mutiny/operators/uni/builders/UniCreateFromKnownItem.java

@@ -23,6 +23,10 @@ public void subscribe(UniSubscriber<? super T> subscriber) {
         new KnownItemSubscription(subscriber).forward();
     }
 
+    public T getItem() {
+        return item;
+    }
+
     private class KnownItemSubscription implements UniSubscription {
 
         private final UniSubscriber<? super T> subscriber;

implementation/src/test/java/io/smallrye/mutiny/operators/UniAwaitTest.java

@@ -154,6 +154,10 @@ public void testInvalidDurations() {
 
         assertThatThrownBy(() -> one.await().atMost(Duration.ofSeconds(0)))
                 .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("duration");
+
+        assertThatThrownBy(
+                () -> Uni.createFrom().failure(new IllegalArgumentException()).await().atMost(Duration.ofMillis(-2000)))
+                .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("duration");
     }
 
     @Test