Class LiveTransferRegistry

java.lang.Object
ecmwf.ecpds.master.LiveTransferRegistry
public final class LiveTransferRegistry extends Object
The Class LiveTransferRegistry. In-memory, MasterServer-side registry of the latest LiveTransferSample received from every DataMover. Acts as the single "source of truth" for the "Live ECPDS Earth" globe visualisation: DataMovers push samples in here (via MasterInterface.updateLiveTransferStatistics(LiveTransferSample[])), and any number of listeners (e.g. a WebSocket broadcaster, added in a later phase) can subscribe to be notified in real-time, and/or poll getVisualizationSamples() for active movements and recent terminal events. Visualization state is bounded and ephemeral. Terminal events are exposed for 15 seconds and retained internally for 60 seconds to suppress delayed active replays. Authoritative byte accounting is separate in TrafficAccounting; the legacy bucket helpers below are not used by the globe KPI.

  • Method Details

    • getInstance

      public static LiveTransferRegistry getInstance()
      Gets the singleton instance.
      Returns:
      the instance

    • update

      public void update(LiveTransferSample[] samples)
      Update the registry with a freshly received batch of samples from a DataMover, and notify listeners.
      Parameters:
      samples - the samples

    • getBytesLast24h

      public long getBytesLast24h()
      Gets the total number of bytes transferred (across every Mover/ProxyHost) over the last rolling 24 hours. This is a single, MasterServer-side, always-on counter (unlike the per-connection "session" figure a globe client used to keep locally), so every open globe page - and every Monitor plugin instance - sees the exact same value, and it survives page reloads/reconnects. The underlying per-minute buckets are also periodically persisted to the SYS_CONFIG table (see snapshotBucketsForPersistence()/restoreBucketsFromPersistence(String)), so a normal MasterServer restart only loses at most a few minutes of history instead of the full 24 hours.
      Returns:
      the total bytes transferred in the last 24 hours

    • getBytesLast24h

      public long getBytesLast24h(String direction)
      Gets the total number of bytes transferred (across every Mover/ProxyHost) over the last rolling 24 hours, for a single direction. Same semantics/persistence as getBytesLast24h(), just filtered to one of LiveTransferSample.DIRECTION_DISSEMINATION or LiveTransferSample.DIRECTION_ACQUISITION, so the globe UI's "Transferred (24h)" KPI can match whichever of Dissemination/Acquisition/Both is currently selected.
      Parameters:
      direction - one of LiveTransferSample.DIRECTION_DISSEMINATION or LiveTransferSample.DIRECTION_ACQUISITION
      Returns:
      the total bytes transferred in the last 24 hours for that direction

    • getBytesLast24hByType

      public Map<String,Long> getBytesLast24hByType()

    • snapshotBucketsForPersistence

      public String snapshotBucketsForPersistence()
      Serializes every currently non-empty, non-stale (i.e. within the last 24h) per-minute bucket into a compact minute:dissBytes:acqBytes triples string, suitable for storage in a single SYS_CONFIG row. Meant to be called periodically (e.g. every few minutes) by the MasterServer, so a restart only loses the handful of minutes since the last save rather than the full rolling 24h window.
      Returns:
      the serialized snapshot, or an empty string if there is nothing (yet) to persist

    • restoreBucketsFromPersistence

      public void restoreBucketsFromPersistence(String data)
      Repopulates the per-minute buckets from a string previously produced by snapshotBucketsForPersistence(), e.g. on MasterServer startup, so the rolling 24h total does not simply reset to zero after a restart. Entries for minutes already outside the current rolling 24h window (i.e. the MasterServer was down for a while) are silently skipped, exactly as they would eventually age out of the live buckets anyway. Accepts both the current minute:dissBytes:acqBytes format and the legacy minute:bytes format (from before the Dissemination/Acquisition split), treating a legacy entry's single value as Dissemination bytes so upgrading a running MasterServer does not lose its rolling 24h total.
      Parameters:
      data - the previously persisted snapshot, as produced by snapshotBucketsForPersistence()

    • snapshotTypeBucketsForPersistence

      public Map<String,String> snapshotTypeBucketsForPersistence()
      Serializes each Host type separately, keeping snapshots within the SYS_CONFIG text column size.

    • restoreTypeBucketsFromPersistence

      public void restoreTypeBucketsFromPersistence(String type, String data)

    • getActiveTransfers

      public Collection<LiveTransferSample> getActiveTransfers()
      Gets a snapshot of every currently active transfer known to the registry.
      Returns:
      the active transfers

    • getVisualizationSamples

      public Collection<LiveTransferSample> getVisualizationSamples()
      Active movements plus short-lived terminal events, retained across independent Monitor polls.

    • addListener

      public void addListener(Consumer<LiveTransferSample> listener)
      Register a listener notified for every sample update (active or terminal).
      Parameters:
      listener - the listener

    • removeListener

      public void removeListener(Consumer<LiveTransferSample> listener)
      Remove a previously registered listener.
      Parameters:
      listener - the listener

    • isEnabled

      public boolean isEnabled()
      Whether the registry currently has at least one interested listener (e.g. an open globe WebSocket). Used by DataMovers to decide whether it is worth sampling/pushing live statistics at all.
      Returns:
      true, if is enabled

    • touch

      public void touch()
      Records that a remote consumer (the Monitor plugin's globe WebSocket broadcaster, polling via ManagementInterface.getLiveTransfers()) is actively interested right now. Used to extend isEnabled() for POLL_ENABLE_WINDOW_MS beyond each poll.

    • setEnabled

      public void setEnabled(boolean enabled)
      Force-enable the registry (e.g. via a configuration flag), regardless of listener count.
      Parameters:
      enabled - the enabled