Package org.gcube.informationsystem.resourceregistry.cluster

Class ClusterEventBus

java.lang.Object
org.gcube.informationsystem.resourceregistry.cluster.ClusterEventBus
public class ClusterEventBus extends Object
Broadcasts ClusterEvents to every service instance in the deployment, and delivers received ones to locally registered consumers.

Two independent channels. Every event is sent over both Redis pub/sub (the fast path) and an OrientDB Live Query on "ClusterEvent" (a slower, dependency-free backstop that keeps working when Redis is down — every deployment already requires OrientDB to be reachable, so this adds no new single point of failure). Neither channel is reliable on its own: Redis pub/sub offers no delivery guarantee, and a Live Query subscription can end by itself. Consumers that cannot tolerate a missed event must therefore reconcile against durable state rather than trust delivery; the bus buys latency, not correctness.

Exactly-once delivery is approximated, not guaranteed. Because both channels carry the same event, and because Redis echoes a message back to its own publisher, an event can arrive up to three times. Two filters reduce that to one local delivery: events whose ClusterEvent.getOriginId() is this instance are dropped (they were already delivered synchronously by publish(ClusterEventType, String)), and events whose ClusterEvent.getEventId() has been seen recently are dropped as duplicates. A consumer should still be written to tolerate a repeat, since the de-duplication window is bounded.

Consumers must not block. They run on the Redis pub/sub listener thread or on an OrientDB Live Query thread. A consumer that performs real work must hand it off — the projection worker, for instance, only records that a rebuild is due and lets its own thread perform it.

Registering is not activating. subscribe(ClusterEventType, Consumer) only records a consumer and performs no I/O; start() establishes the two receiving channels and is called once, at service startup. The split is required, not stylistic: consumers register from paths that run while a database session is being opened, so acquiring a session during registration would re-enter those paths and recurse without end.

Author:
Luca Frosini (ISTI - CNR)

  • Method Details

    • getInstance

      public static ClusterEventBus getInstance()
      Returns the singleton instance of ClusterEventBus.
      Returns:
      the singleton instance

    • getInstanceId

      public static String getInstanceId()
      Returns the identifier of this instance, as stamped on every event it emits.
      Returns:
      this instance's identifier

    • subscribe

      public void subscribe(ClusterEventType eventType, Consumer<ClusterEvent> consumer)
      Registers a consumer for one kind of event. The consumer is invoked for events emitted by this instance and by any other one alike.

      Performs no I/O, deliberately. Registration must stay callable from anywhere, including from low-level code that runs while a database session is being established — ODatabasePoolManager registers its own consumer from inside getPool(...). Opening a session here would re-enter that path and recurse without end. Receiving is therefore activated separately, by start().

      Parameters:
      eventType - the kind of event to receive
      consumer - the callback, which must return promptly (see the class-level note on blocking)

    • start

      public void start()
      Activates both receiving channels, once per JVM. Called at service startup, after the database environment is usable; until it runs, this instance still emits events and still delivers them to its own consumers, but does not observe events emitted elsewhere.

      Never throws: each channel is established independently, and failing to establish one leaves the other in place.

    • publish

      public void publish(ClusterEventType eventType, String payload)
      Emits an event to every instance, this one included: local consumers are invoked synchronously before the remote channels are written, so the emitting instance never has to wait for a network round trip to see the effect of its own change.

      Never throws: a transport failure is logged and the other channel still carries the event. If both fail, receivers are left to their own reconciliation.

      Parameters:
      eventType - the kind of event
      payload - the type-specific payload, may be null

    • shutdown

      public void shutdown()
      Releases the Live Query subscription and its reserved session. The Redis subscription is released with the shared Redis connection, which this bus does not own.