Reactive Programming and Mutiny

package com.example.book;

import jakarta.enterprise.context.ApplicationScoped;

import io.smallrye.mutiny.Uni;

@ApplicationScoped
public class BrokenEventLoopExample {

    // WRONG: this Uni-returning method stays on the event loop, but Thread.sleep()
    // blocks that event loop thread -- and every other request sharing it -- for 2 seconds.
    public Uni<String> brokenDelay() {
        return Uni.createFrom().item(() -> {
            try {
                Thread.sleep(2000); // never do this in a Uni pipeline
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "done";
        });
    }
}

package com.example.book;

import io.smallrye.mutiny.Uni;

public class SubscriptionExample {

    public static void main(String[] args) {
        Uni<String> pipeline = Uni.createFrom().item("hello")
                .onItem().transform(String::toUpperCase);

        // Nothing has run yet -- pipeline is just a description.
        // Only subscribing actually triggers execution:
        pipeline.subscribe().with(
                item -> System.out.println("Item: " + item),
                failure -> System.err.println("Failure: " + failure));
    }
}

package com.example.book;

import java.util.concurrent.CompletionStage;

import org.reactivestreams.Publisher;

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.converters.uni.UniRxConverters;
import io.reactivex.rxjava3.core.Single;

public class MutinyBridgingExample {

    // Uni -> CompletionStage and back
    Uni<Book> uni = fetchBook();
    CompletionStage<Book> stage = uni.subscribeAsCompletionStage();
    Uni<Book> backToUni = Uni.createFrom().completionStage(stage);

    // Multi -> Reactive Streams Publisher and back
    Multi<Book> multi = streamBooks();
    Publisher<Book> publisher = multi.convert().toPublisher();
    Multi<Book> backToMulti = Multi.createFrom().publisher(publisher);

    // Uni -> RxJava Single and back (requires the smallrye-mutiny-rxjava artifact)
    Single<Book> single = uni.convert().with(UniRxConverters.toSingle());
    Uni<Book> fromSingle = Uni.createFrom().converter(io.smallrye.mutiny.converters.uni.UniRxConverters.fromSingle(), single);

    private Uni<Book> fetchBook() {
        return Uni.createFrom().item(new Book(1L, "Effective Java", "Joshua Bloch"));
    }

    private Multi<Book> streamBooks() {
        return Multi.createFrom().item(new Book(1L, "Effective Java", "Joshua Bloch"));
    }
}

package com.example.book;

import org.junit.jupiter.api.Test;

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.helpers.test.UniAssertSubscriber;
import io.smallrye.mutiny.helpers.test.AssertSubscriber;

class MutinyPipelineTest {

    @Test
    void uniEmitsExpectedItem() {
        Uni<String> uni = Uni.createFrom().item("hello").onItem().transform(String::toUpperCase);

        UniAssertSubscriber<String> subscriber = uni.subscribe()
                .withSubscriber(UniAssertSubscriber.create());

        subscriber.awaitItem().assertItem("HELLO");
    }

    @Test
    void multiEmitsExpectedSequenceThenCompletes() {
        Multi<Integer> multi = Multi.createFrom().items(1, 2, 3);

        AssertSubscriber<Integer> subscriber = multi.subscribe()
                .withSubscriber(AssertSubscriber.create(10)); // requests up to 10 items

        subscriber.awaitCompletion().assertItems(1, 2, 3);
    }

    @Test
    void uniPropagatesFailure() {
        Uni<String> failing = Uni.createFrom().failure(new IllegalStateException("boom"));

        UniAssertSubscriber<String> subscriber = failing.subscribe()
                .withSubscriber(UniAssertSubscriber.create());

        subscriber.awaitFailure().assertFailedWith(IllegalStateException.class, "boom");
    }
}

Uni<Book> traced = findBookById(1L)
        .log("after-find")
        .onItem().transform(this::toDto)
        .log("after-map");

package com.example.book;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.context.RequestScoped;

import io.smallrye.mutiny.Uni;

@RequestScoped
public class RequestContext {
    private String correlationId;

    public String getCorrelationId() {
        return correlationId;
    }

    public void setCorrelationId(String correlationId) {
        this.correlationId = correlationId;
    }
}

@ApplicationScoped
public class CorrelationAwareService {

    private final RequestContext requestContext;

    public CorrelationAwareService(RequestContext requestContext) {
        this.requestContext = requestContext;
    }

    public Uni<Book> findWithCorrelation(Long id) {
        // requestContext resolves correctly here even if this stage
        // runs on a different event loop thread than the one that
        // received the original HTTP request -- the duplicated
        // context, not a ThreadLocal, carries it across the hop.
        return findBookById(id)
                .onItem().invoke(book ->
                        LOG.infof("[%s] found %s", requestContext.getCorrelationId(), book));
    }

    private Uni<Book> findBookById(Long id) {
        return Uni.createFrom().item(new Book(id, "Effective Java", "Joshua Bloch"));
    }

    private static final org.jboss.logging.Logger LOG = org.jboss.logging.Logger.getLogger(CorrelationAwareService.class);
}