Realtime Streaming

const WebSocket = require('ws');

class ChainStreamWebSocket {
  constructor(accessToken) {
    this.accessToken = accessToken;
    this.ws = null;
    this.messageId = 0;
    this.reconnectAttempts = 0;
    this.maxReconnectAttempts = 10;
    this.subscriptions = new Set();
    this.pingInterval = null;
  }

  connect() {
    const url = `wss://realtime-dex.chainstream.io/connection/websocket?token=${this.accessToken}`;
    this.ws = new WebSocket(url);

    this.ws.onopen = () => {
      console.log('WebSocket connection established');
      this.reconnectAttempts = 0;
      
      // Send connect message
      this.send({
        connect: {
          token: this.accessToken,
          name: 'nodejs'
        }
      });
    };

    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      this.handleMessage(data);
    };

    this.ws.onclose = (event) => {
      console.log(`Connection closed: ${event.code}`);
      this.stopPing();
      
      if (event.code !== 1000) {
        this.scheduleReconnect();
      }
    };

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error.message);
    };
  }

  handleMessage(data) {
    // Handle connect response
    if (data.connect) {
      console.log('✅ Authentication successful');
      this.startPing(data.connect.ping || 25);
      this.resubscribe();
      return;
    }

    // Handle data push
    if (data.push) {
      console.log(`[${data.push.channel}]`, data.push.pub.data);
      return;
    }

    // Handle error
    if (data.error) {
      console.error('Error:', data.error.message);
      return;
    }
  }

  send(message) {
    message.id = ++this.messageId;
    this.ws.send(JSON.stringify(message));
  }

  subscribe(channel) {
    this.subscriptions.add(channel);
    this.send({ subscribe: { channel } });
  }

  unsubscribe(channel) {
    this.subscriptions.delete(channel);
    this.send({ unsubscribe: { channel } });
  }

  resubscribe() {
    this.subscriptions.forEach(channel => {
      this.send({ subscribe: { channel } });
    });
  }

  startPing(interval) {
    this.pingInterval = setInterval(() => {
      if (this.ws.readyState === WebSocket.OPEN) {
        this.ws.send('{}');
      }
    }, interval * 1000);
  }

  stopPing() {
    if (this.pingInterval) {
      clearInterval(this.pingInterval);
      this.pingInterval = null;
    }
  }

  scheduleReconnect() {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.error('❌ Max reconnect attempts reached');
      return;
    }

    const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
    console.log(`⏳ Reconnecting in ${delay}ms... (attempt ${this.reconnectAttempts + 1})`);
    this.reconnectAttempts++;

    setTimeout(() => this.connect(), delay);
  }

  close() {
    this.stopPing();
    if (this.ws) {
      this.ws.close(1000, 'Normal closure');
    }
  }
}

// Usage example
const client = new ChainStreamWebSocket(process.env.CHAINSTREAM_ACCESS_TOKEN);
client.connect();

// Subscribe after connection is established
setTimeout(() => {
  client.subscribe('dex-candle:sol_6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN_1m');
  client.subscribe('dex-new-token:sol');
}, 1000);

import asyncio
import websockets
import json
import os

class ChainStreamWebSocket:
    def __init__(self, access_token):
        self.access_token = access_token
        self.ws = None
        self.message_id = 0
        self.subscriptions = set()
        self.ping_interval = 25
    
    async def connect(self):
        url = f"wss://realtime-dex.chainstream.io/connection/websocket?token={self.access_token}"
        
        async with websockets.connect(url) as ws:
            self.ws = ws
            print("WebSocket connection established")
            
            # Send connect message
            await self.send({
                "connect": {
                    "token": self.access_token,
                    "name": "python"
                }
            })
            
            # Start heartbeat and message receiving
            await asyncio.gather(
                self.heartbeat(),
                self.receive()
            )
    
    async def send(self, message):
        self.message_id += 1
        message["id"] = self.message_id
        await self.ws.send(json.dumps(message))
    
    async def receive(self):
        async for message in self.ws:
            data = json.loads(message)
            await self.handle_message(data)
    
    async def handle_message(self, data):
        if "connect" in data:
            print("✅ Authentication successful")
            self.ping_interval = data["connect"].get("ping", 25)
            await self.resubscribe()
        elif "push" in data:
            channel = data["push"]["channel"]
            pub_data = data["push"]["pub"]["data"]
            print(f"[{channel}]", pub_data)
        elif "error" in data:
            print(f"Error: {data['error']['message']}")
    
    async def subscribe(self, channel):
        self.subscriptions.add(channel)
        await self.send({"subscribe": {"channel": channel}})
    
    async def resubscribe(self):
        for channel in self.subscriptions:
            await self.send({"subscribe": {"channel": channel}})
    
    async def heartbeat(self):
        while True:
            await asyncio.sleep(self.ping_interval)
            if self.ws and self.ws.open:
                await self.ws.send("{}")

# Usage example
async def main():
    client = ChainStreamWebSocket(os.environ["CHAINSTREAM_ACCESS_TOKEN"])
    
    # Pre-add subscriptions
    client.subscriptions.add("dex-candle:sol_6p6xgHyF7AeE6TZkSmFsko444wqoP15icUSqi2jfGiPN_1m")
    client.subscriptions.add("dex-new-token:sol")
    
    await client.connect()

asyncio.run(main())