Class WsClient

java/lib/src/main/java/io/github/ccxt/ws/WsClient.java:69–811

Netty-based WebSocket client for CCXT. Matches the architecture of C# Client.cs and Go exchange_wsclient.go: - Single async connection per URL - ConcurrentHashMap for futures and subscriptions - Ping/pong keep-alive loop - Reconnection via lazy re-connect on next watch() call

 * - Reconnection via lazy re-connect on next watch() call
 */
public class WsClient {

    private static final ObjectMapper JSON_MAPPER = new ObjectMapper();

    // Shared event loop group across all WsClient instances (one per JVM)
    private static final EventLoopGroup SHARED_EVENT_LOOP = new NioEventLoopGroup(
            Math.max(2, Runtime.getRuntime().availableProcessors()));

    static {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            SHARED_EVENT_LOOP.shutdownGracefully();
        }, "ccxt-ws-shutdown"));
    }

    // State — typed as Object for compatibility with transpiled code that casts these
    public String url;
    public Object futures = new ConcurrentHashMap<String, Future>();
    public Object subscriptions = new ConcurrentHashMap<String, Object>();
    public Object rejections = new ConcurrentHashMap<String, Object>();
    public volatile boolean isConnected = false;
    public final AtomicBoolean startedConnecting = new AtomicBoolean(false);
    public volatile long connectionEstablished = 0;
    public volatile CompletableFuture<Boolean> connected;
    public volatile long lastPong = 0;
    public boolean error = false;
    /**
     * Optional typed reason for a deliberate close. Set by Exchange.close()
     * to an ExchangeClosedByUser before invoking this.close(); the close path
     * then uses it to reject in-flight futures so callers can distinguish
     * "I asked for this" from a remote-side disconnect.
     */
    public volatile Throwable closeReason = null;

    // Guards atomic complete-then-replace of `connected` and ping-thread bookkeeping.
    private final Object connectedLock = new Object();
    private volatile Thread pingThread;

    // Typed accessors for internal use
    @SuppressWarnings("unchecked")
    private ConcurrentHashMap<String, Future> futuresMap() { return (ConcurrentHashMap<String, Future>) futures; }
    @SuppressWarnings("unchecked")
    public ConcurrentHashMap<String, Object> subscriptionsMap() { return (ConcurrentHashMap<String, Object>) subscriptions; }
    @SuppressWarnings("unchecked")
    private ConcurrentHashMap<String, Object> rejectionsMap() { return (ConcurrentHashMap<String, Object>) rejections; }

    // Config
    public long keepAlive = 30000;
    public int maxPingPongMisses = 3;
    public boolean verbose = false;
    public boolean validateServerSsl = true;
    public boolean decompressBinary = true;

    // Callbacks (set by Exchange)
    public BiConsumer<WsClient, Object> handleMessageCallback;
    public Function<WsClient, Object> pingCallback;
    public BiConsumer<WsClient, Object> onCloseCallback;
    public BiConsumer<WsClient, Object> onErrorCallback;