Port UniCreateFromFuture from the master branch
package io.smallrye.mutiny.operators.uni.builders;
-import static io.smallrye.mutiny.helpers.EmptyUniSubscription.propagateFailureEvent;
+import static io.smallrye.mutiny.helpers.EmptyUniSubscription.DONE;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -11,57 +11,72 @@ import io.smallrye.mutiny.operators.AbstractUni;
import io.smallrye.mutiny.subscription.UniSubscriber;
public class UniCreateFromFuture<T> extends AbstractUni<T> {
+
private final Supplier<? extends Future<? extends T>> supplier;
public UniCreateFromFuture(Supplier<? extends Future<? extends T>> supplier) {
this.supplier = supplier; // Already checked
}
- static <O> void forwardFromFuture(Future<? extends O> future,
- UniSubscriber<? super O> subscriber) {
- subscriber.onSubscribe(() -> future.cancel(false));
- if (future.isDone()) {
- O item;
- try {
- item = future.get();
- } catch (ExecutionException e) {
- subscriber.onFailure(e.getCause());
- return;
- } catch (Exception e) {
- subscriber.onFailure(e);
+ @Override
+ public void subscribe(UniSubscriber<? super T> downstream) {
+ Future<? extends T> future = obtainFuture(downstream);
+ if (future == null) {
return;
}
- subscriber.onItem(item);
+ if (future.isDone()) {
+ dispatchImmediateResult(future, downstream);
} else {
- // Because future.get is blocking, we must use a separated thread.
- Infrastructure.getDefaultExecutor().execute(() -> {
- try {
- O item = future.get();
- subscriber.onItem(item);
- } catch (ExecutionException e) {
- subscriber.onFailure(e.getCause());
- } catch (Exception e) {
- subscriber.onFailure(e);
- }
- });
+ dispatchDeferredResult(future, downstream);
}
}
- @Override
- protected void subscribing(UniSubscriber<? super T> subscriber) {
+ private Future<? extends T> obtainFuture(UniSubscriber<? super T> downstream) {
Future<? extends T> future;
try {
future = supplier.get();
- } catch (Throwable e) {
- propagateFailureEvent(subscriber, e);
- return;
+ } catch (Throwable err) {
+ downstream.onSubscribe(DONE);
+ downstream.onFailure(err);
+ return null;
}
-
if (future == null) {
- propagateFailureEvent(subscriber, new NullPointerException("The produced Future is `null`"));
+ downstream.onSubscribe(DONE);
+ downstream.onFailure(new NullPointerException("The produced Future is `null`"));
+ return null;
+ }
+ return future;
+ }
+
+ private void dispatchImmediateResult(Future<? extends T> future, UniSubscriber<? super T> downstream) {
+ T item;
+ try {
+ item = future.get();
+ } catch (ExecutionException e) {
+ downstream.onSubscribe(DONE);
+ downstream.onFailure(e.getCause());
+ return;
+ } catch (Exception err) {
+ downstream.onSubscribe(DONE);
+ downstream.onFailure(err);
return;
}
+ downstream.onSubscribe(DONE);
+ downstream.onItem(item);
+ }
- forwardFromFuture(future, subscriber);
+ private void dispatchDeferredResult(Future<? extends T> future, UniSubscriber<? super T> downstream) {
+ downstream.onSubscribe(() -> future.cancel(false));
+ // Because future.get is blocking, we must use a separated thread.
+ Infrastructure.getDefaultExecutor().execute(() -> {
+ try {
+ T item = future.get();
+ downstream.onItem(item);
+ } catch (ExecutionException e) {
+ downstream.onFailure(e.getCause());
+ } catch (Exception e) {
+ downstream.onFailure(e);
+ }
+ });
}
}
