WebSocket Code Example

Jave

package org.example.ws;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import jakarta.annotation.PostConstruct;
import jakarta.websocket.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.net.URI;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@ClientEndpoint
@Slf4j
@Component
public class WebsocketExample {

    // Local session channel
    private static Session session;

    // WebSocket subscription address — “business” can be stock, crypto, or common; “apikey” is your credential
    private static final String WS_URL = "wss://data.infoway.io/ws?business=crypto&apikey=yourApikey";

    @PostConstruct
    public void connectAll() {
        try {
            //Establish WebSocket connection
            connect(WS_URL);
            //Enable auto-reconnect
            startReconnection(WS_URL);
        } catch (Exception e) {
            log.error("Failed to connect to " + WS_URL + ": " + e.getMessage());
        }
    }

    //The auto-reconnect mechanism starts a scheduled thread to check connection status; if disconnected, it reconnects automatically
    private void startReconnection(String s) {
        ScheduledExecutorService usExecutor = Executors.newScheduledThreadPool(1);
        Runnable usTask = () -> {
            if (session == null || !session.isOpen()) {
                log.debug("reconnection...");
                connect(s);
            }
        };
        usExecutor.scheduleAtFixedRate(usTask, 1000, 10000, TimeUnit.MILLISECONDS);
    }

    //Implementation of WebSocket connection establishment
    private void connect(String s) {
        try {
            WebSocketContainer container = ContainerProvider.getWebSocketContainer();
            session = container.connectToServer(WebsocketExample.class, URI.create(s));
        } catch (DeploymentException | IOException e) {
            log.error("Failed to connect to the server: {}", e.getMessage());
        }
    }

    //The following method is executed when the WebSocket connection is successfully established
    @OnOpen
    public void onOpen(Session session) throws IOException, InterruptedException {
        //This method is triggered upon successful WebSocket connection
        System.out.println("Connection opened: " + session.getId());

        // Send a latest trade details subscription request
        JSONObject tradeSendObj = new JSONObject();
        //Refer to the WebSocket API docs for different protocol codes. 10000 is for latest trade subscription
        tradeSendObj.put("code", 10000);
        //Custom trace ID
        tradeSendObj.put("trace", "01213e9d-90a0-426e-a380-ebed633cba7a");
        //Build the subscription request entity in JSON format
        JSONObject data = new JSONObject();
        //Subscribe to BTCUSDT
        data.put("codes", "BTCUSDT");
        tradeSendObj.put("data", data);
        //Send the latest trade subscription request
        session.getBasicRemote().sendText(tradeSendObj.toJSONString());

        //-----------------------------------------------------------------//
        //Add a delay between different requests
        Thread.sleep(5000);

        //Send the real-time order book (depth) data subscription request
        JSONObject depthSendObj = new JSONObject();
        //Refer to the WebSocket API docs for different protocol codes. 10003 is for real-time order book subscription
        depthSendObj.put("code", 10003);
        //Custom trace ID
        depthSendObj.put("trace", "01213e9d-90a0-426e-a380-ebed633cba7a");
        //Build the subscription request entity in JSON format
        depthSendObj.put("data", data);
        //Send subscription request
        session.getBasicRemote().sendText(depthSendObj.toJSONString());

        
        //-----------------------------------------------------------------//
        //Add a delay between different requests
        Thread.sleep(5000);

        //Send the candlestick data subscription request
        JSONObject klineSendObj = new JSONObject();
        //Protocol 10006 is for candlestick subscription
        klineSendObj.put("code", 10006);
        //Custom trace ID
        klineSendObj.put("trace", "01213e9d-90a0-426e-a380-ebed633cba7a");
        //Build the subscription request entity in JSON format        JSONObject klineData = new JSONObject();
        JSONArray klineDataArray = new JSONArray();

        //Build the entity for 1-minute candle subscription
        JSONObject kline1minObj = new JSONObject();
        //“type” corresponds to the klineType parameter
        kline1minObj.put("type", 1);
        kline1minObj.put("codes", "BTCUSDT");
        klineDataArray.add(kline1minObj);
        klineData.put("arr", klineDataArray);

        klineSendObj.put("data", klineData);
        //Send the subscription request
        session.getBasicRemote().sendText(klineSendObj.toJSONString());

        //Schedule a heartbeat task
        ScheduledExecutorService pingExecutor = Executors.newScheduledThreadPool(1);
        Runnable pingTask = WebsocketExample::ping;
        pingExecutor.scheduleAtFixedRate(pingTask, 30, 30, TimeUnit.SECONDS);

    }

    @OnMessage
    public void onMessage(String message, Session session) {
        //Receives messages returned by the INFOWAY server, includes subscription success/failure notifications and actual market data pushes
        try {
            System.out.println("Message received: " + message);
        } catch (Exception e) {
        }
    }

    @OnClose
    public void onClose(Session session, CloseReason reason) {
        //This method is called when the WebSocket connection is closed
        System.out.println("Connection closed: " + session.getId() + ", reason: " + reason);
    }

    @OnError
    public void onError(Throwable error) {
        error.printStackTrace();
    }

    //Continuously send heartbeat messages to prevent the server from closing the connection due to inactivity
    public static void ping() {
        try {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("code", 10010);
            jsonObject.put("trace", "01213e9d-90a0-426e-a380-ebed633cba7a");
            session.getBasicRemote().sendText(jsonObject.toJSONString());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

}

Python