-
public interface StreamEventAggregator<T extends Object>Adaptive event aggregator that switches between individual and aggregated event delivery based on traffic volume.
Two decoupled coroutines:
Collector: Drains the inbox channel, groups events by type, and packages them into the dispatch queue. Collection ends when either aggregationThreshold is reached or maxWindowMs elapses — whichever comes first. This guarantees a latency ceiling regardless of downstream processing speed.
Dispatcher: Takes packaged work from the dispatch queue and invokes the registered handler. Runs sequentially — one delivery at a time, preserving order.
Low traffic: When the collector drains the channel and finds few events (below aggregationThreshold), each event is dispatched individually via the handler.
Spike: When accumulated events reach or exceed aggregationThreshold, they are grouped by type into a StreamAggregatedEvent and dispatched as a single call.
Natural: While the dispatcher is busy, events accumulate in the inbox channel. The next collection cycle finds them ready — larger accumulation triggers aggregation automatically.
The dispatch queue has a bounded capacity. If the dispatcher can't keep up, the collector logs a warning when the queue is full.
-
-
Method Summary
Modifier and Type Method Description abstract Result<Unit>start()Starts the collector and dispatcher coroutines. abstract Booleanoffer(String raw)Enqueues a raw event for processing. abstract UnitonEvent(SuspendFunction1<Object, Unit> handler)Registers the handler invoked for each delivery. abstract Result<Unit>stop()Stops both coroutines and releases resources. -
-
Method Detail
-
offer
abstract Boolean offer(String raw)
Enqueues a raw event for processing. Non-suspending, suitable for WebSocket callbacks.
Returns
falseif the aggregator has not been started or if the inbox is full/closed.- Parameters:
raw- The raw event (typically a JSON string from the WebSocket).
-
onEvent
abstract Unit onEvent(SuspendFunction1<Object, Unit> handler)
Registers the handler invoked for each delivery.
The handler receives either:
An individual event
T(low traffic), orA StreamAggregatedEvent
<T>wrapping grouped events (spike).
Product SDKs distinguish using
is StreamAggregatedEvent<*>.- Parameters:
handler- A suspending function receiving the event or aggregated event.
-
-
-
-