Message interceptors

Fluxzero offers a flexible and extensible interceptor model to hook into key stages of the message lifecycle:

Interceptor type	Target phase	Typical use cases
Dispatch	Before publishing/handling a message	Mutate, block, enrich, or observe outgoing messages
Handler	Around handler execution	Validation, logging, authentication, result decoration
Batch	Around batch processing	Tracing, retries, context injection, metrics

All interceptors are pluggable, and can be configured via:

• FluxzeroBuilder for global registration
• @Consumer(handlerInterceptors = ...)
• @Consumer(batchInterceptors = ...)

---

Dispatch interceptor

A DispatchInterceptor hooks into the message dispatch phase—just before the message is published to Fluxzero or handled locally.

Java

public class LoggingInterceptor implements DispatchInterceptor {
    @Override
    public Message interceptDispatch(Message message, MessageType type, String topic) {
        log.info("Dispatching: {} to topic {}", type, topic);
        return message;
    }
}

Kotlin

class LoggingInterceptor : DispatchInterceptor {
    override fun interceptDispatch(message: Message, type: MessageType, topic: String): Message {
        log.info("Dispatching: {} to topic {}", type, topic)
        return message
    }
}

Capabilities:

• interceptDispatch(...): Modify, block, or inspect outgoing messages
• modifySerializedMessage(...): Mutate message after serialization but before transmission
• monitorDispatch(...): Observe the final message as it’s sent

Register globally: Assuming you are configuring a FluxzeroBuilder builder:

Java

    builder
        .addDispatchInterceptor(new LoggingInterceptor(), MessageType.COMMAND, MessageType.EVENT);

Kotlin

    builder
        .addDispatchInterceptor(LoggingInterceptor(), MessageType.COMMAND, MessageType.EVENT)

---

Handler interceptor

A HandlerInterceptor allows wrapping the execution of handler methods, ideal for:

• Authorization and access control — prevent unauthorized commands or queries based on the current user
• Auditing and logging — log incoming messages, handler invocations, or emitted results
• Validation hooks — perform extra validation before or after handler execution
• Result transformation — enrich or reformat results before they’re published or returned
• Thread context propagation — populate thread-local state like correlation IDs or security principals

Java

public class AuthorizationInterceptor implements HandlerInterceptor {
    @Override
    public Function<DeserializingMessage, Object> interceptHandling(
            Function<DeserializingMessage, Object> next, HandlerInvoker invoker) {
        return message -> {
            if (!isAuthorized(message)) {
                throw new UnauthorizedException();
            }
            return next.apply(message);
        };
    }
}

Kotlin

class AuthorizationInterceptor : HandlerInterceptor {
    override fun interceptHandling(
        next: Function<DeserializingMessage, Any>, invoker: HandlerInvoker
    ): Function<DeserializingMessage, Any> {
        return Function { message ->
            if (!isAuthorized(message)) {
                throw UnauthorizedException()
            }
            next.apply(message)
        }
    }
}

Register via annotation or builder:

Java

@Consumer(handlerInterceptors = AuthorizationInterceptor.class)
public class SecureCommandHandler { ... }

builder
    .addHandlerInterceptor(new AuthorizationInterceptor(), true, MessageType.COMMAND);

Kotlin

@Consumer(handlerInterceptors = [AuthorizationInterceptor::class])
class SecureCommandHandler { ... }

builder
    .addHandlerInterceptor(AuthorizationInterceptor(), true, MessageType.COMMAND)

Note

Here, true for highPriority means your interceptor runs before other interceptors.

---

Batch interceptor

Wraps around the processing of a full message batch by a single consumer, ideal for:

• Structured logging
• Performance instrumentation
• Scoped resources (e.g. transactions)

Java

public class LoggingBatchInterceptor implements BatchInterceptor {
    @Override
    public Consumer<MessageBatch> intercept(Consumer<MessageBatch> consumer, Tracker tracker) {
        return batch -> {
            log.info("Start processing {} messages", batch.size());
            consumer.accept(batch);
        };
    }
}

Kotlin

class LoggingBatchInterceptor : BatchInterceptor {
    override fun intercept(consumer: Consumer<MessageBatch>, tracker: Tracker): Consumer<MessageBatch> {
        return Consumer { batch ->
            log.info("Start processing {} messages", batch.size())
            consumer.accept(batch)
        }
    }
}

Global install: Assuming you are configuring a FluxzeroBuilder builder:

Java

    builder
        .addBatchInterceptor(new LoggingBatchInterceptor(), MessageType.EVENT);

Kotlin

    builder
        .addBatchInterceptor(LoggingBatchInterceptor(), MessageType.EVENT)

---

Mapping batch interceptor

This specialization of BatchInterceptor can rewrite or filter the batch itself:

Java

MappingBatchInterceptor filterTestMessages = (batch, tracker) -> {
    var filtered = batch.getMessages().stream()
            .filter(m -> !m.getMetadata().containsKey("testOnly"))
            .toList();
    return batch.withMessages(filtered);
};

Kotlin

val filterTestMessages = MappingBatchInterceptor { batch, _ ->
    val filtered = batch.messages.stream()
        .filter { !it.metadata.containsKey("testOnly") }
        .toList()
    batch.withMessages(filtered)
}

Install globally: Assuming you are configuring a FluxzeroBuilder builder:

Java

    builder
        .addBatchInterceptor(filterTestMessages, MessageType.QUERY);

Kotlin

    builder
        .addBatchInterceptor(filterTestMessages, MessageType.QUERY)

---

Interceptors are a central way to add cross-cutting behavior across all stages of message flow, from dispatch to handling and batching — empowering modular, observable, and policy-driven systems.

---

Summary

Interceptor	Runs at	Purpose
DispatchInterceptor	Before a message is published or handled locally	Mutate, block, enrich, or log outgoing messages
HandlerInterceptor	Around handler method execution	Authorization, validation, logging, result transformation
BatchInterceptor	Around an entire batch of messages	Tracing, retries, metrics, resource scoping
MappingBatchInterceptor	Specialized batch interceptor	Rewrite or filter whole batches before processing