Package com.ultikits.ultitools.utils

Class AgentProtocol

java.lang.Object
com.ultikits.ultitools.utils.AgentProtocol
final class AgentProtocol extends Object
The UltiCloud protocol v2 layer of one agent connection attempt (as of 6.3.1): the versioned handshake, the per-connection limits, the reliable-event resend and acknowledgement, the overflow report, inbound control errors and the live CloudChannel.

Handshake (D-12, D-13, contract SEMANTICS 10). onOpen() sends hello as the connection's first frame; nothing else is sent until a welcome answers it. A welcome counts only when it replies to that hello, matches the frozen schema and keeps every limit inside the contract bounds (S-51); otherwise it is answered with error invalid_message, the socket is closed with 1000 and the normal backoff follows. A valid welcome naming another server id than the linked one stops the connection for good, credential kept (contract answer 21, SEMANTICS 16). Only a valid, matching welcome makes the connection established: the backoff counters reset, the heartbeat starts (AgentConnection.markEstablished()) and channel() opens.

Reliable events (D-11, D-15, contract answers 3 and 4). After the welcome every unacknowledged outbox entry is sent again with its own envelope id and timestamp, then a pending drop count is reported once as outbox.overflow and cleared only when that frame was handed to the socket. An ack { ids } removes exactly those ids from the outbox.

Capabilities (contract SEMANTICS 8). An outbound frame whose capability the welcome did not accept is not sent; an inbound request whose capability the welcome did not accept, or whose operator switch is off, is refused with capability_not_accepted by the AgentMessageRouter. The welcome's accepted list is intersected with what the hello declared, so a gateway cannot widen it.

Inbound errors (SEMANTICS 6, 14). An error with replyTo reaches every reply listener and logs one WARNING per error code per minute; an error rate_limited without replyTo is logged at FINE, at most once a minute at INFO. An error is never answered. No log line carries a payload, a message from the gateway or a credential (S-34).

Threads and locks. Inbound frames arrive on the agent scheduler thread. Every frame this class hands to the socket, from any thread, is serialised on one private lock, which is held only while the frame is handed over (AgentConnection.send(String) never blocks) and never while calling a listener, the router or a connection method that calls out. In the session's lock order this lock sits below the session monitor and above the AgentConnection monitor and the CloudOutbox and CommandDedupeSet leaf monitors.

  • Field Details

    • MODULE_CAPABILITY_SENTINEL

      static final String MODULE_CAPABILITY_SENTINEL
      The routing sentinel module.request carries instead of a capability name.
      See Also:

    • NOTICE_INTERVAL_MS

      static final long NOTICE_INTERVAL_MS
      How often one notice of a kind may be logged above FINE.
      See Also:

    • MAX_CAPABILITIES

      static final int MAX_CAPABILITIES
      The schema bound on the number of capabilities in hello and welcome.
      See Also:

    • MAX_CAPABILITY_CHARS

      static final int MAX_CAPABILITY_CHARS
      The schema bound on the length of one capability name.
      See Also:

    • MAX_ACK_IDS

      static final int MAX_ACK_IDS
      The schema bound on the ids of one ack.
      See Also:

    • MAX_TIMESTAMP

      private static final long MAX_TIMESTAMP
      See Also:

    • ERROR_CODE

      private static final Pattern ERROR_CODE

    • WELCOME_KEYS

      private static final String[] WELCOME_KEYS

    • lock

      private final Object lock

    • connection

      private final AgentConnection connection

    • outbox

      private final CloudOutbox outbox

    • dedupe

      private final CommandDedupeSet dedupe

    • expectedServerId

      private final String expectedServerId

    • capabilities

      private final Supplier<List<String>> capabilities

    • settings

      private final Supplier<com.google.gson.JsonObject> settings

    • capabilityEnabled

      private final Predicate<String> capabilityEnabled

    • clock

      private final LongSupplier clock

    • log

      private final BiConsumer<Level,String> log

    • router

      private final AgentMessageRouter router

    • framed

      private final FramedChannel framed

    • channel

      private final CloudChannel channel

    • replyListeners

      private final List<Consumer<CloudRequest>> replyListeners

    • lifecycleListeners

      private final List<AgentProtocol.Listener> lifecycleListeners

    • helloId

      private String helloId

    • declared

      private Set<String> declared

    • welcomeSeen

      private boolean welcomeSeen

    • closed

      private boolean closed

    • lastNotice

      private final Map<String,Long> lastNotice

    • established

      private volatile boolean established

    • limits

      private volatile CloudLimits limits

    • accepted

      private volatile Set<String> accepted

    • gatewayClockOffsetMillis

      private volatile long gatewayClockOffsetMillis

  • Constructor Details

  • Method Details

    • router

      AgentMessageRouter router()
      Returns:
      the router of this connection; track handlers are registered on it (22-21)

    • channel

      CloudChannel channel()
      Returns:
      the live channel: open only between a valid welcome and the close

    • gatewayClockOffsetMillis

      long gatewayClockOffsetMillis()
      Returns:
      the welcome's serverTime minus the local time at its receipt; 0 before welcome

    • isEstablished

      boolean isEstablished()
      Returns:
      whether a valid, matching welcome answered the hello and the connection is not closed

    • limits

      CloudLimits limits()
      Returns:
      the accepted limits, or the contract defaults before welcome

    • acceptedCapabilities

      Set<String> acceptedCapabilities()
      Returns:
      the capabilities the welcome accepted (within those the hello declared); empty before

    • capabilitiesChangedSinceHello

      boolean capabilitiesChangedSinceHello()
      Whether a hello built now would declare another capability set than this connection's hello did (22-22 fix): the set is negotiated only at the handshake, so such a change needs a new connection. Builds the set exactly as onOpen() does.
      Returns:
      false before the hello was sent, or when the set is unchanged

    • reconnectForNewHello

      void reconnectForNewHello(String reason)
      Closes this connection normally (1000) and lets the connection retry after its normal backoff, so the next hello declares the current capability set (22-22 fix). Asynchronous; a no-op unless the socket is open.
      Parameters:
      reason - the reason for the retry line

    • addReplyListener

      void addReplyListener(Consumer<CloudRequest> listener)
      Parameters:
      listener - receives every inbound error that carries a replyTo, on the agent thread; it must not block

    • addLifecycleListener

      void addLifecycleListener(AgentProtocol.Listener listener)
      Parameters:
      listener - receives this connection's establishment and close; a listener added after the connection was established is told at once

    • onOpen

      void onOpen()
      The upgrade was accepted: sends hello, the connection's first frame. Once only.

    • onText

      void onText(String text)
      One inbound text frame other than the heartbeat reply.
      Parameters:
      text - the frame, never logged

    • onClose

      void onClose()
      The connection closed: the channel stops sending and lifecycle listeners are told once.

    • sendReliable

      boolean sendReliable(CloudOutbox.Entry entry)
      Sends one reliable event that was just added to the outbox, if the connection is established and its capability accepted; otherwise it waits in the outbox for the next welcome.
      Parameters:
      entry - the new outbox entry, sent with its own envelope id
      Returns:
      whether it was handed to the socket

    • onWelcome

      private void onWelcome(Envelope envelope, List<Runnable> after)

    • refuseBeforeWelcome

      private void refuseBeforeWelcome(InvalidFrameException invalid, List<Runnable> after)
      A frame that fails the envelope check before welcome. Only a welcome may arrive then, so the frame is treated as a broken welcome (for example the vendored missing-replyTo.json): it is answered with error invalid_message (replyTo the frame's id when that id is a readable ULID, otherwise none) and the attempt is abandoned through the same path as a welcome that breaks the contract, instead of waiting for the welcome deadline.

    • applyLimits

      private void applyLimits(CloudLimits accepted)
      Under the lock: the welcome's limits reach the outbox and the command-dedupe set (D-13).

    • resendUnacked

      private void resendUnacked()
      Under the lock: every unacknowledged entry, with its own id and timestamp (D-11).

    • reportOverflow

      private void reportOverflow()
      Under the lock: one outbox.overflow, cleared only once handed to the socket (D-15).

    • sendEntry

      private boolean sendEntry(CloudOutbox.Entry entry)
      Under the lock.

    • fits

      private boolean fits(String text)

    • onAck

      private void onAck(Envelope envelope)

    • ackIds

      private static List<String> ackIds(Envelope envelope)

    • onError

      private void onError(Envelope envelope, List<Runnable> after)

    • notice

      private void notice(Level level, String key, String message)
      Under the lock: logs message at level at most once per NOTICE_INTERVAL_MS per key, otherwise at FINE.

    • sendErrorUnlocked

      private void sendErrorUnlocked(String replyTo, String code, String message)
      Under the lock: an error frame sent even before welcome (the refused welcome's answer).

    • inboundAllowed

      private boolean inboundAllowed(String capability)
      The router's gate: accepted at the handshake and switched on by the operator.

    • outboundAllowed

      private boolean outboundAllowed(String type)

    • handOff

      private boolean handOff(String text)
      The framed channel's socket hand-off: serialised, and only while established.

    • helloCapabilities

      private Set<String> helloCapabilities()

    • helloSettings

      private com.google.gson.JsonObject helloSettings()

    • safely

      private void safely(Runnable action)