org.hibernate.LazyInitializationException#io.smallrye.mutiny.Uni源码实例Demo

@Test
public void testInvokeUniNotCalledOnItem() {
    AtomicReference<Throwable> container = new AtomicReference<>();
    AtomicInteger called = new AtomicInteger(-1);
    int res = Uni.createFrom().item(3)
            .onFailure().invokeUni(t -> {
                container.set(t);
                return Uni.createFrom().item(22).onItem().invoke(called::set);
            })
            .onFailure().recoverWithItem(1)
            .await().indefinitely();

    assertThat(res).isEqualTo(3);
    assertThat(container).hasValue(null);
    assertThat(called).hasValue(-1);
}
 

@Test
public void testWithCancellationBeforeEmission() {
    UniAssertSubscriber<Integer> test = UniAssertSubscriber.create();
    AtomicBoolean cancelled = new AtomicBoolean();
    @SuppressWarnings("unchecked")
    CompletableFuture<Integer> future = new CompletableFuture() {
        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            cancelled.set(true);
            return true;
        }
    };

    Uni<Integer> uni = Uni.createFrom().item(1).onItem().produceCompletionStage(v -> future);
    uni.subscribe().withSubscriber(test);
    test.cancel();
    test.assertNotCompleted();
    assertThat(cancelled).isTrue();
}
 

/**
 * Produces a new {@link Multi} invoking the given @{code action} when an {@code item} event is received. Note that
 * the received item cannot be {@code null}.
 * <p>
 * Unlike {@link #invoke(Consumer)}, the passed function returns a {@link Uni}. When the produced {@code Uni} sends
 * its result, the result is discarded, and the original {@code item} is forwarded downstream. If the produced
 * {@code Uni} fails, the failure is propagated downstream.
 *
 * If the asynchronous action throws an exception, this exception is propagated downstream.
 *
 * This method preserves the order of the items, meaning that the downstream received the items in the same order
 * as the upstream has emitted them.
 *
 * @param action the function taking the item and returning a {@link Uni}, must not be {@code null}
 * @return the new {@link Multi}
 */
public Multi<T> invokeUni(Function<? super T, ? extends Uni<?>> action) {
    ParameterValidation.nonNull(action, "action");
    return produceUni(i -> {
        Uni<?> uni = action.apply(i);
        if (uni == null) {
            throw new NullPointerException("The `action` produced a `null` Uni");
        }
        return uni.onItemOrFailure().produceUni((ignored, failure) -> {
            if (failure != null) {
                return Uni.createFrom().failure(failure);
            } else {
                return Uni.createFrom().item(i);
            }
        });
    }).concatenate();
}
 

@Test
public void testFromAnUniSendingResultEventInTheFuture() {
    AtomicInteger count = new AtomicInteger();

    Multi<Integer> multi = Multi.createFrom()
            .completionStage(() -> CompletableFuture.supplyAsync(count::incrementAndGet));

    multi.toUni().subscribe().withSubscriber(UniAssertSubscriber.create())
            .await()
            .assertItem(1)
            .assertCompletedSuccessfully();

    Uni.createFrom().multi(multi).subscribe().withSubscriber(UniAssertSubscriber.create())
            .await()
            .assertItem(2)
            .assertCompletedSuccessfully();
}
 

@Transactional
@GET
@Path("/transaction-uni")
public Uni<String> contextPropagationWithTxAndUni() throws SystemException {
    SomeEntity.deleteAll();
    Uni<String> ret = Uni.createFrom().item("OK");
    SomeEntity entity = new SomeEntity();
    entity.name = "Stef";
    entity.persist();
    Transaction t1 = Panache.getTransactionManager().getTransaction();
    Assertions.assertNotNull(t1);

    return ret
            .emitOn(executor)
            .map(text -> {
                Assertions.assertEquals(1, SomeEntity.count());
                Transaction t2;
                try {
                    t2 = Panache.getTransactionManager().getTransaction();
                } catch (SystemException e) {
                    throw new RuntimeException(e);
                }
                Assertions.assertEquals(t1, t2);
                return text;
            });
}
 

/**
 * Creates a new {@link Uni} that completes immediately after being subscribed to with the specified (potentially
 * {@code null}) value. The item is retrieved <strong>lazily</strong> at subscription time, using the passed
 * {@link Supplier}. Unlike {@link #deferred(Supplier)}, the supplier produces an item and not an {@link Uni}.
 * <p>
 * This variant of {@link #item(Supplier)} allows passing a state supplier. This supplier allows
 * sharing some <em>state</em> between the subscribers. It is particularly useful when using {@link Uni#repeat()}
 * as you can pass a shared state (for example a page counter, like an AtomicInteger, if you implement pagination).
 * The state supplier is called once, during the first subscription. Note that the mapper is called for every
 * subscription.
 * <p>
 * The state supplier should produce a container wrapping the shared state. This shared state must be thread-safe.
 *
 * @param stateSupplier the state supplier, must not return {@code null}, must not be {@code null}
 * @param mapper the taking the shared state and producing the item.
 * @param <T> the type of item
 * @param <S> the type of the state
 * @return the produced {@link Uni}
 */
public <T, S> Uni<T> item(Supplier<S> stateSupplier,
        Function<S, ? extends T> mapper) {
    ParameterValidation.nonNull(stateSupplier, "stateSupplier");
    ParameterValidation.nonNull(mapper, "mapper");
    // Flag checking that the state supplier is only called once.
    AtomicBoolean once = new AtomicBoolean();
    // The shared state container.
    AtomicReference<S> state = new AtomicReference<>();

    return Uni.createFrom().deferred(() -> {
        try {
            invokeOnce(once, state, stateSupplier);
        } catch (Throwable e) {
            return Uni.createFrom().failure(e);
        }

        S sharedState = state.get();
        if (sharedState == null) {
            // The state supplier failed or produced null.
            return Uni.createFrom().failure(new IllegalStateException("Invalid shared state"));
        }
        return Uni.createFrom().item(mapper.apply(sharedState));
    });
}
 

@Test
public void testApply() {
    assertThat(Uni.createFrom().item("hello")
            .onItem().ifNotNull().apply(String::toUpperCase)
            .await().indefinitely()).isEqualTo("HELLO");

    assertThat(Uni.createFrom().item(() -> (String) null)
            .onItem().ifNotNull().apply(String::toUpperCase)
            .onItem().ifNull().continueWith("yolo")
            .await().indefinitely()).isEqualTo("yolo");

    assertThatThrownBy(() -> Uni.createFrom().<String> failure(new Exception("boom"))
            .onItem().ifNotNull().apply(String::toUpperCase)
            .onItem().ifNull().continueWith("yolo")
            .await().indefinitely()).hasMessageContaining("boom");
}
 

@Test
public void testWithSixUnis() {
    Uni<Integer> uni1 = Uni.createFrom().item(1);
    Uni<Integer> uni2 = Uni.createFrom().item(2);
    Uni<Integer> uni3 = Uni.createFrom().item(3);
    Uni<Integer> uni4 = Uni.createFrom().item(4);
    Uni<Integer> uni5 = Uni.createFrom().item(5);
    Uni<Integer> uni6 = Uni.createFrom().item(6);

    UniAssertSubscriber<Tuple6<Integer, Integer, Integer, Integer, Integer, Integer>> subscriber = Uni.combine()
            .all()
            .unis(uni1, uni2, uni3, uni4, uni5, uni6).asTuple()
            .subscribe().withSubscriber(UniAssertSubscriber.create());

    subscriber.assertCompletedSuccessfully();

    assertThat(subscriber.getItem().asList()).containsExactly(1, 2, 3, 4, 5, 6);
}
 

@GET
@Path("/reactiveRemoveTransientEntity")
@Produces(MediaType.APPLICATION_JSON)
public Uni<String> reactiveRemoveTransientEntity() {
    return mutinySession
            .flatMap(mutinySession -> {
                return populateDB()
                        .flatMap(junk -> selectNameFromId(5))
                        .map(name -> {
                            if (name == null)
                                throw new AssertionError("Database was not populated properly");
                            return name;
                        })
                        .flatMap(junk -> mutinySession.merge(new GuineaPig(5, "Aloi")))
                        .flatMap(aloi -> mutinySession.remove(aloi))
                        .flatMap(junk -> mutinySession.flush())
                        .flatMap(junk -> selectNameFromId(5))
                        .map(result -> {
                            if (result == null)
                                return "OK";
                            else
                                return result;
                        });
            });
}
 

@Override
public Uni<Void> onPartitionsAssigned(KafkaConsumer<?, ?> consumer, Set<TopicPartition> set) {
    // will perform the underlying operation but simulate an error on the first attempt
    return super.onPartitionsAssigned(consumer, set)
            .onItem()
            .produceUni(a -> {
                if (!set.isEmpty() && failOnFirstAttempt.getAndSet(false)) {
                    return Uni
                            .createFrom()
                            .failure(new Exception("testing failure"));
                } else {
                    return Uni
                            .createFrom()
                            .item(a);
                }
            });
}
 

/**
 * Extract the Authorization header and validate the bearer token if it exists. If it does, and is validated, this
 * builds the org.jboss.security.SecurityContext authenticated Subject that drives the container APIs as well as
 * the authorization layers.
 *
 * @param context - the http request exchange object
 * @param identityProviderManager - the current security context that
 * @return one of AUTHENTICATED, NOT_AUTHENTICATED or NOT_ATTEMPTED depending on the header and authentication outcome.
 */
@Override
public Uni<SecurityIdentity> authenticate(RoutingContext context,
        IdentityProviderManager identityProviderManager) {
    String authHeader = context.request().headers().get("Authorization");
    String bearerToken = authHeader != null ? authHeader.substring(7) : null;
    if (bearerToken != null) {
        // Install the OAuth2 principal as the caller
        return identityProviderManager
                .authenticate(new TokenAuthenticationRequest(new TokenCredential(bearerToken, "bearer")));

    }
    // No suitable header has been found in this request,
    return Uni.createFrom().nullItem();
}
 

@Test
public void testWithExceptionThrownByAStage() {
    UniAssertSubscriber<String> ts = UniAssertSubscriber.create();
    CompletionStage<String> cs = new CompletableFuture<>();
    Uni.createFrom().completionStage(() -> cs
            .thenApply(String::toUpperCase)
            .<String> thenApply(s -> {
                throw new IllegalStateException("boom");
            })).subscribe().withSubscriber(ts);
    cs.toCompletableFuture().complete("bonjour");
    ts.assertFailure(IllegalStateException.class, "boom");
}
 

@Test
public void testThatValueIsNotEmittedBeforeSubscription() {
    UniAssertSubscriber<Integer> ts = UniAssertSubscriber.create();
    AtomicBoolean called = new AtomicBoolean();
    CompletableFuture<Integer> cs = new CompletableFuture<>();
    cs.complete(1);
    Uni<Integer> uni = Uni.createFrom().completionStage(cs)
            .onItem().invoke(i -> called.set(true));

    assertThat(called).isFalse();

    uni.subscribe().withSubscriber(ts);
    ts.assertCompletedSuccessfully().assertItem(1);
    assertThat(called).isTrue();
}
 

@Test
public void testFlatMapMultiWithNull() {
    Uni.createFrom().voidItem()
            .onItem().produceMulti(x -> Multi.createFrom().range(1, 5))
            .subscribe().withSubscriber(MultiAssertSubscriber.create(10))
            .await()
            .assertCompletedSuccessfully()
            .assertReceived(1, 2, 3, 4);
}
 

@Test
public void testDelayMultiRandom() {
    // tag::delay-multi-random[]
    Random random = new Random();
    List<Integer> delayed = Multi.createFrom().items(1, 2, 3, 4, 5)
            .onItem().produceUni(i -> Uni.createFrom().item(i).onItem().delayIt().by(Duration.ofMillis(random.nextInt(100) + 1)))
            .merge()
            .collectItems().asList()
            .await().indefinitely();
    // end::delay-multi-random[]
    assertThat(delayed).containsExactlyInAnyOrder(1, 2, 3, 4, 5);
}
 

@Test
public void rx() {
    Multi<Integer> multi = Multi.createFrom().range(1, 3);
    Uni<Integer> uni = Uni.createFrom().item(1);
    // tag::rx[]

    int result = uni
            .map(i -> i + 1)
            .await().indefinitely();

    int result2 = uni
            .flatMap(i -> Uni.createFrom().item(i + 1))
            .await().indefinitely();

    List<Integer> list = multi
            .map(i -> i + 1)
            .collectItems().asList()
            .await().indefinitely();

    List<Integer> list2 = multi
            .flatMap(i -> Multi.createFrom().items(i, i))
            .collectItems().asList()
            .await().indefinitely();

    List<Integer> list3 = multi
            .concatMap(i -> Multi.createFrom().items(i, i))
            .collectItems().asList()
            .await().indefinitely();

    // end::rx[]
    assertThat(result).isEqualTo(2);
    assertThat(result2).isEqualTo(2);
    assertThat(list).containsExactly(2, 3);
    assertThat(list2).containsExactly(1, 1, 2, 2);
    assertThat(list3).containsExactly(1, 1, 2, 2);
}
 

@POST
@Path("/subscribe")
public Uni<Response> subscribe() {
    return Uni.createFrom()
        .completionStage(sns.subscribe(s -> s.topicArn(topicArn).protocol("http").endpoint(notificationEndpoint())))
        .onItem().apply(SubscribeResponse::subscriptionArn)
        .onItem().invoke(this::setSubscriptionArn)
        .onItem().invoke(arn -> LOGGER.infov("Subscribed Quarks shield with id = {0} ", arn))
        .onItem().apply(arn -> Response.ok().entity(arn).build());
}
 

@Override
public Uni<SecurityIdentity> authenticate(TrustedAuthenticationRequest request,
        AuthenticationRequestContext context) {
    TestIdentityController.TestIdentity ident = TestIdentityController.idenitities.get(request.getPrincipal());
    if (ident == null) {
        return Uni.createFrom().optional(Optional.empty());
    }
    return Uni.createFrom().completionStage(CompletableFuture
            .completedFuture(QuarkusSecurityIdentity.builder().setPrincipal(new QuarkusPrincipal(request.getPrincipal()))
                    .addRoles(ident.roles).build()));
}
 

@Test
public void testWithOnTerminationActionWithResult() {
    UniAssertSubscriber<Integer> subscriber = UniAssertSubscriber.create();
    AtomicInteger onTerminationCalled = new AtomicInteger();
    Uni.createFrom().<Integer> emitter(emitter -> {
        emitter.onTermination(onTerminationCalled::incrementAndGet);
        emitter.complete(1);
    }).subscribe().withSubscriber(subscriber);

    assertThat(onTerminationCalled).hasValue(1);
    subscriber.cancel();
    assertThat(onTerminationCalled).hasValue(1);

    subscriber.assertCompletedSuccessfully();
}
 

@Test
public void testNoRepeatUntil() {
    AtomicInteger count = new AtomicInteger();
    List<Integer> list = Uni.createFrom().item(count::getAndIncrement)
            .repeat().until(x -> true)
            .collectItems().asList()
            .await().indefinitely();

    assertThat(list).isEmpty();
    assertThat(count).hasValue(1);
}
 

@Funq("void-function")
public Uni<Void> voidFunction(boolean willThrow) {
    if (willThrow) {
        return Uni.createFrom().failure(new RuntimeException(TEST_EXCEPTION_MSG));
    } else {
        return Uni.createFrom().item((Void) null);
    }
}
 

@Test
public void testThatValueIsNotEmittedBeforeSubscription() {
    UniAssertSubscriber<Integer> ts = UniAssertSubscriber.create();
    AtomicBoolean called = new AtomicBoolean();
    Uni<Integer> uni = Uni.createFrom().publisher(Flowable.generate(emitter -> {
        called.set(true);
        emitter.onNext(1);
        emitter.onComplete();
    }));

    assertThat(called).isFalse();
    uni.subscribe().withSubscriber(ts);
    ts.assertCompletedSuccessfully().assertItem(1);
    assertThat(called).isTrue();
}
 

@Incoming(NO_ACKNOWLEDGMENT_CS)
@Acknowledgment(Acknowledgment.Strategy.NONE)
@Outgoing("sink-" + NO_ACKNOWLEDGMENT_UNI)
public Uni<String> processorWithNoAckUni(String input) {
    return Uni.createFrom().item(() -> {
        processed(NO_ACKNOWLEDGMENT_UNI, input);
        return input + "1";
    });
}
 

@Test
public void uniExportToReactor() {
    Uni<String> uni = Uni.createFrom().item("hello");
    // tag::uni-export[]
    Mono<String> mono = uni.convert().with(UniReactorConverters.toMono());
    Flux<String> flux = uni.convert().with(UniReactorConverters.toFlux());
    // end::uni-export[]

    assertThat(mono.block()).isEqualTo("hello");
    assertThat(flux.blockFirst()).isEqualTo("hello");
}
 

public Uni<List<Fruit>> list() {
    return getCollection().find()
            .map(doc -> {
                Fruit fruit = new Fruit();
                fruit.setName(doc.getString("name"));
                fruit.setDescription(doc.getString("description"));
                return fruit;
            }).collectItems().asList();
}
 

@Test
public void testWithFourUnisAndDeprecatedApis() {
    Uni<Integer> uni = Uni.createFrom().item(1);
    Uni<Integer> uni2 = Uni.createFrom().item(2);
    Uni<Integer> uni3 = Uni.createFrom().item(3);

    UniAssertSubscriber<Tuple4<Integer, Integer, Integer, Integer>> subscriber = uni.and()
            .unis(uni, uni2, uni3).asTuple()
            .subscribe().withSubscriber(UniAssertSubscriber.create());

    assertThat(subscriber.getItem().asList()).containsExactly(1, 1, 2, 3);
}
 

@Test
public void testTimeout() {
    AtomicReference<String> thread = new AtomicReference<>();

    Uni.createFrom().emitter(e -> {
        // do nothing
    })
            .ifNoItem().after(Duration.ofMillis(10)).recoverWithItem("hello")
            .onItem().invoke(l -> thread.set(Thread.currentThread().getName()))
            .await().indefinitely();
    assertThat(thread.get()).startsWith("my-thread-");
}
 

@Test
public void testThatDelayDoNotImpactFailures() {
    long begin = System.currentTimeMillis();
    UniAssertSubscriber<Void> subscriber = UniAssertSubscriber.create();
    Uni.createFrom().<Void> failure(new Exception("boom")).onItem().delayIt()
            .onExecutor(executor)
            .by(Duration.ofMillis(100)).subscribe().withSubscriber(subscriber);
    subscriber.await();
    long end = System.currentTimeMillis();
    assertThat(end - begin).isLessThan(100);
    subscriber.assertCompletedWithFailure().assertFailure(Exception.class, "boom");
}
 

@Test
public void testInvokeUniWithSubFailure() {
    AtomicReference<Throwable> failure = new AtomicReference<>();

    MultiAssertSubscriber<Integer> subscriber = failed.onFailure().invokeUni(i -> {
        failure.set(i);
        return Uni.createFrom().failure(new IllegalStateException("d'oh"));
    }).subscribe().withSubscriber(MultiAssertSubscriber.create(10));

    subscriber
            .assertHasFailedWith(CompositeException.class, "boom")
            .assertHasFailedWith(CompositeException.class, "d'oh")
            .assertReceived(1, 2);
    assertThat(failure).hasValue(BOOM);
}
 

@Test
public void testWithNonNullValue() {
    UniAssertSubscriber<String> ts = UniAssertSubscriber.create();
    CompletionStage<String> cs = new CompletableFuture<>();
    Uni.createFrom().completionStage(cs).subscribe().withSubscriber(ts);
    cs.toCompletableFuture().complete("1");
    ts.assertCompletedSuccessfully().assertItem("1");
}