Connecting to streams

wscat -c "wss://api.metastreams.io/v1/stream" -H "Authorization: Bearer $API_KEY"

import WebSocket from "ws";

const socket = new WebSocket("wss://api.metastreams.io/v1/stream", {
  headers: { Authorization: `Bearer ${process.env.API_KEY}` },
});
// A refused handshake arrives as an `error` event. Without a listener, Node exits.
socket.on("error", (error) => console.error("stream error:", error.message));

import asyncio
import os

from websockets.asyncio.client import connect

async def main():
    async with connect(
        "wss://api.metastreams.io/v1/stream",
        additional_headers={"Authorization": f"Bearer {os.environ['API_KEY']}"},
    ) as socket:
        ...

asyncio.run(main())

import WebSocket from "ws";

const URL = "wss://api.metastreams.io/v1/stream";
const subscriptions = [
  { action: "subscribe", channel: "spot.trades", filters: { chain: "solana", token: "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263" } },
];

function connect(attempt = 0) {
  const socket = new WebSocket(URL, { headers: { Authorization: `Bearer ${process.env.API_KEY}` } });
  let lastPong = Date.now();
  let heartbeat: NodeJS.Timeout | undefined;
  let unauthorized = false;

  // A refused handshake arrives here too. Without an `error` listener, Node exits.
  socket.on("error", (error) => {
    unauthorized = error.message.includes("Unexpected server response: 401");
    console.error("stream error:", error.message);
  });

  socket.on("open", () => {
    attempt = 0;
    for (const frame of subscriptions) socket.send(JSON.stringify(frame));
    heartbeat = setInterval(() => {
      if (Date.now() - lastPong > 45_000) return socket.terminate();
      socket.send(JSON.stringify({ action: "ping" }));
    }, 30_000);
  });

  socket.on("message", (raw) => {
    const frame = JSON.parse(raw.toString());
    if (frame.type === "pong") lastPong = Date.now();
    if (frame.type === "update") handleUpdate(frame);
    if (frame.type === "error") console.error(frame.code, frame.message);
  });

  socket.on("close", (code) => {
    clearInterval(heartbeat);
    if (unauthorized) return console.error("The API key was refused. Not reconnecting.");
    const delay = code === 1001 ? 0 : Math.min(30_000, 1000 * 2 ** attempt) + Math.random() * 1000;
    setTimeout(() => connect(attempt + 1), delay);
  });
}

function handleUpdate(frame: { channel: string; data: unknown }) {
  console.log(frame.channel, frame.data);
}

connect();

import asyncio
import json
import os
import random

from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed, InvalidStatus

URL = "wss://api.metastreams.io/v1/stream"
HEADERS = {"Authorization": f"Bearer {os.environ['API_KEY']}"}
SUBSCRIPTIONS = [
    {"action": "subscribe", "channel": "spot.trades", "filters": {"chain": "solana", "token": "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263"}},
]

async def run():
    attempt = 0
    while True:
        close_code = None
        try:
            # websockets pings the server every 20 seconds by default.
            async with connect(URL, additional_headers=HEADERS) as socket:
                attempt = 0
                for frame in SUBSCRIPTIONS:
                    await socket.send(json.dumps(frame))
                async for raw in socket:
                    frame = json.loads(raw)
                    if frame["type"] == "update":
                        print(frame["channel"], frame["data"])
                    elif frame["type"] == "error":
                        print("error", frame["code"], frame["message"])
                close_code = socket.close_code
        except InvalidStatus as error:
            if error.response.status_code == 401:
                raise
        except ConnectionClosed as error:
            close_code = error.rcvd.code if error.rcvd else None
        except OSError:
            pass
        if close_code == 1001:
            continue
        await asyncio.sleep(min(30, 2 ** attempt) + random.random())
        attempt += 1

asyncio.run(run())