oscope/src/feed.mjs at main · oscope/oscope

oscope/src/feed.mjs

OScope OScope: open source memecoin terminal for Solana f8ff931·

Oct 20, 2026

Code

65 lines (61 loc) · 3.06 KB

	// The live trade stream (one websocket, reopened when it drops) and the SOL price.

	import WebSocket from 'ws';

	export function connectFeed(url, onFrame, log = console.log) {
	let ws = null, wait = 1000, alive = 0, timer = null, closed = false;
	const state = { url, connected: false, since: 0, reconnects: 0 };

	const open = () => {
	if (closed) return;
	ws = new WebSocket(url, { perMessageDeflate: true, handshakeTimeout: 15000 });
	ws.on('open', () => {
	state.connected = true;
	state.since = Date.now();
	if (!state.reconnects) log(`feed: connected to ${url}`);
	});
	ws.on('message', buf => {
	alive = Date.now();
	const raw = buf.toString();
	let f;
	try { f = JSON.parse(raw); } catch { return; }
	// onFrame gets each transaction and its exact text (the firehose passes the text on untouched)
	if (Array.isArray(f)) for (const x of f) onFrame(x, null);
	else onFrame(f, raw);
	});
	ws.on('close', (code, reason) => {
	const lived = state.connected ? Date.now() - state.since : 0;
	// a connection that lasted a while starts the wait over; one that drops at once backs off
	if (lived > 15000) wait = 1000;
	const why = `${code}${reason && reason.length ? ' ' + reason.toString().slice(0, 120) : ''}`;
	if (lived > 15000 \|\| state.reconnects < 3 \|\| state.reconnects % 20 === 0) {
	log(`feed: closed (${why}) after ${Math.round(lived / 1000)} s, retrying in ${Math.round(wait / 1000)} s${lived < 2000 && state.reconnects > 1 ? '. The stream may allow only a few connections per IP address: close other apps reading it.' : ''}`);
	}
	state.connected = false;
	state.reconnects++;
	setTimeout(open, wait);
	wait = Math.min(wait * 2, 30000);
	});
	ws.on('error', err => { log(`feed: ${err.message}`); try { ws.terminate(); } catch {} });
	};
	open();
	// a stream that goes quiet for 30 s is reopened
	timer = setInterval(() => { if (state.connected && Date.now() - alive > 30000) { log('feed: quiet for 30 s, reopening'); try { ws.terminate(); } catch {} } }, 10000);
	state.close = () => { closed = true; clearInterval(timer); try { ws.close(); } catch {} };
	return state;
	}

	// SOL in dollars, from public exchange prices.
	export function solPrice(log = console.log) {
	const st = { usd: 0, at: 0 };
	const sources = [
	async () => +(await (await fetch('https://api.coinbase.com/v2/prices/SOL-USD/spot', { signal: AbortSignal.timeout(6000) })).json()).data.amount,
	async () => +(await (await fetch('https://api.kraken.com/0/public/Ticker?pair=SOLUSD', { signal: AbortSignal.timeout(6000) })).json()).result.SOLUSD.c[0],
	async () => +(await (await fetch('https://api.binance.com/api/v3/ticker/price?symbol=SOLUSDT', { signal: AbortSignal.timeout(6000) })).json()).price,
	];
	const tick = async () => {
	for (const s of sources) {
	try { const v = await s(); if (v > 1) { st.usd = v; st.at = Date.now(); return; } } catch {}
	}
	if (!st.usd) log('sol price: no source answered, retrying');
	};
	tick();
	setInterval(tick, 20000).unref();
	return st;
	}