WebSocket Code Example

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);
        }
    }

}

import json
import time
import schedule
import threading
import websocket
from loguru import logger

class WebsocketExample:
    def __init__(self):
        self.session = None
        self.ws_url = "wss://data.infoway.io/ws?business=crypto&apikey=yourApikey"
        self.reconnecting = False
        self.is_ws_connected = False  # Add connection status flag

    def connect_all(self):
        """Establish WebSocket connection and start auto-reconnect mechanism"""
        try:
            self.connect(self.ws_url)
            self.start_reconnection(self.ws_url)
        except Exception as e:
            logger.error(f"Failed to connect to {self.ws_url}: {str(e)}")

    def start_reconnection(self, url):
        """Start scheduled reconnection checks"""
        def check_connection():
            if not self.is_connected():
                logger.debug("Reconnection attempt...")
                self.connect(url)
        
        # Use a thread to periodically check connection status
        schedule.every(10).seconds.do(check_connection)
        def run_scheduler():
            while True:
                schedule.run_pending()
                time.sleep(1)
        threading.Thread(target=run_scheduler, daemon=True).start()

    def is_connected(self):
        """Check WebSocket connection status"""
        return self.session and self.is_ws_connected

    def connect(self, url):
        """Establish a WebSocket connection"""
        try:
            if self.is_connected():
                self.session.close()
            
            self.session = websocket.WebSocketApp(
                url,
                on_open=self.on_open,
                on_message=self.on_message,
                on_error=self.on_error,
                on_close=self.on_close
            )
            
            # Start the WebSocket connection (non-blocking)
            threading.Thread(target=self.session.run_forever, daemon=True).start()
        except Exception as e:
            logger.error(f"Failed to connect to the server: {str(e)}")

    def on_open(self, ws):
        """Callback when WebSocket connection is successfully established"""
        logger.info(f"Connection opened")
        self.is_ws_connected = True  # Set connection status to True
        
        try:
            # Send realtime trade subscription request
            trade_send_obj = {
                "code": 10000,
                "trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
                "data": {"codes": "BTCUSDT"}
            }
            self.send_message(trade_send_obj)
            
            # Wait a while between requests
            time.sleep(5)
            
            # Send realtime order book subscription request
            depth_send_obj = {
                "code": 10003,
                "trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
                "data": {"codes": "BTCUSDT"}
            }
            self.send_message(depth_send_obj)
            
            # Wait a while between requests
            time.sleep(5)
            
            # Send realtime candlestick (K-line) subscription request
            kline_data = {
                "arr": [
                    {
                        "type": 1,
                        "codes": "BTCUSDT"
                    }
                ]
            }
            kline_send_obj = {
                "code": 10006,
                "trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
                "data": kline_data
            }
            self.send_message(kline_send_obj)
            
            # Start the scheduled heartbeat task
            schedule.every(30).seconds.do(self.ping)
            
        except Exception as e:
            logger.error(f"Error sending initial messages: {str(e)}")

    def on_message(self, ws, message):
        """Callback for incoming messages"""
        try:
            logger.info(f"Message received: {message}")
        except Exception as e:
            logger.error(f"Error processing message: {str(e)}")

    def on_close(self, ws, close_status_code, close_msg):
        """Callback when the connection is closed"""
        logger.info(f"Connection closed: {close_status_code} - {close_msg}")
        self.is_ws_connected = False  # Set connection status to False

    def on_error(self, ws, error):
        """Error handling callback"""
        logger.error(f"WebSocket error: {str(error)}")
        self.is_ws_connected = False  # Set connection status to False on error

    def send_message(self, message_obj):
        """Send a message to the WebSocket server"""
        if self.is_connected():
            try:
                self.session.send(json.dumps(message_obj))
            except Exception as e:
                logger.error(f"Error sending message: {str(e)}")
        else:
            logger.warning("Cannot send message: Not connected")

    def ping(self):
        """Send heartbeat packet"""
        ping_obj = {
            "code": 10010,
            "trace": "01213e9d-90a0-426e-a380-ebed633cba7a"
        }
        self.send_message(ping_obj)

# Usage example
if __name__ == "__main__":
    ws_client = WebsocketExample()
    ws_client.connect_all()
    
    # Keep the main thread running
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        logger.info("Exiting...")
        if ws_client.is_connected():
            ws_client.session.close()

PreviousSubscription Address NextWebSocket Subscribe method

Last updated 3 months ago