Class DefaultTracking

java.lang.Object

io.fluxzero.sdk.tracking.DefaultTracking

All Implemented Interfaces:

Tracking, AutoCloseable

public class DefaultTracking
extends Object
implements Tracking

Default implementation of the Tracking interface that coordinates message tracking for a specific MessageType.

This class is responsible for:

• Assigning handler objects to appropriate ConsumerConfigurations based on declared filters
• Creating and managing Tracker instances for those consumers and their associated topics
• Ensuring messages are deserialized, dispatched, and (if applicable) responded to with proper error handling
• Invoking handlers using the provided HandlerFactory and HandlerInvoker
• Integrating with ResultGateway to send back command/query/web responses when needed

Supports per-consumer batch interceptors and general batch processing logic, including:

• Functional and technical exception management with retry hooks
• Tracking exclusivity to prevent handlers from being assigned to multiple consumers simultaneously
• Internal shutdown coordination and pending message flushes via close()

Typical Usage

This class is used internally when starting a Fluxzero.registerHandlers(List) invocation for a given MessageType, and typically shouldn't be used directly by application developers.

See Also:

• Tracking
• ConsumerConfiguration
• Tracker
• ResultGateway

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static final record	DefaultTracking.PreparedHandler	A prepared handler and its effective tracking target after expansion, preserving instance-specific assignment.

Constructor Summary

Constructors

Constructor	Description
DefaultTracking(MessageType messageType, ResultGateway resultGateway, List<ConsumerConfiguration> customConfigurations, List<ConsumerConfiguration> defaultConfigurations, Serializer serializer, HandlerFactory handlerFactory)
DefaultTracking(MessageType messageType, ResultGateway resultGateway, List<ConsumerConfiguration> customConfigurations, List<ConsumerConfiguration> defaultConfigurations, Serializer serializer, HandlerFactory handlerFactory, BiConsumer<Object, ConsumerConfiguration> handlerInitializer)

Method Summary

Modifier and Type	Method	Description
void	close()	Shuts down all started trackers and waits briefly for asynchronous results (e.g. command responses) to complete.
protected Consumer<List<SerializedMessage>>	createConsumer(ConsumerConfiguration config, List<Handler<DeserializingMessage>> handlers)
protected List<DeserializingMessage>	deserializeMessageList(List<SerializedMessage> serializedMessages, String topic, TrackingClient trackingClient, Map<String, ChunkedDeserializingMessage> activeChunkedMessages, int recoveryMaxFetchSize)
protected Object	doHandle(DeserializingMessage message, HandlerInvoker h, Handler<DeserializingMessage> handler, ConsumerConfiguration config)
protected Object	doHandle(DeserializingMessage message, HandlerMethod<? super DeserializingMessage> h, Handler<DeserializingMessage> handler, ConsumerConfiguration config)
protected HandlerMethod<? super DeserializingMessage>	getHandlerMethodOrNull(DeserializingMessage message, Handler<DeserializingMessage> handler, ConsumerConfiguration config)
protected Optional<HandlerInvoker>	getInvoker(DeserializingMessage message, Handler<DeserializingMessage> handler, ConsumerConfiguration config)
protected Object	handle(DeserializingMessage message, HandlerInvoker h, Handler<DeserializingMessage> handler, ConsumerConfiguration config)
protected Object	handle(DeserializingMessage message, HandlerMethod<? super DeserializingMessage> h, Handler<DeserializingMessage> handler, ConsumerConfiguration config)
Registration	observeConsumers(Fluxzero fluxzero, List<?> handlers, BiFunction<ConsumerConfiguration, List<DefaultTracking.PreparedHandler>, Registration> observer)	Describes effective consumers using the ordinary tracking preparation without starting trackers, initializing schedules, or opening document readers.
protected Object	processError(Throwable e, DeserializingMessage message, HandlerDescriptor h, Handler<DeserializingMessage> handler, ConsumerConfiguration config, Callable<Object> retry)
protected Object	processError(Throwable e, DeserializingMessage message, HandlerInvoker h, Handler<DeserializingMessage> handler, ConsumerConfiguration config)
protected CompletionStage<Void>	reportResult(Object result, HandlerDescriptor h, DeserializingMessage message, ConsumerConfiguration config)
protected List<String>	retainTrackingTopics(ConsumerConfiguration configuration, List<Handler<DeserializingMessage>> activeHandlers, List<Handler<DeserializingMessage>> addedHandlers, Fluxzero fluxzero)
protected boolean	shouldSendResponse(HandlerDescriptor invoker, DeserializingMessage request, Object result, ConsumerConfiguration config)
Registration	start(Fluxzero fluxzero, List<?> handlers)	Starts tracking by assigning the given handlers to configured consumers and creating topic-specific or shared trackers.
protected Registration	startTracking(ConsumerConfiguration configuration, List<Handler<DeserializingMessage>> handlers, String topic, Fluxzero fluxzero)
protected void	stopTracker(DeserializingMessage message, Handler<DeserializingMessage> handler, Throwable e)
protected Set<String>	trackingTopics(List<Handler<DeserializingMessage>> handlers)
protected CompletionStage<Void>	tryHandle(DeserializingMessage message, Handler<DeserializingMessage> handler, ConsumerConfiguration config, boolean reportResult)

Methods inherited from class Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface Tracking

start

Constructor Details

DefaultTracking

public DefaultTracking(MessageType messageType,
 ResultGateway resultGateway,
 List<ConsumerConfiguration> customConfigurations,
 List<ConsumerConfiguration> defaultConfigurations,
 Serializer serializer,
 HandlerFactory handlerFactory)

DefaultTracking

public DefaultTracking(MessageType messageType,
 ResultGateway resultGateway,
 List<ConsumerConfiguration> customConfigurations,
 List<ConsumerConfiguration> defaultConfigurations,
 Serializer serializer,
 HandlerFactory handlerFactory,
 BiConsumer<Object, ConsumerConfiguration> handlerInitializer)

Method Details

start

public Registration start(Fluxzero fluxzero,
 List<?> handlers)

Starts tracking by assigning the given handlers to configured consumers and creating topic-specific or shared trackers.

Throws a TrackingException if handlers can't be matched to consumers.

Specified by:

start in interface Tracking

Parameters:

fluxzero - the owning Fluxzero instance

handlers - the handler instances to assign and activate

Returns:

a Registration that can be used to stop all created trackers

Throws:

TrackingException - if no consumer is found for a handler

observeConsumers

public Registration observeConsumers(Fluxzero fluxzero,
 List<?> handlers,
 BiFunction<ConsumerConfiguration, List<DefaultTracking.PreparedHandler>, Registration> observer)

Describes effective consumers using the ordinary tracking preparation without starting trackers, initializing schedules, or opening document readers. The observer owns its returned registration; this instance also cancels it on shutdown. Intended for synchronous test infrastructure validating consumer admission.

retainTrackingTopics

protected List<String> retainTrackingTopics(ConsumerConfiguration configuration,
 List<Handler<DeserializingMessage>> activeHandlers,
 List<Handler<DeserializingMessage>> addedHandlers,
 Fluxzero fluxzero)

trackingTopics

protected Set<String> trackingTopics(List<Handler<DeserializingMessage>> handlers)

startTracking

protected Registration startTracking(ConsumerConfiguration configuration,
 List<Handler<DeserializingMessage>> handlers,
 String topic,
 Fluxzero fluxzero)

createConsumer

protected Consumer<List<SerializedMessage>> createConsumer(ConsumerConfiguration config,
 List<Handler<DeserializingMessage>> handlers)

deserializeMessageList

protected List<DeserializingMessage> deserializeMessageList(List<SerializedMessage> serializedMessages,
 String topic,
 TrackingClient trackingClient,
 Map<String, ChunkedDeserializingMessage> activeChunkedMessages,
 int recoveryMaxFetchSize)

tryHandle

protected CompletionStage<Void> tryHandle(DeserializingMessage message,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config,
 boolean reportResult)

getHandlerMethodOrNull

protected HandlerMethod<? super DeserializingMessage> getHandlerMethodOrNull(DeserializingMessage message,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config)

getInvoker

protected Optional<HandlerInvoker> getInvoker(DeserializingMessage message,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config)

handle

protected Object handle(DeserializingMessage message,
 HandlerInvoker h,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config)

handle

protected Object handle(DeserializingMessage message,
 HandlerMethod<? super DeserializingMessage> h,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config)

doHandle

protected Object doHandle(DeserializingMessage message,
 HandlerInvoker h,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config)

doHandle

protected Object doHandle(DeserializingMessage message,
 HandlerMethod<? super DeserializingMessage> h,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config)

processError

protected Object processError(Throwable e,
 DeserializingMessage message,
 HandlerInvoker h,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config)

processError

protected Object processError(Throwable e,
 DeserializingMessage message,
 HandlerDescriptor h,
 Handler<DeserializingMessage> handler,
 ConsumerConfiguration config,
 Callable<Object> retry)

reportResult

protected CompletionStage<Void> reportResult(Object result,
 HandlerDescriptor h,
 DeserializingMessage message,
 ConsumerConfiguration config)

shouldSendResponse

protected boolean shouldSendResponse(HandlerDescriptor invoker,
 DeserializingMessage request,
 Object result,
 ConsumerConfiguration config)

stopTracker

protected void stopTracker(DeserializingMessage message,
 Handler<DeserializingMessage> handler,
 Throwable e)

close

public void close()

Shuts down all started trackers and waits briefly for asynchronous results (e.g. command responses) to complete. Incomplete chunked payloads fail before trackers are joined, releasing handlers waiting for missing input. Fully received payloads and ordinary asynchronous results retain their normal completion grace period.

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Tracking