Quarkus Topic

Team 519 views

Reactive Programming in Quarkus with Mutiny

1. What is Reactive Programming?

Reactive programming is a programming paradigm oriented around asynchronous data streams. It allows you to:

Handle high concurrency efficiently.
Work with non-blocking operations (no thread is wasted waiting for I/O).
Build scalable applications.

Key concepts: Publisher, Subscriber, Stream, Backpressure.

2. Why Use Reactive in Quarkus?

Quarkus supports reactive programming to build highly scalable, event-driven microservices. Benefits:

Non-blocking REST APIs.
Efficient database access with reactive drivers.
Handles thousands of requests with fewer threads.

3. Reactive Libraries in Quarkus

Quarkus uses Mutiny as its main reactive library.

Mutiny provides:
- Uni<T> → Represents a single item or failure.
- Multi<T> → Represents a stream of items.

Other reactive tools you may use:

Vert.x (under the hood for reactive HTTP & WebSockets).
Non-blocking operations: Built for reactive systems
Composable API: Chain operations fluently

Basic Uni Operations

java

package com.example.reactive;

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.Multi;
import jakarta.enterprise.context.ApplicationScoped;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;

@ApplicationScoped
public class MutinyBasics {

    // Creating Uni instances
    public Uni<String> createSimpleUni() {
        return Uni.createFrom().item("Hello, Reactive World!");
    }

    public Uni<String> createUniFromSupplier() {
        return Uni.createFrom().item(() -> {
            // Simulate computation
            try { Thread.sleep(100); } catch (InterruptedException e) { }
            return "Computed value";
        });
    }

    public Uni<String> createUniFromFailure() {
        return Uni.createFrom().failure(new RuntimeException("Something went wrong"));
    }

    public Uni<Void> createEmptyUni() {
        return Uni.createFrom().nullItem();
    }

    // Transforming Uni values
    public Uni<String> transformUni() {
        return Uni.createFrom().item("hello")
                .onItem().transform(String::toUpperCase)
                .onItem().transform(s -> s + " WORLD!")
                .onItem().delayIt().by(Duration.ofMillis(100));
    }

    // Filtering Uni
    public Uni<String> filterUni() {
        return Uni.createFrom().item("test")
                .onItem().filter(s -> s.length() > 3)
                .onItem().ifNull().continueWith("default");
    }

    // Handling failures
    public Uni<String> handleFailures() {
        return Uni.createFrom().<String>failure(new RuntimeException("Boom!"))
                .onFailure().recoverWithItem("fallback")
                .onFailure().retry().atMost(3)
                .onFailure(IllegalArgumentException.class).recoverWithItem("illegal arg fallback");
    }

    // Combining Uni instances
    public Uni<String> combineUni() {
        Uni<String> first = Uni.createFrom().item("Hello");
        Uni<String> second = Uni.createFrom().item("World");

        return Uni.combine().all().unis(first, second)
                .asTuple()
                .onItem().transform(tuple -> tuple.getItem1() + " " + tuple.getItem2());
    }

    // Chaining asynchronous operations
    public Uni<String> chainAsyncOperations() {
        return getUserById(1L)
                .onItem().transformToUni(user -> getProfileByUserId(user.id()))
                .onItem().transform(profile -> "Profile: " + profile.name());
    }

    // Mock methods for demonstration
    private Uni<User> getUserById(Long id) {
        return Uni.createFrom().item(new User(id, "john_doe"));
    }

    private Uni<Profile> getProfileByUserId(Long userId) {
        return Uni.createFrom().item(new Profile(userId, "John Doe", "john@example.com"));
    }

    record User(Long id, String username) {}
    record Profile(Long userId, String name, String email) {}
}

Basic Multi Operations

java

package com.example.reactive;

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.subscription.Cancellable;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

@ApplicationScoped
public class MultiBasics {

    // Creating Multi instances
    public Multi<String> createMultiFromItems() {
        return Multi.createFrom().items("Apple", "Banana", "Cherry");
    }

    public Multi<Integer> createMultiFromRange() {
        return Multi.createFrom().range(1, 10);
    }

    public Multi<String> createMultiFromIterable() {
        return Multi.createFrom().iterable(List.of("One", "Two", "Three"));
    }

    public Multi<Long> createMultiFromTicks() {
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
                .select().first(5); // Emit 5 ticks then complete
    }

    public Multi<String> createMultiFromEmitter() {
        return Multi.createFrom().emitter(em -> {
            em.emit("First");
            em.emit("Second");
            em.emit("Third");
            em.complete();
        });
    }

    // Transforming Multi streams
    public Multi<String> transformMulti() {
        return Multi.createFrom().items("hello", "world", "reactive")
                .onItem().transform(String::toUpperCase)
                .onItem().transform(s -> ">> " + s + " <<")
                .select().where(s -> s.length() > 5);
    }

    // Filtering and selecting items
    public Multi<Integer> filterMulti() {
        return Multi.createFrom().range(1, 20)
                .select().where(n -> n % 2 == 0) // Even numbers only
                .select().first(5) // First 5 even numbers
                .select().last(3); // Last 3 of those 5
    }

    // Skipping items
    public Multi<Integer> skipItems() {
        return Multi.createFrom().range(1, 10)
                .skip().first(3) // Skip first 3
                .skip().last(2); // Skip last 2
    }

    // Collecting Multi to Uni
    public Uni<List<String>> collectMultiToList() {
        return Multi.createFrom().items("A", "B", "C")
                .collect().asList();
    }

    public Uni<String> collectMultiToString() {
        return Multi.createFrom().items("Hello", " ", "World", "!")
                .collect().asString()
                .onItem().transform(s -> s + " 👋");
    }

    // Grouping operations
    public Multi<String> groupMulti() {
        return Multi.createFrom().items("apple", "banana", "avocado", "berry", "apricot")
                .group().intoLists().of(2)
                .onItem().transform(list -> String.join(" & ", list));
    }

    // Merging and concatenating streams
    public Multi<String> mergeMultis() {
        Multi<String> first = Multi.createFrom().items("A", "B", "C");
        Multi<String> second = Multi.createFrom().items("X", "Y", "Z");
        
        return Multi.createBy().merging().streams(first, second);
    }

    public Multi<String> concatMultis() {
        Multi<String> first = Multi.createFrom().items("First", "Second");
        Multi<String> second = Multi.createFrom().items("Third", "Fourth");
        
        return Multi.createBy().concatenating().streams(first, second);
    }

    // Error handling in Multi
    public Multi<String> handleMultiErrors() {
        return Multi.createFrom().<String>failure(new RuntimeException("Stream error"))
                .onFailure().recoverWithItem("fallback item")
                .onFailure().retry().atMost(2);
    }

    // Backpressure handling
    public Multi<Integer> withBackpressure() {
        return Multi.createFrom().range(1, 1000)
                .onOverflow().buffer(100) // Buffer 100 items
                .onOverflow().dropPreviousItems() // Strategy for overflow
                .select().first(50); // Limit output
    }

    // Hot streams (broadcast)
    public Multi<String> createHotStream() {
        AtomicInteger counter = new AtomicInteger();
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
                .onSubscription().invoke(() -> System.out.println("New subscriber!"))
                .onItem().transform(tick -> "Event " + counter.incrementAndGet())
                .broadcast().toAllSubscribers();
    }
}

3. Reactive REST Endpoints

Reactive JAX-RS Resource

java

package com.example.resource;

import com.example.service.ReactiveUserService;
import com.example.model.User;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.Multi;
import jakarta.inject.Inject;
import jakarta.ws.rs.*;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import org.jboss.resteasy.reactive.RestStreamElementType;

import java.time.Duration;
import java.util.List;

@Path("/reactive/users")
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
public class ReactiveUserResource {

    @Inject
    ReactiveUserService userService;

    // GET /reactive/users - Get all users as List (Uni)
    @GET
    public Uni<List<User>> getAllUsers() {
        return userService.getAllUsers();
    }

    // GET /reactive/users/stream - Stream users (Server-Sent Events)
    @GET
    @Path("/stream")
    @Produces(MediaType.SERVER_SENT_EVENTS)
    @RestStreamElementType(MediaType.APPLICATION_JSON)
    public Multi<User> streamUsers() {
        return userService.streamUsers();
    }

    // GET /reactive/users/{id} - Get user by ID
    @GET
    @Path("/{id}")
    public Uni<Response> getUserById(@PathParam("id") Long id) {
        return userService.getUserById(id)
                .onItem().transform(user -> 
                    user != null 
                        ? Response.ok(user).build()
                        : Response.status(Response.Status.NOT_FOUND).build()
                );
    }

    // POST /reactive/users - Create new user
    @POST
    public Uni<Response> createUser(User user) {
        return userService.createUser(user)
                .onItem().transform(createdUser -> 
                    Response.status(Response.Status.CREATED)
                            .entity(createdUser)
                            .build()
                )
                .onFailure().recoverWithItem(failure -> 
                    Response.status(Response.Status.BAD_REQUEST)
                            .entity(new ErrorResponse(failure.getMessage()))
                            .build()
                );
    }

    // PUT /reactive/users/{id} - Update user
    @PUT
    @Path("/{id}")
    public Uni<Response> updateUser(@PathParam("id") Long id, User user) {
        return userService.updateUser(id, user)
                .onItem().transform(updatedUser -> 
                    Response.ok(updatedUser).build()
                )
                .onFailure().recoverWithItem(failure -> 
                    Response.status(Response.Status.NOT_FOUND)
                            .entity(new ErrorResponse(failure.getMessage()))
                            .build()
                );
    }

    // DELETE /reactive/users/{id} - Delete user
    @DELETE
    @Path("/{id}")
    public Uni<Response> deleteUser(@PathParam("id") Long id) {
        return userService.deleteUser(id)
                .onItem().transform(success -> 
                    success 
                        ? Response.noContent().build()
                        : Response.status(Response.Status.NOT_FOUND).build()
                );
    }

    // GET /reactive/users/search - Search users with query params
    @GET
    @Path("/search")
    public Uni<List<User>> searchUsers(
            @QueryParam("name") String name,
            @QueryParam("email") String email,
            @QueryParam("page") @DefaultValue("0") int page,
            @QueryParam("size") @DefaultValue("20") int size) {
        
        return userService.searchUsers(name, email, page, size);
    }

    // GET /reactive/users/{id}/events - Stream user events (SSE)
    @GET
    @Path("/{id}/events")
    @Produces(MediaType.SERVER_SENT_EVENTS)
    @RestStreamElementType(MediaType.APPLICATION_JSON)
    public Multi<UserEvent> streamUserEvents(@PathParam("id") Long userId) {
        return userService.streamUserEvents(userId)
                .onOverflow().buffer(1000);
    }

    // GET /reactive/users/bulk - Bulk operations
    @POST
    @Path("/bulk")
    public Uni<BulkOperationResult> bulkCreateUsers(List<User> users) {
        return userService.bulkCreateUsers(users);
    }

    // Utility classes
    public static class ErrorResponse {
        public String error;
        public long timestamp;

        public ErrorResponse(String error) {
            this.error = error;
            this.timestamp = System.currentTimeMillis();
        }
    }

    public static class UserEvent {
        public String type;
        public User user;
        public long timestamp;

        public UserEvent(String type, User user) {
            this.type = type;
            this.user = user;
            this.timestamp = System.currentTimeMillis();
        }
    }

    public static class BulkOperationResult {
        public int created;
        public int failed;
        public List<String> errors;

        public BulkOperationResult(int created, int failed, List<String> errors) {
            this.created = created;
            this.failed = failed;
            this.errors = errors;
        }
    }
}

Reactive Service Implementation

java

package com.example.service;

import com.example.model.User;
import com.example.resource.ReactiveUserResource;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.Multi;
import jakarta.enterprise.context.ApplicationScoped;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

@ApplicationScoped
public class ReactiveUserService {

    private final AtomicLong idGenerator = new AtomicLong(1000);
    private final List<User> users = new ArrayList<>();

    public ReactiveUserService() {
        // Initialize with some mock data
        users.add(new User(1L, "john_doe", "john@example.com"));
        users.add(new User(2L, "jane_smith", "jane@example.com"));
        users.add(new User(3L, "bob_wilson", "bob@example.com"));
    }

    public Uni<List<User>> getAllUsers() {
        return Uni.createFrom().item(users)
                .onItem().delayIt().by(Duration.ofMillis(100)); // Simulate DB call
    }

    public Multi<User> streamUsers() {
        return Multi.createFrom().iterable(users)
                .onItem().call(user -> 
                    Uni.createFrom().item(user)
                       .onItem().delayIt().by(Duration.ofMillis(500))
                ); // Simulate streaming with delay
    }

    public Uni<User> getUserById(Long id) {
        return Uni.createFrom().item(
            users.stream()
                .filter(user -> user.id().equals(id))
                .findFirst()
                .orElse(null)
        ).onItem().ifNull().failWith(new NotFoundException("User not found with id: " + id));
    }

    public Uni<User> createUser(User user) {
        return Uni.createFrom().item(() -> {
            User newUser = new User(
                idGenerator.incrementAndGet(),
                user.username(),
                user.email()
            );
            users.add(newUser);
            return newUser;
        }).onItem().delayIt().by(Duration.ofMillis(200)); // Simulate DB write
    }

    public Uni<User> updateUser(Long id, User userUpdate) {
        return Uni.createFrom().item(() -> {
            for (int i = 0; i < users.size(); i++) {
                User existing = users.get(i);
                if (existing.id().equals(id)) {
                    User updated = new User(
                        existing.id(),
                        userUpdate.username() != null ? userUpdate.username() : existing.username(),
                        userUpdate.email() != null ? userUpdate.email() : existing.email()
                    );
                    users.set(i, updated);
                    return updated;
                }
            }
            return null;
        }).onItem().ifNull().failWith(new NotFoundException("User not found with id: " + id));
    }

    public Uni<Boolean> deleteUser(Long id) {
        return Uni.createFrom().item(() -> 
            users.removeIf(user -> user.id().equals(id))
        );
    }

    public Uni<List<User>> searchUsers(String name, String email, int page, int size) {
        return Uni.createFrom().item(() -> {
            List<User> filtered = users.stream()
                .filter(user -> 
                    (name == null || user.username().toLowerCase().contains(name.toLowerCase())) &&
                    (email == null || user.email().toLowerCase().contains(email.toLowerCase()))
                )
                .toList();

            // Simple pagination
            int fromIndex = Math.min(page * size, filtered.size());
            int toIndex = Math.min((page + 1) * size, filtered.size());
            
            return filtered.subList(fromIndex, toIndex);
        });
    }

    public Multi<ReactiveUserResource.UserEvent> streamUserEvents(Long userId) {
        return Multi.createFrom().ticks().every(Duration.ofSeconds(2))
                .onItem().transform(tick -> {
                    String eventType = tick % 3 == 0 ? "PROFILE_VIEW" : 
                                     tick % 3 == 1 ? "ACTIVITY_UPDATE" : "STATUS_CHANGE";
                    return new ReactiveUserResource.UserEvent(eventType, getUserById(userId).await().indefinitely());
                })
                .select().first(10); // Emit 10 events then complete
    }

    public Uni<ReactiveUserResource.BulkOperationResult> bulkCreateUsers(List<User> newUsers) {
        return Multi.createFrom().iterable(newUsers)
                .onItem().transformToUniAndMerge(this::createUser)
                .collect().asList()
                .onItem().transform(createdUsers -> {
                    int created = createdUsers.size();
                    int failed = newUsers.size() - created;
                    return new ReactiveUserResource.BulkOperationResult(created, failed, List.of());
                });
    }

    // Custom exception
    public static class NotFoundException extends RuntimeException {
        public NotFoundException(String message) {
            super(message);
        }
    }
}

// User record
package com.example.model;

public record User(Long id, String username, String email) {
    public User {
        if (username == null || username.isBlank()) {
            throw new IllegalArgumentException("Username cannot be null or blank");
        }
        if (email == null || email.isBlank()) {
            throw new IllegalArgumentException("Email cannot be null or blank");
        }
    }
}

4. Reactive REST Client

REST Client Interface

java

package com.example.client;

import com.example.model.ExternalUser;
import io.smallrye.mutiny.Uni;
import jakarta.ws.rs.*;
import jakarta.ws.rs.core.MediaType;
import org.eclipse.microprofile.rest.client.inject.RegisterRestClient;

import java.util.List;
import java.util.concurrent.CompletionStage;

@RegisterRestClient(baseUri = "https://jsonplaceholder.typicode.com")
@Path("/users")
public interface ExternalUserService {

    // Synchronous call (blocking)
    @GET
    @Path("/{id}")
    ExternalUser getUserById(@PathParam("id") Long id);

    // Reactive call with Uni
    @GET
    @Path("/{id}")
    Uni<ExternalUser> getUserByIdAsync(@PathParam("id") Long id);

    // Reactive call with CompletionStage
    @GET
    @Path("/{id}")
    CompletionStage<ExternalUser> getUserByIdCompletion(@PathParam("id") Long id);

    @GET
    Uni<List<ExternalUser>> getAllUsers();

    @POST
    Uni<ExternalUser> createUser(ExternalUser user);

    @PUT
    @Path("/{id}")
    Uni<ExternalUser> updateUser(@PathParam("id") Long id, ExternalUser user);

    @DELETE
    @Path("/{id}")
    Uni<Void> deleteUser(@PathParam("id") Long id);
}

// External User model
package com.example.model;

public class ExternalUser {
    public Long id;
    public String name;
    public String username;
    public String email;
    public String phone;
    public String website;
    
    public Address address;
    public Company company;
    
    public static class Address {
        public String street;
        public String suite;
        public String city;
        public String zipcode;
        public Geo geo;
        
        public static class Geo {
            public String lat;
            public String lng;
        }
    }
    
    public static class Company {
        public String name;
        public String catchPhrase;
        public String bs;
    }
}

Using Reactive REST Client

java

package com.example.service;

import com.example.client.ExternalUserService;
import com.example.model.ExternalUser;
import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.Multi;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.eclipse.microprofile.rest.client.inject.RestClient;

import java.time.Duration;
import java.util.List;

@ApplicationScoped
public class ExternalDataService {

    @Inject
    @RestClient
    ExternalUserService externalUserService;

    // Get user with timeout and retry
    public Uni<ExternalUser> getUserWithResilience(Long id) {
        return externalUserService.getUserByIdAsync(id)
                .onItem().delayIt().by(Duration.ofMillis(100)) // Simulate processing
                .onFailure().retry().atMost(3)
                .onFailure().recoverWithItem(this::getFallbackUser)
                .ifNoItem().after(Duration.ofSeconds(5)).failWith(new TimeoutException("Request timeout"));
    }

    // Process multiple users in parallel
    public Uni<List<ExternalUser>> getMultipleUsers(List<Long> ids) {
        return Multi.createFrom().iterable(ids)
                .onItem().transformToUniAndMerge(this::getUserWithResilience)
                .collect().asList();
    }

    // Stream users with backpressure
    public Multi<ExternalUser> streamUsersWithBackpressure() {
        return Multi.createFrom().range(1, 10)
                .onItem().transformToUniAndMerge(id -> 
                    externalUserService.getUserByIdAsync(id)
                        .onFailure().recoverWithNull()
                )
                .select().where(user -> user != null)
                .onOverflow().buffer(50);
    }

    // Chain multiple external calls
    public Uni<String> getUserProfileSummary(Long userId) {
        return externalUserService.getUserByIdAsync(userId)
                .onItem().transformToUni(user -> 
                    getAdditionalUserData(user.id)
                        .onItem().transform(additionalData -> 
                            String.format("User: %s (%s) - %s", 
                                user.name, user.email, additionalData)
                        )
                );
    }

    // Bulk operations with error handling
    public Uni<BulkResult> bulkCreateUsers(List<ExternalUser> users) {
        return Multi.createFrom().iterable(users)
                .onItem().transformToUniAndMerge(user -> 
                    externalUserService.createUser(user)
                        .onFailure().recoverWithNull()
                )
                .collect().asList()
                .onItem().transform(results -> {
                    long successCount = results.stream().filter(r -> r != null).count();
                    long failureCount = users.size() - successCount;
                    return new BulkResult(successCount, failureCount);
                });
    }

    private Uni<String> getAdditionalUserData(Long userId) {
        return Uni.createFrom().item("Additional data for user " + userId)
                .onItem().delayIt().by(Duration.ofMillis(200));
    }

    private ExternalUser getFallbackUser() {
        ExternalUser fallback = new ExternalUser();
        fallback.id = -1L;
        fallback.name = "Fallback User";
        fallback.email = "fallback@example.com";
        return fallback;
    }

    public static class BulkResult {
        public long successCount;
        public long failureCount;
        
        public BulkResult(long successCount, long failureCount) {
            this.successCount = successCount;
            this.failureCount = failureCount;
        }
    }

    public static class TimeoutException extends RuntimeException {
        public TimeoutException(String message) {
            super(message);
        }
    }
}

5. Reactive Messaging with Kafka

Configuration

properties

# application.properties
# Kafka Configuration
kafka.bootstrap.servers=localhost:9092

# Outgoing Channels
mp.messaging.outgoing.user-created.connector=smallrye-kafka
mp.messaging.outgoing.user-created.topic=user-events
mp.messaging.outgoing.user-created.value.serializer=org.apache.kafka.common.serialization.StringSerializer

mp.messaging.outgoing.user-actions.connector=smallrye-kafka
mp.messaging.outgoing.user-actions.topic=user-actions
mp.messaging.outgoing.user-actions.value.serializer=io.quarkus.kafka.client.serialization.ObjectMapperSerializer

# Incoming Channels
mp.messaging.incoming.user-events.connector=smallrye-kafka
mp.messaging.incoming.user-events.topic=user-events
mp.messaging.incoming.user-events.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
mp.messaging.incoming.user-events.group.id=user-service-group

mp.messaging.incoming.user-actions.connector=smallrye-kafka
mp.messaging.incoming.user-actions.topic=user-actions
mp.messaging.incoming.user-actions.value.deserializer=io.quarkus.kafka.client.serialization.ObjectMapperDeserializer
mp.messaging.incoming.user-actions.group.id=user-service-group

Event Models

java

package com.example.messaging;

import java.time.Instant;

public class UserEvent {
    public String eventId;
    public String eventType;
    public Long userId;
    public String username;
    public Instant timestamp;
    
    public UserEvent() {}
    
    public UserEvent(String eventType, Long userId, String username) {
        this.eventId = java.util.UUID.randomUUID().toString();
        this.eventType = eventType;
        this.userId = userId;
        this.username = username;
        this.timestamp = Instant.now();
    }
    
    public static UserEvent created(Long userId, String username) {
        return new UserEvent("USER_CREATED", userId, username);
    }
    
    public static UserEvent updated(Long userId, String username) {
        return new UserEvent("USER_UPDATED", userId, username);
    }
    
    public static UserEvent deleted(Long userId) {
        UserEvent event = new UserEvent("USER_DELETED", userId, null);
        event.username = null;
        return event;
    }
}

package com.example.messaging;

import java.time.Instant;

public class UserAction {
    public String actionId;
    public String actionType;
    public Long userId;
    public String details;
    public Instant timestamp;
    
    public UserAction() {}
    
    public UserAction(String actionType, Long userId, String details) {
        this.actionId = java.util.UUID.randomUUID().toString();
        this.actionType = actionType;
        this.userId = userId;
        this.details = details;
        this.timestamp = Instant.now();
    }
}

Reactive Messaging Producers

java

package com.example.messaging;

import io.smallrye.mutiny.Uni;
import io.smallrye.reactive.messaging.kafka.Record;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import org.eclipse.microprofile.reactive.messaging.OnOverflow;

import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;

@ApplicationScoped
public class UserEventProducer {

    @Inject
    @Channel("user-created")
    @OnOverflow(value = OnOverflow.Strategy.BUFFER, bufferSize = 1000)
    Emitter<String> userCreatedEmitter;

    @Inject
    @Channel("user-actions")
    @OnOverflow(value = OnOverflow.Strategy.BUFFER, bufferSize = 1000)
    Emitter<UserAction> userActionsEmitter;

    public Uni<Void> sendUserCreatedEvent(Long userId, String username) {
        return Uni.createFrom().item(() -> {
            UserEvent event = UserEvent.created(userId, username);
            userCreatedEmitter.send(event.eventId + ":" + event.eventType + ":" + userId);
            return null;
        });
    }

    public Uni<Void> sendUserAction(UserAction action) {
        return Uni.createFrom().completionStage(
            userActionsEmitter.send(action).toCompletableFuture()
        );
    }

    // Send with key for partitioning
    public Uni<Void> sendUserEventWithKey(UserEvent event) {
        String key = event.userId != null ? event.userId.toString() : "unknown";
        return Uni.createFrom().completionStage(
            userCreatedEmitter.send(Record.of(key, event.eventId + ":" + event.eventType))
                .toCompletableFuture()
        );
    }

    // Batch sending
    public Uni<Void> sendBatchEvents(List<UserEvent> events) {
        return Uni.createFrom().voidItem()
                .onItem().invoke(() -> {
                    for (UserEvent event : events) {
                        userCreatedEmitter.send(event.eventId + ":" + event.eventType + ":" + event.userId);
                    }
                });
    }
}

Reactive Messaging Consumers

java

package com.example.messaging;

import io.smallrye.mutiny.Multi;
import io.smallrye.reactive.messaging.kafka.KafkaRecord;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import org.eclipse.microprofile.reactive.messaging.Acknowledgment;

import jakarta.enterprise.context.ApplicationScoped;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

@ApplicationScoped
public class UserEventConsumer {

    private final AtomicInteger processedCount = new AtomicInteger();

    // Consume user events
    @Incoming("user-events")
    @Acknowledgment(Acknowledgment.Strategy.MANUAL)
    public Uni<Void> processUserEvent(KafkaRecord<String, String> record) {
        return Uni.createFrom().item(record.getPayload())
                .onItem().transform(payload -> {
                    String[] parts = payload.split(":");
                    if (parts.length >= 3) {
                        String eventId = parts[0];
                        String eventType = parts[1];
                        Long userId = Long.parseLong(parts[2]);
                        
                        System.out.printf("Processing event: %s for user %d%n", eventType, userId);
                        
                        // Process the event
                        processEvent(eventType, userId);
                        
                        processedCount.incrementAndGet();
                    }
                    return null;
                })
                .onItem().transformToUni(v -> 
                    Uni.createFrom().completionStage(record.ack().toCompletableFuture())
                )
                .onFailure().recoverWithItem(failure -> {
                    System.err.println("Failed to process event: " + failure.getMessage());
                    return null;
                });
    }

    // Consume user actions with Object deserialization
    @Incoming("user-actions")
    public Uni<Void> processUserAction(UserAction action) {
        return Uni.createFrom().item(action)
                .onItem().invoke(a -> {
                    System.out.printf("Processing action: %s for user %d%n", 
                        a.actionType, a.userId);
                    
                    // Process the action
                    handleUserAction(a);
                })
                .onFailure().recoverWithItem(failure -> {
                    System.err.println("Failed to process action: " + failure.getMessage());
                    return null;
                });
    }

    // Process and forward events with transformation
    @Incoming("user-events")
    @Outgoing("processed-events")
    public Multi<String> transformAndForward(String payload) {
        return Multi.createFrom().item(payload)
                .onItem().transform(p -> {
                    // Add processing metadata
                    return "[PROCESSED] " + p + " | count: " + processedCount.get();
                })
                .onFailure().recoverWithItem("ERROR: Failed to process event");
    }

    // Stream processing: Count events per second
    @Incoming("user-events")
    @Outgoing("event-metrics")
    public Multi<EventMetric> calculateMetrics(String payload) {
        return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
                .onItem().transform(tick -> {
                    int count = processedCount.getAndSet(0);
                    return new EventMetric("user-events", count, System.currentTimeMillis());
                });
    }

    private void processEvent(String eventType, Long userId) {
        // Business logic for processing events
        switch (eventType) {
            case "USER_CREATED":
                System.out.println("Handling user creation: " + userId);
                break;
            case "USER_UPDATED":
                System.out.println("Handling user update: " + userId);
                break;
            case "USER_DELETED":
                System.out.println("Handling user deletion: " + userId);
                break;
            default:
                System.out.println("Unknown event type: " + eventType);
        }
    }

    private void handleUserAction(UserAction action) {
        // Business logic for handling user actions
        switch (action.actionType) {
            case "LOGIN":
                System.out.println("User logged in: " + action.userId);
                break;
            case "LOGOUT":
                System.out.println("User logged out: " + action.userId);
                break;
            case "PROFILE_UPDATE":
                System.out.println("User updated profile: " + action.userId);
                break;
            default:
                System.out.println("Unknown action type: " + action.actionType);
        }
    }

    public static class EventMetric {
        public String eventType;
        public int count;
        public long timestamp;
        
        public EventMetric(String eventType, int count, long timestamp) {
            this.eventType = eventType;
            this.count = count;
            this.timestamp = timestamp;
        }
    }
}

6. Advanced Reactive Patterns

Circuit Breaker Pattern

java

package com.example.patterns;

import io.smallrye.faulttolerance.api.CircuitBreakerName;
import io.smallrye.mutiny.Uni;
import org.eclipse.microprofile.faulttolerance.*;
import jakarta.enterprise.context.ApplicationScoped;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

@ApplicationScoped
public class ResilientService {

    private final AtomicInteger callCount = new AtomicInteger();
    private final AtomicInteger failureCount = new AtomicInteger();

    @CircuitBreaker(
        requestVolumeThreshold = 4,
        failureRatio = 0.5,
        delay = 10000,
        successThreshold = 2
    )
    @Timeout(5000)
    @Retry(maxRetries = 3, delay = 1000)
    @Bulkhead(value = 5, waitingTaskQueue = 10)
    @CircuitBreakerName("external-service")
    public Uni<String> callExternalServiceWithResilience(String input) {
        return Uni.createFrom().item(() -> {
            callCount.incrementAndGet();
            
            // Simulate occasional failures
            if (callCount.get() % 5 == 0) {
                failureCount.incrementAndGet();
                throw new RuntimeException("Simulated external service failure");
            }
            
            // Simulate processing time
            try { Thread.sleep(100); } catch (InterruptedException e) { }
            
            return "Processed: " + input;
        });
    }

    @Fallback(fallbackMethod = "fallbackResponse")
    public Uni<String> callWithFallback(String input) {
        return Uni.createFrom().item(() -> {
            if (input.equals("fail")) {
                throw new RuntimeException("Forced failure");
            }
            return "Success: " + input;
        });
    }

    private Uni<String> fallbackResponse(String input) {
        return Uni.createFrom().item("Fallback for: " + input);
    }

    // Async bulkhead pattern
    @Bulkhead(value = 3, waitingTaskQueue = 20)
    public Uni<String> processWithBulkhead(String task) {
        return Uni.createFrom().item(task)
                .onItem().delayIt().by(Duration.ofMillis(500))
                .onItem().transform(t -> "Completed: " + t);
    }
}

Reactive Caching

java

package com.example.patterns;

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.Multi;
import jakarta.enterprise.context.ApplicationScoped;

import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;

@ApplicationScoped
public class ReactiveCacheService {

    private final ConcurrentHashMap<String, CacheEntry> cache = new ConcurrentHashMap<>();

    public Uni<String> getWithCache(String key) {
        CacheEntry entry = cache.get(key);
        
        if (entry != null && !entry.isExpired()) {
            return Uni.createFrom().item(entry.value);
        }
        
        return computeValue(key)
                .onItem().transform(value -> {
                    cache.put(key, new CacheEntry(value, Duration.ofMinutes(5)));
                    return value;
                });
    }

    public Uni<String> computeValue(String key) {
        return Uni.createFrom().item(() -> {
            // Simulate expensive computation
            try { Thread.sleep(200); } catch (InterruptedException e) { }
            return "Computed value for: " + key;
        });
    }

    public Multi<String> streamWithCache(String key, int count) {
        return Multi.createFrom().range(1, count + 1)
                .onItem().transformToUniAndMerge(i -> 
                    getWithCache(key + "-" + i)
                );
    }

    public Uni<Void> invalidateCache(String key) {
        return Uni.createFrom().item(() -> {
            cache.remove(key);
            return null;
        });
    }

    private static class CacheEntry {
        final String value;
        final long expiryTime;
        
        CacheEntry(String value, Duration ttl) {
            this.value = value;
            this.expiryTime = System.currentTimeMillis() + ttl.toMillis();
        }
        
        boolean isExpired() {
            return System.currentTimeMillis() > expiryTime;
        }
    }
}

This extensive guide covers reactive programming in Quarkus with Mutiny, including REST endpoints, reactive clients, Kafka messaging, and advanced patterns like circuit breakers and caching. The examples demonstrate how to build high-performance, non-blocking applications that can handle high concurrency efficiently.

Comments (0)

No comments yet. Be the first to comment!

QUARKUS BASIC

QUARKUS ADVANCED

QUARKUS INTERVIEW QUESTION

Quarkus Topic

Reactive Programming in Quarkus with Mutiny

1. What is Reactive Programming?

2. Why Use Reactive in Quarkus?

3. Reactive Libraries in Quarkus

Basic Uni Operations

Basic Multi Operations

3. Reactive REST Endpoints

Reactive JAX-RS Resource

Reactive Service Implementation

4. Reactive REST Client

REST Client Interface

Using Reactive REST Client

5. Reactive Messaging with Kafka

Configuration

Event Models

Reactive Messaging Producers

Reactive Messaging Consumers

6. Advanced Reactive Patterns

Circuit Breaker Pattern

Reactive Caching

Comments (0)

LecturreAI Assistant