Package dev.agenor.core

Interface MessageHandler

Functional Interface:
This is a functional interface and can therefore be used as the assignment target for a lambda expression or method reference.
@FunctionalInterface public interface MessageHandler
Functional interface for handling received messages in the Agenor framework.

A MessageHandler is a callback that processes incoming messages. Handlers are registered with the MessageDispatcher to receive messages matching specific criteria (topics, receivers, or predicates).

Functional Interface: Being a @FunctionalInterface, handlers can be created using:

  • Lambda expressions
  • Method references
  • Anonymous classes

Asynchronous by Default: The handle(Message) method returns a CompletableFuture, enabling non-blocking, asynchronous message processing. This is crucial for:

  • Avoiding blocking the thread that delivers the message
  • Processing multiple messages concurrently
  • Chaining message processing operations
  • Error handling and recovery

Example - Lambda Handler: 


 // Simple lambda handler
 MessageHandler handler = message -> {
     log.info("Received: {}", message.content());
     return CompletableFuture.completedFuture(null);
 };

 // Subscribe
 topicSubscriber.subscribeTopic("notifications", handler);

Example - Method Reference: 


 public class OrderProcessor {

     public CompletableFuture<Void> handleOrder(Message message) {
         OrderData order = message.getContent(OrderData.class);
         return processOrderAsync(order);
     }

     public void setupSubscription(TopicSubscriber subscriber) {
         // Method reference as handler
         subscriber.subscribeTopic("orders", this::handleOrder);
     }
 }

Example - Async Processing: 


 MessageHandler asyncHandler = message -> {
     return CompletableFuture.supplyAsync(() -> {
         // Long-running operation in separate thread
         Data data = message.getContent(Data.class);
         Result result = expensiveOperation(data);

         // Send response
         Message response = message.reply(result)
             .senderId("processor")
             .build();
         messageService.send(response);

         return null;
     });
 };

Example - Synchronous Wrapper: Use sync(SyncMessageHandler) for simple synchronous handlers: 


 // Synchronous handler wrapped in async
 MessageHandler handler = MessageHandler.sync(message -> {
     log.info("Processing: {}", message.content());
     processSync(message);
     // No need to return CompletableFuture
 });

Error Handling: An unhandled exception completes the returned future exceptionally, and that outcome is the message's outcome: the mailbox does not acknowledge it, so on a transport with redelivery it is delivered again and eventually dead-lettered (ADR-033). Other handlers for the same message still run; the first failure is rethrown with the rest attached as suppressed exceptions.

A failure is therefore not contained by default. To contain one, catch it in the handler, where the decision is visible — and make handlers idempotent, because a redelivered message runs them again. 


 MessageHandler resilientHandler = message -> {
     return CompletableFuture.supplyAsync(() -> {
         try {
             processMessage(message);
             return null;
         } catch (ValidationException e) {
             log.warn("Invalid message: {}", e.getMessage());
             sendErrorReply(message, e);
             return null;
         } catch (Exception e) {
             log.error("Processing failed", e);
             // Rethrow to mark future as failed
             throw new CompletionException(e);
         }
     });
 };

Best Practices:

  • Keep handlers focused on a single responsibility
  • Use async processing for I/O or long-running operations
  • Handle errors explicitly; don't let exceptions propagate silently
  • Log at appropriate levels for debugging and monitoring
  • Consider idempotency for retry scenarios
  • Don't block; return futures for all long operations
Since:
0.1.0
See Also:

  • Method Details

    • handle

      CompletableFuture<Void> handle(Message message)
      Handles a received message asynchronously.

      This method is invoked when a message matching the subscription criteria is delivered. The handler should process the message and return a CompletableFuture that completes when processing is done.

      Asynchronous Processing: The method returns immediately with a future. Actual processing can be:

      • Immediate - Use CompletableFuture.completedFuture(null)
      • Async - Use CompletableFuture.supplyAsync(...)
      • Chained - Return an existing future from async operations

      Thread Safety: This method may be called concurrently from multiple threads for different messages. Implementations must be thread-safe if they access shared state.

      Error Handling:

      • Handle expected errors within the handler
      • Let unexpected errors complete the future exceptionally
      • Don't silently swallow exceptions

      Performance:

      • Don't block the calling thread
      • Use CompletableFuture.runAsync() for heavy processing
      • Consider thread pool sizing for concurrent messages

      Examples:

      Immediate completion: 

      
 public CompletableFuture<Void> handle(Message message) {
     log.info("Received: {}", message.content());
     return CompletableFuture.completedFuture(null);
 }

      Async processing: 

      
 public CompletableFuture<Void> handle(Message message) {
     return CompletableFuture.runAsync(() -> {
         expensiveOperation(message.content());
     });
 }

      Chained operations: 

      
 public CompletableFuture<Void> handle(Message message) {
     return fetchDataAsync()
         .thenCompose(data -> processAsync(data))
         .thenCompose(result -> sendReply(message, result))
         .exceptionally(ex -> {
             log.error("Handler failed", ex);
             return null;
         });
 }
      Parameters:
      message - the received message, never null
      Returns:
      a CompletableFuture that completes when processing is finished, or completes exceptionally if processing fails
      Throws:
      RuntimeException - if processing fails (will complete future exceptionally)
      See Also:

    • sync

      static MessageHandler sync(MessageHandler.SyncMessageHandler syncHandler)
      Creates an async handler wrapper for synchronous message handlers.

      This factory method simplifies creating handlers for simple synchronous operations. It wraps a MessageHandler.SyncMessageHandler that processes messages synchronously and returns void, converting it to the async MessageHandler interface.

      When to Use:

      • Quick, in-memory processing
      • Simple logging or routing logic
      • Operations that complete instantly

      When NOT to Use:

      • I/O operations (database, network, files)
      • CPU-intensive computations
      • Operations that could block

      Error Handling: An exception thrown by the sync handler is caught and returned as a failed future. That does not contain it: the failed future is how the failure reaches the mailbox, which then does not acknowledge the message (ADR-033). To contain a failure, catch it inside the handler body.

      Examples:

      Simple logging: 

      
 MessageHandler handler = MessageHandler.sync(message -> {
     log.info("Received message: {}", message.content());
 });

      Quick processing: 

      
 MessageHandler handler = MessageHandler.sync(message -> {
     String data = message.getContent(String.class);
     if (data != null) {
         cache.put(message.id(), data);
     }
 });

      With error handling: 

      
 MessageHandler handler = MessageHandler.sync(message -> {
     try {
         quickProcess(message);
     } catch (Exception e) {
         log.error("Processing failed", e);
         throw e;  // Becomes failed future
     }
 });
      Parameters:
      syncHandler - the synchronous message handler
      Returns:
      an async MessageHandler wrapping the synchronous handler
      Throws:
      NullPointerException - if syncHandler is null
      See Also: