Package com.framed.streamer.dispatcher

Class CountingDispatcher

java.lang.Object
com.framed.core.Service
com.framed.io.dispatch.Dispatcher
com.framed.streamer.dispatcher.CountingDispatcher
public class CountingDispatcher extends Dispatcher
A sink that performs no I/O and only counts what reaches it — the measurement instrument for throughput and latency benchmarks.

Every other Dispatcher in this module writes somewhere (a file, InfluxDB), so a run against them measures the sink as much as the framework. This one deliberately does nothing but increment counters, isolating the cost of the acquisition path itself: event-bus dispatch, DataPoint parsing and the async push queue. It answers three questions per run:

  • Keep-up — how many datapoints arrived, per channel, and how many the bounded push queue had to drop.
  • Throughput — datapoints per second between the first and last push.
  • End-to-end latency — now − dataPoint.timestamp() at push time. Valid as an emit→sink latency whenever producer and sink share a clock (same host); across hosts it also carries the clock offset and should not be read as latency.

A summary line is printed on stop(), so a full-app deployment emits it via the launcher's shutdown hook; in-process harnesses can read the same figures from snapshot().

Configuration keys

  • devices — JSON array of device identifiers whose announced channels to count.
  • channels — JSON array of channel identifiers to count; empty counts every channel the announced devices produce.

The channel filter scopes the measurement, not the traffic: a datapoint on an unlisted channel still crosses the bus, is still parsed, still occupies the push queue and still counts as activity for awaitQuiescence(java.time.Duration, java.time.Duration) — it is only left out of the figures. That is what makes a filtered run comparable to an unfiltered one, and it is why the filter is the right instrument when one producer group carries several stages of a pipeline. A reactor network, for instance, announces every reactor's output under one group, so an unfiltered sink mixes the intermediate results into the latency distribution of the final one.

Threading: push(com.framed.io.dispatch.DataPoint<?>) runs on the dispatcher's single worker thread and onDrop(com.framed.io.dispatch.DataPoint<?>, java.lang.Throwable) may run on either that thread or an event-bus handler thread; all counter updates are taken under a private lock, so snapshot() is safe to call from any thread at any time and always observes a mutually consistent set of figures.

  • Constructor Details

    • CountingDispatcher

      public CountingDispatcher(EventBus eventBus, org.json.JSONArray devices, org.json.JSONArray channels)
      Creates a counting sink bound to the announced channels of the given devices. Parameter names match the JSON configuration keys resolved by Factory; the EventBus is injected automatically.
      Parameters:
      eventBus - the event bus to discover channels and receive samples on
      devices - the device identifiers whose announced channels this sink binds to
      channels - the channel identifiers to count; empty counts every announced channel

  • Method Details

    • push

      public void push(DataPoint<?> dataPoint)
      Description copied from class: Dispatcher
      Pushes a single datapoint to the underlying sink.

      Invoked on the dispatcher's worker thread. Implementations may throw IOException to signal a transient failure, which triggers retry with exponential backoff.

      Specified by:
      push in class Dispatcher
      Parameters:
      dataPoint - the datapoint to push

    • pushBatch

      public void pushBatch(List<DataPoint<?>> batch)
      Description copied from class: Dispatcher
      Pushes a batch of datapoints to the underlying sink.
      Specified by:
      pushBatch in class Dispatcher
      Parameters:
      batch - the datapoints to push

    • onDrop

      protected void onDrop(DataPoint<?> dp, Throwable cause)
      Counts the drop instead of printing it. A benchmark can saturate the queue by design, and the inherited stderr trace would then cost more than the work being measured.
      Overrides:
      onDrop in class Dispatcher
      Parameters:
      dp - the datapoint that could not be queued or pushed
      cause - the reason the datapoint was dropped

    • onHandlerError

      protected void onHandlerError(String deviceID, String address, Object rawMsg, Exception e)
      Counts the failure instead of printing a stack trace, for the same reason as onDrop(com.framed.io.dispatch.DataPoint<?>, java.lang.Throwable).

      Not subject to the channel filter: a message that failed to parse has no channel identifier to filter on, so this count spans every channel the sink is bound to.

      Overrides:
      onHandlerError in class Dispatcher
      Parameters:
      deviceID - the identifier of the device whose message failed to be handled
      address - the channel address the failing message arrived on
      rawMsg - the raw, unparsed message payload
      e - the exception raised while handling the message

    • snapshot

      public CountingDispatcher.Snapshot snapshot()
      Returns a consistent snapshot of everything counted so far. Safe to call at any time from any thread; figures are cumulative, not reset by reading.
      Returns:
      the current counts, throughput and latency distribution

    • reset

      public void reset()
      Discards everything counted so far, as if the sink had just been constructed.

      For harnesses that warm up or prime the pipeline before the run they actually want to measure: without it, the setup traffic would show up in the counts, in the latency distribution and — because it moves the start of the timing window earlier — in the throughput figure. Call it while the sink is quiescent; concurrent pushes race with the clear and may survive it.

    • awaitQuiescence

      public boolean awaitQuiescence(Duration quietPeriod, Duration timeout) throws InterruptedException
      Blocks until the sink has been idle for quietPeriod, i.e. until the push queue has drained. Benchmarks need this before reading a final snapshot(): the producer finishing its last publish says nothing about the async queue behind this sink, and calling Dispatcher.shutdown(Duration) instead would drop whatever is still queued.

      Idleness is inferred from arrivals, so this reports "drained" for a producer that has merely stalled longer than quietPeriod. Callers that need a true end-of-stream should observe producer completion first and use this only to await the queue behind it.

      Every arrival counts as activity, including one the channel filter excludes from the figures: the queue this waits on is shared by the whole bound stream, so a filtered run and an unfiltered one drain at the same instant.

      Parameters:
      quietPeriod - how long nothing must arrive for the sink to count as drained
      timeout - give up after this long
      Returns:
      true if the sink went quiet, false if timeout elapsed first
      Throws:
      InterruptedException - if the calling thread is interrupted while waiting

    • stop

      public void stop()
      Prints the run summary, then stops the push worker. Invoked by the launcher's shutdown hook in a full-app deployment, which is what makes the figures visible without any extra tooling.

      The summary deliberately goes to System.out rather than through the inherited logger: java.util.logging's LogManager installs its own shutdown hook, and once it has reset the handlers every subsequent log record is silently discarded. Since this summary is the measurement, it cannot depend on winning that race.

      Overrides:
      stop in class Dispatcher