Class AbstractWebsocketClient

java.lang.Object
io.fluxzero.sdk.common.websocket.AbstractWebsocketClient
All Implemented Interfaces:
WebsocketEndpoint, AutoCloseable
Direct Known Subclasses:
WebSocketEventStoreClient, WebsocketGatewayClient, WebsocketKeyValueClient, WebsocketSchedulingClient, WebSocketSearchClient, WebsocketTrackingClient
public abstract class AbstractWebsocketClient extends Object implements WebsocketEndpoint, AutoCloseable
Abstract base class for all WebSocket-based clients in the Fluxzero Java client.

This class provides robust connection management, message dispatching, result handling, batching, metrics publishing, and ping-based health checking. It underpins core components such as WebsocketGatewayClient, providing the shared infrastructure needed for durable, resilient WebSocket communication with the Fluxzero Runtime.

Core Responsibilities

  • Establishing and maintaining WebSocket connections with automatic reconnection support
  • Managing message sending and batching via Request and RequestBatch
  • Receiving and processing incoming RequestResult and ResultBatch messages
  • Supporting command guarantees (e.g., SENT, STORED) with retries and backpressure handling
  • Sending periodic ping frames to detect connection drops
  • Integrating with the Fluxzero metrics infrastructure for custom performance telemetry

Key Features

  • Session Pooling: Maintains multiple concurrent sessions to handle high-throughput scenarios
  • Request Backlogs: Each session has a backlog to buffer and batch outgoing requests
  • Ping Scheduling: Scheduled tasks detect broken sessions using WebSocket pings
  • Auto Retry: Failed requests are retried if the session is closed unexpectedly
  • Async Result Handling: Responses are handled on a separate thread pool to avoid blocking I/O
  • Metrics Publishing: Optional emission of message-related metrics based on configuration
See Also:

  • Field Details

    • CONNECTION_TIMEOUT_FAILSAFE_GRACE

      protected static final Duration CONNECTION_TIMEOUT_FAILSAFE_GRACE

    • CONNECTION_RETRY_LOG_INTERVAL

      protected static final int CONNECTION_RETRY_LOG_INTERVAL
      See Also:

    • CLIENT_HANDSHAKE_CONFIGURATOR_USER_PROPERTY

      protected static final String CLIENT_HANDSHAKE_CONFIGURATOR_USER_PROPERTY

    • CLIENT_SESSION_ID_USER_PROPERTY

      protected static final String CLIENT_SESSION_ID_USER_PROPERTY

    • RUNTIME_SESSION_ID_USER_PROPERTY

      protected static final String RUNTIME_SESSION_ID_USER_PROPERTY

    • NEGOTIATED_SESSION_ID_USER_PROPERTY

      protected static final String NEGOTIATED_SESSION_ID_USER_PROPERTY

    • RUNTIME_VERSION_USER_PROPERTY

      protected static final String RUNTIME_VERSION_USER_PROPERTY

    • SELECTED_COMPRESSION_ALGORITHM_USER_PROPERTY

      protected static final String SELECTED_COMPRESSION_ALGORITHM_USER_PROPERTY

    • SELECTED_TRANSPORT_FORMAT_USER_PROPERTY

      protected static final String SELECTED_TRANSPORT_FORMAT_USER_PROPERTY

    • defaultWebsocketConnector

      public static WebsocketConnector defaultWebsocketConnector

    • defaultObjectMapper

      public static com.fasterxml.jackson.databind.ObjectMapper defaultObjectMapper

  • Constructor Details

    • AbstractWebsocketClient

      public AbstractWebsocketClient(URI endpointUri, WebSocketClient client, boolean allowMetrics)
      Creates a WebSocket client using the given endpoint URI, client implementation, and a flag to allow metrics. Uses a default WebSocket container, default object mapper, and a single WebSocket session.
      Parameters:
      endpointUri - the URI of the WebSocket endpoint to connect to
      client - the client implementation that provides configuration and gateway access
      allowMetrics - whether metrics should be published for each request

    • AbstractWebsocketClient

      public AbstractWebsocketClient(URI endpointUri, WebSocketClient client, boolean allowMetrics, int numberOfSessions)
      Creates a WebSocket client with multiple parallel sessions using default settings. This constructor allows you to specify the number of WebSocket sessions to use, which is useful for increasing throughput and isolating message streams.
      Parameters:
      endpointUri - the URI of the WebSocket endpoint to connect to
      client - the client implementation that provides configuration and gateway access
      allowMetrics - whether metrics should be published for each request
      numberOfSessions - the number of WebSocket sessions to maintain concurrently

    • AbstractWebsocketClient

      public AbstractWebsocketClient(WebsocketConnector connector, URI endpointUri, WebSocketClient client, boolean allowMetrics, Duration reconnectDelay, com.fasterxml.jackson.databind.ObjectMapper objectMapper, int numberOfSessions)
      Constructs a WebSocket client with fine-grained control over connection setup. This constructor allows you to specify a custom container, reconnect delay, object mapper, and session count. It is primarily used for advanced configuration or test scenarios.
      Parameters:
      connector - the WebSocket connector to use for establishing connections
      endpointUri - the WebSocket server endpoint
      client - the client providing config and access to the Fluxzero Runtime
      allowMetrics - flag to enable or disable automatic metrics publishing
      reconnectDelay - the delay between reconnect attempts if the connection is lost
      objectMapper - the Jackson object mapper for (de)serializing requests and responses
      numberOfSessions - the number of WebSocket sessions to establish in parallel

  • Method Details

    • createConnectionRetryConfiguration

      protected io.fluxzero.common.RetryConfiguration createConnectionRetryConfiguration(URI endpointUri, Duration reconnectDelay)

    • logSuccessfulReconnect

      protected void logSuccessfulReconnect(URI endpointUri, io.fluxzero.common.RetryStatus status)

    • logConnectionRetryStatus

      protected void logConnectionRetryStatus(URI endpointUri, io.fluxzero.common.RetryStatus status)

    • connectToServer

      protected WebsocketSession connectToServer(WebsocketConnector connector, URI endpointUri) throws Exception
      Throws:
      Exception

    • getConnectionTimeoutFailsafeGrace

      protected Duration getConnectionTimeoutFailsafeGrace()

    • onOpen

      public void onOpen(WebsocketSession session)
      Description copied from interface: WebsocketEndpoint
      Called after the opening handshake has completed and the session metadata is available.
      Specified by:
      onOpen in interface WebsocketEndpoint
      Parameters:
      session - the newly opened session

    • send

      protected <R extends io.fluxzero.common.api.RequestResult> CompletableFuture<R> send(io.fluxzero.common.api.Request request)

    • sendAndWait

      protected <R extends io.fluxzero.common.api.RequestResult> R sendAndWait(io.fluxzero.common.api.Request request)

    • sendCommand

      protected CompletableFuture<Void> sendCommand(io.fluxzero.common.api.Command command)

    • onMessage

      public void onMessage(byte[] bytes, WebsocketSession session)
      Description copied from interface: WebsocketEndpoint
      Called when a complete binary message has been received.
      Specified by:
      onMessage in interface WebsocketEndpoint
      Parameters:
      bytes - the full binary message payload
      session - the session that received the message

    • onMessage

      public void onMessage(byte[] bytes, WebsocketSession session, WebsocketEndpoint.ReceiveTiming receiveTiming)
      Description copied from interface: WebsocketEndpoint
      Called when a complete binary message has been received, with low-level receive timing when available.
      Specified by:
      onMessage in interface WebsocketEndpoint
      Parameters:
      bytes - the full binary message payload
      session - the session that received the message
      receiveTiming - timing captured by the underlying websocket listener

    • captureReceiveTiming

      public boolean captureReceiveTiming()
      Description copied from interface: WebsocketEndpoint
      Returns whether this endpoint wants low-level receive timing for binary messages.
      Specified by:
      captureReceiveTiming in interface WebsocketEndpoint
      Returns:
      true when the websocket session should capture receive and dispatch timestamps

    • handleMessage

      protected void handleMessage(byte[] bytes, WebsocketSession session, WebsocketEndpoint.ReceiveTiming receiveTiming)

    • handleResult

      protected void handleResult(io.fluxzero.common.api.RequestResult result, String batchId, String sessionId)

    • handleResult

      protected void handleResult(io.fluxzero.common.api.RequestResult result, String batchId, String sessionId, io.fluxzero.sdk.common.websocket.WebsocketResultDiagnostics.ResultTiming clientResultTiming)

    • schedulePing

      protected AbstractWebsocketClient.PingRegistration schedulePing(WebsocketSession session)

    • sendPing

      protected void sendPing(WebsocketSession session)

    • onPong

      public void onPong(ByteBuffer message, WebsocketSession session)
      Description copied from interface: WebsocketEndpoint
      Called when a pong frame has been received.
      Specified by:
      onPong in interface WebsocketEndpoint
      Parameters:
      message - the pong application data
      session - the session that received the pong

    • handlePong

      protected void handlePong(WebsocketSession session)

    • abort

      protected void abort(WebsocketSession session, String reason)

    • onClose

      public void onClose(WebsocketSession session, WebsocketCloseReason closeReason)
      Description copied from interface: WebsocketEndpoint
      Called once when a session closes or is aborted.
      Specified by:
      onClose in interface WebsocketEndpoint
      Parameters:
      session - the closed session
      closeReason - the close status and optional reason

    • handleClose

      protected void handleClose(WebsocketSession session, WebsocketCloseReason closeReason)

    • retryOutstandingRequestsAsync

      protected void retryOutstandingRequestsAsync(String sessionId)

    • retryOutstandingRequests

      protected void retryOutstandingRequests(String sessionId)

    • retryOutstandingRequestsDelay

      protected Duration retryOutstandingRequestsDelay()
      Delay before outstanding requests from a closed session are resent on a replacement session.

    • onError

      public void onError(WebsocketSession session, Throwable e)
      Description copied from interface: WebsocketEndpoint
      Called when the underlying WebSocket implementation reports an error.
      Specified by:
      onError in interface WebsocketEndpoint
      Parameters:
      session - the session that failed
      e - the reported error

    • handleError

      protected void handleError(WebsocketSession session, Throwable e)

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable

    • toString

      public String toString()
      Overrides:
      toString in class Object

    • close

      protected void close(boolean clearOutstandingRequests)

    • tryPublishMetrics

      protected void tryPublishMetrics(io.fluxzero.common.api.JsonType message, io.fluxzero.common.api.Metadata metadata)

    • metricsMetadata

      protected io.fluxzero.common.api.Metadata metricsMetadata()

    • createConnectionSetup

      protected static AbstractWebsocketClient.ConnectionSetup createConnectionSetup(WebSocketClient.ClientConfig clientConfig)

    • getNegotiatedSessionId

      protected String getNegotiatedSessionId(WebsocketSession session)

    • getCompressionAlgorithm

      protected io.fluxzero.common.serialization.compression.CompressionAlgorithm getCompressionAlgorithm(WebsocketSession session)

    • getTransportFormat

      protected io.fluxzero.common.websocket.WebSocketTransportFormat getTransportFormat(WebsocketSession session)

    • transportCodec

      protected io.fluxzero.common.websocket.WebSocketTransportCodec transportCodec(WebsocketSession session)

    • getRuntimeVersion

      protected Optional<String> getRuntimeVersion(WebsocketSession session)