Package com.opencqrs.framework.eventhandler

Class EventHandlingProcessor

java.lang.Object

com.opencqrs.framework.eventhandler.EventHandlingProcessor

All Implemented Interfaces:

Runnable

public class EventHandlingProcessor
extends Object
implements Runnable

Asynchronous event processor observing an event stream to be handled by matching EventHandlerDefinitions all belonging to the same processing group and partition with configurable progress tracking and retry in case of errors.

See Also:

• run()
• start()
• stop()

Constructor Summary

Constructors

Constructor	Description
EventHandlingProcessor(long partition, String subject, Boolean recursive, EventReader eventReader, ProgressTracker progressTracker, EventSequenceResolver eventSequenceResolver, PartitionKeyResolver partitionKeyResolver, List<EventHandlerDefinition> eventHandlerDefinitions, BackOff backoff)	Creates a pre-configured instance of this.

Method Summary

Modifier and Type	Method	Description
String	getGroupId()
long	getPartition()
void	run()	Enters the event processing loop, running infinitely unless interrupted or CqrsFrameworkException.NonTransientException is thrown.
Future<?>	start()	Starts this using a Executors.newFixedThreadPool(int) with size 2.
void	stop()	Stops this by shutting down the thread pool initialized during start().

Methods inherited from class java.lang.Object

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

Constructor Details

EventHandlingProcessor

public EventHandlingProcessor(long partition,
 String subject,
 Boolean recursive,
 EventReader eventReader,
 ProgressTracker progressTracker,
 EventSequenceResolver eventSequenceResolver,
 PartitionKeyResolver partitionKeyResolver,
 List<EventHandlerDefinition> eventHandlerDefinitions,
 BackOff backoff)

Creates a pre-configured instance of this.

Parameters:

partition - the partition number handled by this with respect to the processing group

subject - the subject to observe

recursive - whether the subject should be observed recursively, that is including child subjects

eventReader - the event source

progressTracker - the progress tracker to maintain the progress within the observed event stream

eventSequenceResolver - the event sequence resolver to determine the event sequence id

partitionKeyResolver - the partition key resolver to determine if the event needs to be handled this

eventHandlerDefinitions - a list of EventHandlerDefinition to dispatch events to

backoff - a configurable back-off strategy for retryable errors

Method Details

getPartition

public long getPartition()

getGroupId

public String getGroupId()

run

public void run()

Enters the event processing loop, running infinitely unless interrupted or CqrsFrameworkException.NonTransientException is thrown. This involves:

1. fetching the ProgressTracker.current(String, long) current progress} for the configured processing group and partition
2. observing the event stream for the configured subject starting from the current progress
3. checking if the raw event's sequence id is relevant for this partition, otherwise skip it
4. upcasting any observed event
5. resolving the Java event type
6. converting the upcasted event to a Java object
7. checking if the converted event's sequence id is relevant for this partition, otherwise skip it
8. passing the event (and associated information) to each matching EventHandler
9. ProgressTracker.proceed(String, long, Supplier) proceeding the progress} of the event handling loop iteration (also for non-relevant events previously skipped)

Errors or exceptions occurring throughout the event processing loop are handled as follows:

• CqrsFrameworkException.NonTransientExceptions thrown by any matching EventHandler won't be retried and will terminate the processing loop unrecoverably
• any other Throwable thrown by any matching EventHandler is subject to retry
• CqrsFrameworkException.TransientExceptions thrown by framework components are subject to retry
• any thread interruption before or after calling the EventHandler will terminate the processing loop, assuming this was stopped
• any other Throwable thrown by any of the framework components won't be retried and will terminate the processing loop unrecoverably

Retry of failed EventHandlers or framework components will cause the event processor to back off from the event processing loop, waiting before retrying the failed event according to the aforementioned event processing loop. Once the back off is exhausted the erroneous event will be skipped, continuing with the next observable event, once available.

This method is assumed to be started within a thread pool with one additional spare thread, which is used to dispatch any raw Event received via EventReader.consumeRaw(EventReader.ClientRequestor, BiConsumer). This effectively offloads event upcasting, type resolution, deserialization, and the actual event handling from the underlying HttpClient HttpResponse.BodySubscriber thread, in order to be able to stop() this properly.

Specified by:

run in interface Runnable

start

public Future<?> start()

Starts this using a Executors.newFixedThreadPool(int) with size 2. The first thread within the pool is used to start the event processing loop, while the second one is used for the raw Event dispatching.

Returns:

a future for the event processing loop to determine, when it ends (prematurely)

stop

public void stop()

Stops this by shutting down the thread pool initialized during start().