Code examples

// npm install [email protected]
const WebSocket = require("ws");

const url = "wss://ws.newlistings.pro/v2/full";
const key = process.env.NLF_KEY;
if (!key) throw new Error("Set NLF_KEY before starting the client.");
let retryMs = 1000;

function readJson(raw) {
  try { return JSON.parse(raw.toString()); } catch { return null; }
}

function retryAfterMs(value = "") {
  if (/^\d+$/.test(value)) return Number(value) * 1000;
  return Math.max(0, Date.parse(value) - Date.now()) || 0;
}

function receiveSession() {
  return new Promise((resolve) => {
    let stop = false;
    let minimumWaitMs = 0;
    let rejectedUpgrade = false;
    const ws = new WebSocket(url, {
      headers: { authorization: `Bearer ${key}` },
      handshakeTimeout: 10000,
    });

    ws.on("unexpected-response", (_request, response) => {
      rejectedUpgrade = true;
      const status = response.statusCode;
      minimumWaitMs = retryAfterMs(response.headers["retry-after"]);
      stop = status < 500 && status !== 408 && status !== 429;
      let body = "";
      response.setEncoding("utf8");
      response.on("data", (chunk) => { body += chunk.slice(0, 65536 - body.length); });
      response.on("end", () => {
        const message = readJson(body);
        console.error("upgrade rejected", status, message?.code);
        if (["AUTHENTICATION_FAILED", "KEY_EXPIRED", "INVALID_REQUEST"]
            .includes(message?.code)) stop = true;
        ws.terminate();
      });
      response.on("error", () => ws.terminate());
      response.once("close", () => ws.terminate());
      response.setTimeout(10000, () => response.destroy());
    });

    ws.on("message", (data) => {
      const message = readJson(data);
      if (!message || typeof message !== "object") {
        console.error("Invalid JSON message; inspect the response before retrying.");
        stop = true;
        ws.close();
      } else if (message.type === "success" && message.code === "READY") {
        retryMs = 1000;
        console.log("ready", message.subscription);
      } else if (message.type === "error") {
        console.error(message.code, message.message);
        stop = message.code !== "SERVER_UNAVAILABLE";
        ws.close();
      } else if (["announcement", "tweet"].includes(message.type)) {
        console.log(message);
      }
    });

    ws.on("error", (error) => {
      if (!rejectedUpgrade) console.error(error.message);
    });
    // Only close completes a session. Errors never schedule a second retry.
    ws.once("close", (code) => resolve({ stop: stop || code === 1008, minimumWaitMs }));
  });
}

async function connect() {
  while (true) {
    const result = await receiveSession();
    if (result.stop) {
      console.error("Stopped. Check the request, key, and connection limits.");
      return;
    }
    const waitMs = Math.max(retryMs, result.minimumWaitMs) + Math.random() * 250;
    console.log("retry in", Math.ceil(waitMs), "ms");
    await new Promise((resolve) => setTimeout(resolve, waitMs));
    retryMs = Math.min(retryMs * 2, 30000);
  }
}

connect().catch(console.error);

# pip install websocket-client==1.9.2
import json
import os
import random
import time
from email.utils import parsedate_to_datetime

import websocket

URL = "wss://ws.newlistings.pro/v2/full"
HEADERS = [f"authorization: Bearer {os.environ['NLF_KEY']}"]

def read_json(raw):
    try:
        value = json.loads(raw)
        return value if isinstance(value, dict) else {}
    except (ValueError, TypeError):
        return {}

def retry_after_seconds(value=""):
    if value.isdigit():
        return int(value)
    try:
        return max(0, parsedate_to_datetime(value).timestamp() - time.time())
    except (ValueError, TypeError, OverflowError):
        return 0

def connect():
    retry_seconds = 1
    while True:
        ws, stop, minimum_wait = None, False, 0
        try:
            ws = websocket.create_connection(URL, header=HEADERS, timeout=10)
            ws.settimeout(None)
            while True:
                opcode, raw = ws.recv_data(control_frame=True)
                if opcode == websocket.ABNF.OPCODE_CLOSE:
                    stop = len(raw) >= 2 and int.from_bytes(raw[:2], "big") == 1008
                    break
                if opcode != websocket.ABNF.OPCODE_TEXT:
                    continue
                message = read_json(raw)
                if not message:
                    print("Invalid JSON message; inspect the response before retrying.")
                    stop = True
                    break
                if message.get("type") == "success" and message.get("code") == "READY":
                    retry_seconds = 1
                    print("ready", message.get("subscription"))
                elif message.get("type") == "error":
                    print(message.get("code"), message.get("message"))
                    stop = message.get("code") != "SERVER_UNAVAILABLE"
                    break
                elif message.get("type") in ("announcement", "tweet"):
                    print(message)
        except websocket.WebSocketBadStatusException as error:
            headers = {k.lower(): v for k, v in (error.resp_headers or {}).items()}
            minimum_wait = retry_after_seconds(headers.get("retry-after", ""))
            code = read_json(error.resp_body).get("code")
            print("upgrade rejected", error.status_code, code)
            stop = error.status_code < 500 and error.status_code not in (408, 429)
            stop = stop or code in ("AUTHENTICATION_FAILED", "KEY_EXPIRED", "INVALID_REQUEST")
        except (OSError, websocket.WebSocketException) as error:
            print(type(error).__name__)
        finally:
            if ws is not None:
                ws.close()
        if stop:
            print("Stopped. Check the request, key, and connection limits.")
            return
        wait = max(retry_seconds, minimum_wait) + random.uniform(0, 0.25)
        print("retry in", round(wait, 2), "seconds")
        time.sleep(wait)
        retry_seconds = min(retry_seconds * 2, 30)

if __name__ == "__main__":
    try:
        connect()
    except KeyboardInterrupt:
        pass