Smallrye Mutiny: Async for mere mortals

client.sendAsync(request, HttpResponse.BodyHandlers.ofString())
        .thenRun(() -> {
            long elapsed = System.nanoTime() - requestStart;
            minResponseTime.getAndAccumulate(elapsed, Math::min);
            maxResponseTime.getAndAccumulate(elapsed, Math::max);
        })
        .thenRun(performedRequests::incrementAndGet)
        .exceptionally(t -> {
            failures.incrementAndGet();
            performedRequests.incrementAndGet();
            return null;
        });

package at.htl.feature.user;

import io.quarkus.hibernate.orm.panache.PanacheRepository;
import jakarta.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class UserRepository implements PanacheRepository<User> {
}

package at.htl.feature.user;

import jakarta.inject.Inject;
import jakarta.transaction.Transactional;
import jakarta.ws.rs.*;
import jakarta.ws.rs.core.Context;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.UriInfo;

import java.net.URI;
import java.util.List;

@Path("/user")
public class UserResource {
    @Inject
    UserRepository userRepository;

    @GET
    @Produces(MediaType.APPLICATION_JSON)
    public List<UserDto> getAll() {
        return userRepository.listAll()
                .stream()
                .map(user -> new UserDto(
                        user.getId(),
                        user.getFirstName(),
                        user.getLastName(),
                        user.getEmail()
                ))
                .toList();
    }

    // tag::get_simple_classic[]
    @GET
    @Path("{id}")
    @Produces(MediaType.APPLICATION_JSON)
    public UserDto getById(@PathParam("id") Long id) {
        // = Blocking=
        User user = userRepository.findById(id);
        // = Blocking =

        return new UserDto(
                user.getId(),
                user.getFirstName(),
                user.getLastName(),
                user.getEmail()
        );
    }
    // end::get_simple_classic[]

    // tag::create_classic[]
    @POST
    @Produces(MediaType.APPLICATION_JSON)
    @Consumes(MediaType.APPLICATION_JSON)
    @Transactional
    public Response createNew(User user, @Context UriInfo uriInfo) {
        if (user == null) {
            return Response.status(Response.Status.BAD_REQUEST).build();
        }

        // = Blocking =
        userRepository.persistAndFlush(user);
        // = Blocking =

        URI uri = uriInfo
                .getAbsolutePathBuilder()
                .path(Long.toString(user.getId()))
                .build();

        return Response
                .created(uri)
                .entity(user)
                .build();
    }
    // end::create_classic[]
}

package at.htl.feature.user;

import io.quarkus.hibernate.reactive.panache.PanacheRepository;
import jakarta.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class UserRepository implements PanacheRepository<User> {
}

package at.htl.feature.user;

import io.quarkus.hibernate.reactive.panache.Panache;
import io.quarkus.hibernate.reactive.panache.common.WithSession;
import io.quarkus.hibernate.reactive.panache.common.WithTransaction;
import io.quarkus.logging.Log;
import io.smallrye.mutiny.Uni;
import jakarta.inject.Inject;
import jakarta.ws.rs.*;
import jakarta.ws.rs.core.Context;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.UriInfo;

import java.net.URI;
import java.util.List;

@Path("/user")
public class UserResource {
    @Inject
    UserRepository userRepository;

    @GET
    @Produces(MediaType.APPLICATION_JSON)
    // @WithSession
    // Not needed (https://quarkus.io/guides/hibernate-reactive-panache#transactions)
    // and: https://stackoverflow.com/questions/77528402/missing-withsession-with-quarkus-resteasy-reactive-causes-java-lang-illegalstat
    public Uni<List<UserDto>> getAll() {
        return userRepository.listAll()
                .onItem().transform(users ->
                        users
                                .stream()
                                .map(user -> new UserDto(
                                        user.getId(),
                                        user.getFirstName(),
                                        user.getLastName(),
                                        user.getEmail())
                                ).toList()
                );
    }

    // tag::get_simple[]
    @GET
    @Path("{id}")
    @Produces(MediaType.APPLICATION_JSON)
    // @WithSession (1)
    public Uni<UserDto> getById(@PathParam("id") Long id) { (2)
        return userRepository.findById(id) (3)
                .onItem().transform(user -> new UserDto( (4)
                        user.getId(),
                        user.getFirstName(),
                        user.getLastName(),
                        user.getEmail())
                );
    }
    // end::get_simple[]

    // tag::create[]
    @POST
    @Produces(MediaType.APPLICATION_JSON)
    @Consumes(MediaType.APPLICATION_JSON)
    @WithTransaction (5)
    public Uni<Response> createNew(User user, @Context UriInfo uriInfo) { (6)
        return Uni.createFrom()
                .item(user) (7)
                .onItem()
                    .ifNull()
                    .failWith(new WebApplicationException(Response.Status.BAD_REQUEST)) (8)
                .onItem()
                    .ifNotNull()
                    .transformToUni(ignored -> userRepository.persistAndFlush(user)) (9)
                .onItem().transform(u -> { (10)
                    URI uri = uriInfo
                            .getAbsolutePathBuilder()
                            .path(Long.toString(u.getId()))
                            .build();

                    return Response
                            .created(uri)
                            .entity(u)
                            .build();
                });
    }
    // end::create[]
}

package at.htl;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class Stress {
    private static final AtomicInteger performedRequests = new AtomicInteger(0);
    private static final AtomicInteger failures = new AtomicInteger(0);
    private static final AtomicLong maxResponseTime = new AtomicLong(Long.MIN_VALUE);
    private static final AtomicLong minResponseTime = new AtomicLong(Long.MAX_VALUE);

    private static void printUsage() {
        System.err.println("[usage]: <url> <total-requests> <mode>");
        System.err.println("where total-requests > 0");
        System.err.println("mode is one of GET,POST");
    }

    public static void main(String[] args) {
        if (args.length < 3) {
            printUsage();
            return;
        }

        String url = args[0];
        int totalRequests = Integer.parseInt(args[1]);
        String mode = args[2];
        System.out.printf("Performing %,d requests%n", totalRequests);

        if (totalRequests <= 0) {
            printUsage();
            return;
        }

        HttpRequest request;

        if (mode.equalsIgnoreCase("GET")) {
            request = HttpRequest.newBuilder()
                    .uri(URI.create(url))
                    .GET()
                    .build();
        } else if (mode.equalsIgnoreCase("POST")) {
            request = HttpRequest.newBuilder()
                    .uri(URI.create(url))
                    .POST(HttpRequest.BodyPublishers.ofString(
                            """
                                     {
                                       "firstName": "Horst",
                                       "lastName": "Flock",
                                       "email": "h.flock@yahoomail.com",
                                       "address": "Bauernweg 5, 4033 Basel",
                                       "dateOfBirth": "2003-11-21"
                                     }
                                    """
                    ))
                    .build();
        } else {
            printUsage();
            return;
        }

        long start;
        try (HttpClient client = HttpClient.newHttpClient()) {
            start = System.nanoTime();
            for (int i = 0; i < totalRequests; i++) {
                long requestStart = System.nanoTime();
                // tag::send_webrequest[]
                client.sendAsync(request, HttpResponse.BodyHandlers.ofString())
                        .thenRun(() -> {
                            long elapsed = System.nanoTime() - requestStart;
                            minResponseTime.getAndAccumulate(elapsed, Math::min);
                            maxResponseTime.getAndAccumulate(elapsed, Math::max);
                        })
                        .thenRun(performedRequests::incrementAndGet)
                        .exceptionally(t -> {
                            failures.incrementAndGet();
                            performedRequests.incrementAndGet();
                            return null;
                        });
                // end::send_webrequest[]
            }
        }

        while (performedRequests.get() < totalRequests) {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        long end = System.nanoTime();
        long totalElapsedNanos = end - start;

        System.out.printf("Total: %f ms%n", totalElapsedNanos / 1000000.0);
        System.out.printf("Avg: %f ms%n", totalElapsedNanos / (double) totalRequests / 1000000.0);
        System.out.printf("Min: %f ms%n", minResponseTime.get() / 1000000.0);
        System.out.printf("Max: %f ms%n", maxResponseTime.get() / 1000000.0);
        System.out.printf("Failures: %d%n", failures.get());
    }
}