Package com.framed.communicator.driver.protocol.replay

Class ReplayProtocol

java.lang.Object
com.framed.core.Service
com.framed.io.protocol.Protocol
com.framed.communicator.driver.protocol.replay.ReplayProtocol
public class ReplayProtocol extends Protocol
Replays a recorded .jsonl session onto the event bus, on the parsed addresses the original parsers used.

A verification and measurement harness rather than a clinical data source: it drives a deployment from a real recording, so reactors and sinks can be exercised end to end without hardware. Each line of the recording is republished on "<className>.<deviceID>.<channelID>.parsed" with the same four-field envelope a parser emits.

Replay clock

The recording's relative timing is what the replay preserves, not its absolute dates. For an event recorded at t the scheduled emit time is

  emit(t) = replayStart + (t - firstRecordedTimestamp) / speed

computed in nanoseconds and stamped at microsecond resolution — the resolution Timer.formatter renders — so a recording timestamped to the microsecond is reproduced to the microsecond. Pacing itself is only as fine as Thread.sleep, but the stamp does not inherit that coarseness.

and that scheduled instant is also the timestamp the event carries on the bus. Two consequences make this the load-bearing design decision of the class:

  • Timestamp-sensitive logic sees the recording it was recorded from. At speed = 1.0 the emitted stamps are the recorded stamps shifted by a constant, so every inter-sample interval — and every set of samples that shared one device frame timestamp — is reproduced exactly. Reactors whose windows, staleness timers and logical-time gates key off those stamps therefore behave as they did on the original session. Stamping with Instant.now() instead, as this class previously did, splits a shared frame timestamp into distinct ones and shifts every window edge by the scheduling jitter, which changes which results a reactor network produces.
  • Downstream latency stays measurable. The stamps are contemporaneous with the run, so now − timestamp at a sink is an emit→sink latency (for a sink sharing the producer's clock) rather than the age of the recording.

The stamp is the scheduled emit time, not the actual one, so any lag of the replay thread itself is included in that latency instead of being hidden by re-stamping. The lag is reported separately by stats() and in the summary logged at the end of the replay.

At speed <= 0 the recording is replayed as fast as it can be published; there is then no schedule, and each event is stamped with the instant it is actually published. That mode measures throughput, and deliberately does not reproduce the recording's timing.

Repetition

repeat replays the recording that many times back to back. Pass r of repeat shifts every stamp by r × cycle, where cycle is the recording's own span plus one seam gap, so the emitted stream is a concatenation rather than an overlay: stamps stay strictly ordered across the seam and no instant is ever emitted twice. The seam gap is the smallest positive interval between two distinct recorded stamps, so it is never wider than a transition the recording already contains. The offset is applied in recorded time and then divided by speed like any other, so a seam occupies seamGap / speed of wall clock.

Repetition exists because a speed multiplier alone cannot hold a benchmark's sample size. Compressing a recording onto wall clock shortens the run but not a downstream reactor's window, which is denominated in seconds and therefore eats window × speed of recorded session before the first result appears — at high speed that is most of the recording. Repeating the recording proportionally to speed keeps the wall-clock duration, and so the usable fraction, constant across a sweep.

The seam is a discontinuity in the signal, not just the clock: the recording's last sample is followed by its first. That is harmless for throughput measurement but makes a repeated run unsuitable for comparison against a single-pass control.

Configuration keys

  • filePath — the .jsonl recording to replay.
  • devices — device identifiers to replay; other devices in the file are skipped.
  • channels — channel identifiers to replay; empty replays every channel.
  • speed — wall-clock speed multiplier (1.0 = real time, 0 = as fast as possible).
  • repeat — how many times to replay the recording back to back; 1 plays it once. Values below 1 are treated as 1.
  • startDelaySeconds — delay before the first event, leaving remote peers and sinks time to come up.
  • graceSeconds — how long to keep running after the last event, so downstream processing can finish.
  • exitOnCompletion — terminate the JVM after the grace period, which lets the launcher's shutdown hook flush and summarise the sinks. Set false when the replay is one service among others that should outlive it.

Threading: connect() starts one daemon-less replay thread; every publication happens on it. stop() interrupts that thread and returns without waiting for the remaining events.

  • Constructor Details

    • ReplayProtocol

      public ReplayProtocol(String id, EventBus eventBus, String filePath, org.json.JSONArray devices, org.json.JSONArray channels, double speed, int repeat, double startDelaySeconds, double graceSeconds, boolean exitOnCompletion)
      Creates a replay of filePath. Parameter names match the JSON configuration keys resolved by Factory; the EventBus is injected automatically.
      Parameters:
      id - service id
      eventBus - the bus to replay onto
      filePath - path to the .jsonl recording
      devices - device identifiers to replay; others in the file are skipped
      channels - channel identifiers to replay; empty replays every channel
      speed - wall-clock speed multiplier; <= 0 replays as fast as possible
      repeat - how many times to replay the recording back to back; below 1 is treated as 1
      startDelaySeconds - delay before the first event is published
      graceSeconds - how long to keep running after the last event
      exitOnCompletion - whether to terminate the JVM once the grace period has elapsed

  • Method Details

    • connect

      public void connect()
      Starts the replay on its own thread; a second call while one is running does nothing.
      Specified by:
      connect in class Protocol

    • stop

      public void stop()
      Interrupts the replay thread; does not wait for it to finish.
      Overrides:
      stop in class Service

    • stats

      public ReplayProtocol.ReplayStats stats()
      Returns how much the replay published and how well it held its schedule.

      Safe to call from any thread at any time, but the figures are computed once the last event has been published: before then this returns ReplayProtocol.ReplayStats.EMPTY. Use publishedEvents() to observe progress during a run.

      Returns:
      the pacing figures of this replay, or ReplayProtocol.ReplayStats.EMPTY while it is still running

    • publishedEvents

      public long publishedEvents()
      Returns how many events have been published so far. Updated as the replay runs, so it is also meaningful for a replay that was interrupted or is still in flight.
      Returns:
      the number of events published

    • awaitCompletion

      public boolean awaitCompletion(Duration timeout) throws InterruptedException
      Blocks until the replay has published its last event.
      Parameters:
      timeout - how long to wait
      Returns:
      true if the replay finished within timeout
      Throws:
      InterruptedException - if the calling thread is interrupted while waiting