Package 

Interface StreamEventAggregator


  • 
    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 Boolean offer(String raw) Enqueues a raw event for processing.
      abstract Unit onEvent(SuspendFunction1<Object, Unit> handler) Registers the handler invoked for each delivery.
      abstract Result<Unit> stop() Stops both coroutines and releases resources.
      • Methods inherited from class java.lang.Object

        clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
    • Method Detail

      • start

         abstract Result<Unit> start()

        Starts the collector and dispatcher coroutines.

      • offer

         abstract Boolean offer(String raw)

        Enqueues a raw event for processing. Non-suspending, suitable for WebSocket callbacks.

        Returns false if 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:

        Product SDKs distinguish using is StreamAggregatedEvent<*>.

        Parameters:
        handler - A suspending function receiving the event or aggregated event.
      • stop

         abstract Result<Unit> stop()

        Stops both coroutines and releases resources. Idempotent.