Package com.framed.arn

Class Reactor

java.lang.Object
com.framed.core.Service
com.framed.arn.Reactor
Direct Known Subclasses:
DislocationClassificationReactor, HemodynamicRhythmClassificationReactor, InterpretationReactor, LimitClassificationReactor, RespiratoryRateEstimationReactor, RRMismatchClassificationReactor, SFComputationReactor, TrendClassificationReactor
public abstract class Reactor extends Service
A reactive multi-input Service that evaluates a set of firing rules on incoming messages, constructs exactly one consistent snapshot per evaluation cycle, and triggers a user-defined reactionFunction(Map) once whenever at least one rule is satisfied.

Concrete reactors extend this class, supply their firing rules and input/output channels via the constructor, and implement reactionFunction(Map) to compute and publish derived results (typically via RuleUtils.publishResult(com.framed.core.EventBus, java.lang.Object, java.lang.String, java.util.List<java.lang.String>, java.time.Instant)).

Core semantics

  • Each input channel maintains:
    • a monotonic absolute sequence counter (never reset), and
    • the latest JSON message value.
  • Each rule is a map of channel → token where tokens:
    • "*" = ANY: at least 1 new message since this rule last fired
    • "N" = AT_LEAST(N): at least N new messages since last fired
    • "r:v" = REQUIRE_VALUE(v): at least 1 new message AND latest value == v
  • Each rule maintains its own last-consumed sequence pointer per channel. This gives each rule independent delta semantics.
  • On each incoming message:
    • all rules are evaluated against a stable snapshot of latest data + seq counters;
    • every satisfied rule updates its per-channel pointers;
    • a single snapshot is created;
    • reactionFunction(Map) is called once;
    • latency is published according to all three modes (A, B, C2).

Snapshot semantics

The snapshot contains:
  • channel → value extracted from the JSON field "value"
  • channel-timestamp → Instant parsed from JSON "timestamp"
This snapshot is immutable and provided to reactionFunction(Map).

Latency publishing modes

After a firing, the reactor publishes:
  1. (A) Per-channel latency: For every input channel.
  2. (B) One global latency: Based on the earliest channel timestamp.
  3. (C2) Rule-participation latency: Only for channels whose delta ≥ 1 for at least one satisfied rule.
The latency messages are deduplicated per timestamp to avoid double publications.

Thread safety

All rule evaluation and delta accounting is done under a single lock, while reactionFunction(Map) is executed outside the lock.

  • Field Details

    • addressGroup

      protected final String addressGroup
      Producer group under which this reactor announces its output channels (see Service.announceAddress(String, String)). Defaults to "CDSS" so that sinks listing the "CDSS" device discover all reactor outputs; can be made a constructor parameter to support multiple reactor groups.
      See Also:

    • lastLogicalFireTs

      protected volatile Instant lastLogicalFireTs
      Last logical timestamp at which this reactor fired.

    • id

      protected final String id
      Actor instance identifier (used in latency tags, classification IDs, etc.).

    • inputChannels

      protected final List<String> inputChannels
      Input channels this reactor listens to.

    • outputChannels

      protected final List<String> outputChannels
      Output channels this reactor may publish to.

  • Constructor Details

    • Reactor

      protected Reactor(EventBus eventBus, String id, List<Map<String,String>> firingRules, List<String> inputChannels, List<String> outputChannels)
      Constructs a rule-based Actor.
      Parameters:
      eventBus - the event bus providing input messages and publishing outputs
      id - reactor identifier
      firingRules - list of rules (channel → token)
      inputChannels - list of channels this reactor subscribes to
      outputChannels - list of channels this reactor may publish to

    • Reactor

      protected Reactor(EventBus eventBus, String id, List<Map<String,String>> firingRules, List<String> inputChannels, List<String> outputChannels, boolean atomic)
      Constructs a rule-based reactor.
      Parameters:
      eventBus - the event bus providing input messages and publishing outputs
      id - reactor identifier
      firingRules - list of rules (channel → token)
      inputChannels - list of channels this reactor subscribes to
      outputChannels - list of channels this reactor may publish to
      atomic - if true, reactionFunction(Map) and latency publishing run inside the evaluation lock (serialized); if false, they run after the lock is released

  • Method Details

    • reactionFunction

      public abstract void reactionFunction(Map<String,Object> latestSnapshot)
      Called once per evaluation cycle when any rule is satisfied.
      Parameters:
      latestSnapshot - immutable snapshot of latest channel values and timestamps

    • getInputChannels

      public List<String> getInputChannels()
      Returns the input channels this reactor listens to.
      Returns:
      the immutable list of input channel names

    • getOutputChannels

      public List<String> getOutputChannels()
      Returns the output channels this reactor may publish to.
      Returns:
      the immutable list of output channel names

    • valueMatchesExpected

      protected boolean valueMatchesExpected(Object actual, String expected)
      Default equality check for the "r:v" (REQUIRE_VALUE) token: compares the latest value's string form against the expected value. Override to customize matching.
      Parameters:
      actual - the latest value received on the channel
      expected - the expected value string from the rule token
      Returns:
      true if the value matches