Class AbstractWebsocketClient

java.lang.Object
jakarta.websocket.Endpoint
io.fluxzero.sdk.common.websocket.AbstractWebsocketClient
All Implemented Interfaces:
AutoCloseable
Direct Known Subclasses:
WebSocketEventStoreClient, WebsocketGatewayClient, WebsocketKeyValueClient, WebsocketSchedulingClient, WebSocketSearchClient, WebsocketTrackingClient
public abstract class AbstractWebsocketClient extends jakarta.websocket.Endpoint implements 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

    • 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

    • defaultWebSocketContainer

      public static jakarta.websocket.WebSocketContainer defaultWebSocketContainer

    • 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(jakarta.websocket.WebSocketContainer container, 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:
      container - the WebSocket container 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 RetryConfiguration createConnectionRetryConfiguration(URI endpointUri, Duration reconnectDelay)

    • logSuccessfulReconnect

      protected void logSuccessfulReconnect(URI endpointUri, RetryStatus status)

    • logConnectionRetryStatus

      protected void logConnectionRetryStatus(URI endpointUri, RetryStatus status)

    • connectToServer

      protected jakarta.websocket.Session connectToServer(jakarta.websocket.WebSocketContainer container, URI endpointUri) throws Exception
      Throws:
      Exception

    • getConnectionTimeoutFailsafeGrace

      protected Duration getConnectionTimeoutFailsafeGrace()

    • onOpen

      public void onOpen(jakarta.websocket.Session session, jakarta.websocket.EndpointConfig config)
      Specified by:
      onOpen in class jakarta.websocket.Endpoint

    • send

      protected <R extends RequestResult> CompletableFuture<R> send(Request request)

    • sendAndWait

      protected <R extends RequestResult> R sendAndWait(Request request)

    • sendCommand

      protected CompletableFuture<Void> sendCommand(Command command)

    • onMessage

      public void onMessage(byte[] bytes, jakarta.websocket.Session session)

    • handleMessage

      protected void handleMessage(byte[] bytes, jakarta.websocket.Session session)

    • handleResult

      protected void handleResult(RequestResult result, String batchId, String sessionId)

    • schedulePing

      protected AbstractWebsocketClient.PingRegistration schedulePing(jakarta.websocket.Session session)

    • sendPing

      protected void sendPing(jakarta.websocket.Session session)

    • onPong

      public void onPong(jakarta.websocket.PongMessage message, jakarta.websocket.Session session)

    • handlePong

      protected void handlePong(jakarta.websocket.Session session)

    • abort

      protected void abort(jakarta.websocket.Session session, String reason)

    • onClose

      public void onClose(jakarta.websocket.Session session, jakarta.websocket.CloseReason closeReason)
      Overrides:
      onClose in class jakarta.websocket.Endpoint

    • handleClose

      protected void handleClose(jakarta.websocket.Session session, jakarta.websocket.CloseReason closeReason)

    • retryOutstandingRequestsAsync

      protected void retryOutstandingRequestsAsync(String sessionId)

    • retryOutstandingRequests

      protected void retryOutstandingRequests(String sessionId)

    • onError

      public void onError(jakarta.websocket.Session session, Throwable e)
      Overrides:
      onError in class jakarta.websocket.Endpoint

    • handleError

      protected void handleError(jakarta.websocket.Session 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(JsonType message, Metadata metadata)

    • metricsMetadata

      protected Metadata metricsMetadata()

    • createConnectionSetup

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

    • getNegotiatedSessionId

      protected String getNegotiatedSessionId(jakarta.websocket.Session session)

    • getCompressionAlgorithm

      protected CompressionAlgorithm getCompressionAlgorithm(jakarta.websocket.Session session)

    • getRuntimeVersion

      protected Optional<String> getRuntimeVersion(jakarta.websocket.Session session)