WebSocket Streaming

const WebSocket = require("ws");
const apiKey = process.env.THERUNDOWN_API_KEY;
if (!apiKey) throw Error('Set THERUNDOWN_API_KEY');

const ws = new WebSocket(
  "wss://therundown.io/api/v2/ws/markets?sport_ids=4&market_ids=1,2,3",
  { headers: { "X-TheRundown-Key": apiKey } }
);

ws.on("open", () => {
  console.log("Connected to V2 Markets WebSocket");
});

ws.on("message", (raw) => {
  const msg = JSON.parse(raw.toString());

  // Handle heartbeat
  if (msg.meta?.type === "heartbeat") {
    console.log("Heartbeat received");
    return;
  }

  // Process market price update
  const d = msg.data;
  console.log(
    `Update: event=${d.event_id} market=${d.market_id} aff=${d.affiliate_id}`
  );
  console.log(
    `  line=${d.line} price=${d.price} (was ${d.previous_price}, delta=${d.price_delta})`
  );
});

ws.on("error", (error) => {
  console.error("WebSocket error:", error);
});

ws.on("close", (code, reason) => {
  console.log(`WebSocket closed: code=${code} reason=${reason}`);
});

import asyncio
import json
import os
import websockets

API_KEY = os.environ["THERUNDOWN_API_KEY"]
WS_URL = (
    f"wss://therundown.io/api/v2/ws/markets"
    f"?sport_ids=4&market_ids=1,2,3"
)


async def listen():
    async with websockets.connect(
        WS_URL, additional_headers={"X-TheRundown-Key": API_KEY}
    ) as ws:
        print("Connected to V2 Markets WebSocket")

        async for message in ws:
            msg = json.loads(message)

            # Handle heartbeat
            if msg.get("meta", {}).get("type") == "heartbeat":
                print("Heartbeat received")
                continue

            # Process market price update
            d = msg["data"]
            print(
                f"Update: event={d['event_id']} market={d['market_id']} aff={d['affiliate_id']}"
            )
            print(
                f"  line={d['line']} price={d['price']}"
                f" (was {d['previous_price']}, delta={d['price_delta']})"
            )


asyncio.run(listen())

let lastHeartbeat = Date.now();

ws.on("message", (raw) => {
  const msg = JSON.parse(raw.toString());

  if (msg.meta?.type === "heartbeat") {
    lastHeartbeat = Date.now();
    return; // Do not process as data
  }

  // Process actual price update
  handleUpdate(msg.data);
});

// Monitor heartbeat health
setInterval(() => {
  const elapsed = Date.now() - lastHeartbeat;
  if (elapsed > 60_000) {
    console.warn("No heartbeat in 60s, reconnecting...");
    ws.close();
    // Trigger reconnection logic
  }
}, 10_000);

import time

last_heartbeat = time.time()


async def listen_with_heartbeat():
    global last_heartbeat

    async with websockets.connect(
        WS_URL, additional_headers={"X-TheRundown-Key": API_KEY}
    ) as ws:
        async for message in ws:
            msg = json.loads(message)

            if msg.get("meta", {}).get("type") == "heartbeat":
                last_heartbeat = time.time()
                continue

            # Check heartbeat staleness
            if time.time() - last_heartbeat > 60:
                print("No heartbeat in 60s, reconnecting...")
                break

            handle_update(msg["data"])

const WebSocket = require("ws");

function createReconnectingWebSocket(url, onMessage) {
  const apiKey = process.env.THERUNDOWN_API_KEY;
  if (!apiKey) throw Error('Set THERUNDOWN_API_KEY');

  let ws;
  let reconnectDelay = 1000;
  const MAX_DELAY = 30000;

  function connect() {
    ws = new WebSocket(url, {
      headers: { "X-TheRundown-Key": apiKey },
    });

    ws.on("open", () => {
      console.log("WebSocket connected");
      reconnectDelay = 1000; // Reset backoff
    });

    ws.on("message", (raw) => {
      const msg = JSON.parse(raw.toString());
      if (msg.meta?.type === "heartbeat") return;
      onMessage(msg.data);
    });

    ws.on("close", () => {
      const jitter = Math.random() * 1000;
      const delay = Math.min(reconnectDelay + jitter, MAX_DELAY);
      console.log(`WebSocket closed. Reconnecting in ${Math.round(delay)}ms...`);

      setTimeout(() => {
        reconnectDelay = Math.min(reconnectDelay * 2, MAX_DELAY);
        connect();
      }, delay);
    });

    ws.on("error", (error) => {
      console.error("WebSocket error:", error);
      ws.close(); // Triggers onclose -> reconnect
    });
  }

  connect();

  return {
    close: () => {
      reconnectDelay = MAX_DELAY + 1; // Prevent reconnection
      ws.close();
    },
  };
}

// Usage
const connection = createReconnectingWebSocket(
  "wss://therundown.io/api/v2/ws/markets?sport_ids=4",
  (update) => {
    console.log(`Price update: event=${update.event_id} market=${update.market_id} price=${update.price}`);
  }
);

import asyncio
import json
import os
import random
import websockets


async def connect_with_reconnect(url, on_message):
    reconnect_delay = 1.0
    max_delay = 30.0

    while True:
        try:
            async with websockets.connect(
                url, additional_headers={"X-TheRundown-Key": API_KEY}
            ) as ws:
                print("WebSocket connected")
                reconnect_delay = 1.0  # Reset backoff

                async for message in ws:
                    msg = json.loads(message)
                    if msg.get("meta", {}).get("type") == "heartbeat":
                        continue
                    await on_message(msg["data"])

        except (websockets.ConnectionClosed, ConnectionError) as e:
            jitter = random.uniform(0, 1)
            delay = min(reconnect_delay + jitter, max_delay)
            print(f"WebSocket closed ({e}). Reconnecting in {delay:.1f}s...")
            await asyncio.sleep(delay)
            reconnect_delay = min(reconnect_delay * 2, max_delay)

        except Exception as e:
            print(f"Unexpected error: {e}")
            await asyncio.sleep(5)


async def handle_update(update):
    print(f"Price update: event={update['event_id']} market={update['market_id']} price={update['price']}")


# Run
API_KEY = os.environ["THERUNDOWN_API_KEY"]
WS_URL = "wss://therundown.io/api/v2/ws/markets?sport_ids=4"
asyncio.run(connect_with_reconnect(WS_URL, handle_update))