diff --git a/README.md b/README.md index 0ddf820..3d937f2 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,7 @@ export default function App() { | `useNewsCount(params)` | `/1/count` | Aggregate counts (requires `from_date`, `to_date`) | | `useCryptoCount(params)` | `/1/crypto/count` | Aggregate crypto counts | | `useMarketCount(params)` | `/1/market/count` | Aggregate market counts | +| `useNewsStream(registrationId)` | `wss://ws.newsdata.io/ws/event` | Real-time stream ([below](#real-time-news-websocket)) | Every hook has the same shape: @@ -111,6 +112,85 @@ useLatestNews({ country: ['us', 'gb'], language: 'en', size: 20 }); Inline param objects are safe — the hook compares by **value**, not reference, so re-renders only re-fetch when the values change. +## Real-time news (WebSocket) + +`useNewsStream` opens a real-time connection for a **registered query** and +accumulates the matching articles. Register the query first — the returned +`registration_id` identifies it from then on: + +```jsx +import { useEffect, useState } from 'react'; +import { useNewsDataClient, useNewsStream } from 'newsdataapi'; + +function BitcoinTicker() { + const client = useNewsDataClient(); + const [registrationId, setRegistrationId] = useState(null); + + useEffect(() => { + client.websocketRegister({ q: 'bitcoin', language: 'en' }) + .then(({ results }) => setRegistrationId(results.registration_id)); + }, [client]); + + // A falsy id defers connecting until registration resolves. + const { articles, error, isConnected } = useNewsStream(registrationId); + + if (error) return

Stream error: {error.message}

; + + return ( + <> +

{isConnected ? 'live' : 'connecting…'}

+ + + ); +} +``` + +The hook returns `{ articles, latest, error, isConnected }`. Articles +accumulate **newest first** and are capped at `maxArticles` (default 100) so a +long-lived stream can't grow without bound. The connection opens on mount, +closes on unmount, and reconnects through a capped exponential backoff. + +```jsx +const { articles, latest } = useNewsStream(registrationId, { + enabled: true, // defer connecting when false + maxArticles: 100, // cap on retained articles + reconnect: true, // auto-reconnect on transient drops +}); +``` + +`error` is set for a permanent rejection — bad API key or unknown `registration_id`, +exhausted API credits, or too many simultaneous devices — which surfaces as `NewsdataWebSocketAuthError` and is **not** retried. + +The server always accepts the handshake and then closes with code **1008** when +the connection is refused, carrying one of three reasons: `invalid credentials +or registration not found`, `api limit reached`, or `device limit reached` (more +than 5 devices on one `registration_id`). Every other close code — including +`1013` (`send timeout`, meaning the client read too slowly) — is transient and +reconnects. + +**Each delivered article consumes 1 API credit per connected device.** + +Transient drops reconnect silently and leave `error` null. + +Managing registered queries goes through the client: +`websocketRegister(params)`, `websocketFetch()`, and +`websocketDelete(registrationId)`. Registering an identical query twice +rejects with a `NewsdataApiError` whose `statusCode` is 409; the existing id is +at `err.responseBody.results.registration_id`. + +Outside React, use the same `NewsDataApiWebSocket` class the hook wraps: + +```js +import { NewsDataApiWebSocket } from 'newsdataapi'; + +const ws = new NewsDataApiWebSocket(client); +for await (const response of ws.stream(registrationId)) { + console.log(response.results); +} +``` + ## Provider Two ways to wire it up: @@ -163,7 +243,9 @@ NewsdataError (catch-all base) │ ├── NewsdataAuthError (401 / 403) │ ├── NewsdataRateLimitError (429; .retryAfter) │ └── NewsdataServerError (5xx) -└── NewsdataNetworkError (.cause) +├── NewsdataNetworkError (.cause) +└── NewsdataWebSocketError (real-time stream) + └── NewsdataWebSocketAuthError (policy-violation close 1008) ``` Validation errors are thrown **before** the request is sent (no API quota diff --git a/src/core/client.js b/src/core/client.js index a89d908..ddeaa05 100644 --- a/src/core/client.js +++ b/src/core/client.js @@ -3,6 +3,9 @@ import { BASE_URL, ENDPOINTS, + ENDPOINT_METHODS, + RESULTS_OPTIONAL, + WS_NEWS_TYPE, DEFAULT_REQUEST_TIMEOUT, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_BACKOFF, @@ -124,6 +127,67 @@ export class NewsDataApiClient { return this.#request('sources', validated); } + // ---- real-time query management --------------------------------------- + + /** + * Register a real-time WebSocket query. POST /1/websocket/register + * + * Takes the familiar filter names (`q`, `country`, `language`, `domain`, …); + * no date or paging filters apply, since a registered query matches news as + * it is published. The new query's id is at `results.registration_id` — pass + * it to `NewsDataApiWebSocket#stream`. + * + * Registering an identical query twice rejects with a `NewsdataApiError` + * whose `statusCode` is 409; the existing id is at + * `err.responseBody.results.registration_id`. + * @returns {Promise} + */ + websocketRegister(params = {}) { + const { rawQuery = null, ...rest } = params; + const validated = validateParams('websocket_register', rest, rawQuery); + validated.news_type = WS_NEWS_TYPE; + return this.#request('websocket_register', validated); + } + + /** + * List the account's registered real-time queries. GET /1/websocket/fetch + * One entry per query at `results.queries`. + * @returns {Promise} + */ + websocketFetch() { + return this.#request('websocket_fetch', {}); + } + + /** + * Delete a registered real-time query. DELETE /1/websocket/delete + * @param {string} registrationId + * @returns {Promise} + */ + websocketDelete(registrationId) { + if (typeof registrationId !== 'string' || registrationId === '') { + throw new NewsdataValidationError( + 'registrationId must be a non-empty string', + 'registration_id', + ); + } + return this.#request('websocket_delete', { registration_id: registrationId }); + } + + /** The API key, for the WebSocket handshake URL. @internal */ + get apiKeyForWebSocket() { + return this.#apiKey; + } + + /** The configured fetch, reused by the WebSocket handshake probe. @internal */ + get fetchForWebSocket() { + return this.#fetch; + } + + /** Forward a log line from the WebSocket layer. @internal */ + logForWebSocket(level, message) { + this.#log(level, message); + } + // ---- dispatch --------------------------------------------------------- /** @@ -179,16 +243,17 @@ export class NewsDataApiClient { const search = new URLSearchParams({ ...params, apikey: this.#apiKey }); const fullUrl = `${this.#endpointUrl(endpoint)}?${search.toString()}`; const logUrl = redactApiKey(fullUrl); + const method = ENDPOINT_METHODS[endpoint] ?? 'GET'; for (let attempt = 1; attempt <= this.#maxRetries; attempt += 1) { - this.#log('info', `GET ${logUrl} (attempt ${attempt}/${this.#maxRetries})`); + this.#log('info', `${method} ${logUrl} (attempt ${attempt}/${this.#maxRetries})`); let res; const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), this.#timeout); try { res = await this.#fetch(fullUrl, { - method: 'GET', + method, headers: { Accept: 'application/json' }, signal: controller.signal, }); @@ -220,7 +285,7 @@ export class NewsDataApiClient { throw new NewsdataApiError(`Non-JSON response from API (status ${status})`, status); } - if (status === 200 && this.#isSuccess(body)) { + if (status === 200 && this.#isSuccess(body, endpoint)) { if (this.#includeHeaders) { body.responseHeaders = Object.fromEntries(res.headers.entries()); } @@ -325,14 +390,11 @@ export class NewsDataApiClient { } } - #isSuccess(body) { - return ( - body - && typeof body === 'object' - && body.status === 'success' - && body.results !== null - && body.results !== undefined - ); + #isSuccess(body, endpoint) { + if (!body || typeof body !== 'object' || body.status !== 'success') return false; + // The websocket management endpoints may answer without a `results` field. + if (RESULTS_OPTIONAL.includes(endpoint)) return true; + return body.results !== null && body.results !== undefined; } #errorMessage(body, status) { diff --git a/src/core/constants.js b/src/core/constants.js index b37de6b..e1b7de0 100644 --- a/src/core/constants.js +++ b/src/core/constants.js @@ -28,8 +28,34 @@ export const ENDPOINTS = Object.freeze({ count: 'count', crypto_count: 'crypto/count', market_count: 'market/count', + websocket_register: 'websocket/register', + websocket_fetch: 'websocket/fetch', + websocket_delete: 'websocket/delete', }); +// HTTP method per endpoint; anything absent is a GET. +export const ENDPOINT_METHODS = Object.freeze({ + websocket_register: 'POST', + websocket_delete: 'DELETE', +}); + +// The websocket management endpoints answer with a success envelope that may +// carry no `results` field, so they are exempt from the results-present check +// applied to the news endpoints. +export const RESULTS_OPTIONAL = Object.freeze([ + 'websocket_register', 'websocket_fetch', 'websocket_delete', +]); + +// Real-time WebSocket defaults (NewsDataApiWebSocket). +export const WS_BASE_URL = 'wss://ws.newsdata.io/ws/event'; +// The feed a registered query matches against. +export const WS_NEWS_TYPE = 'latest'; +// Close code the server uses for a permanent rejection. +export const WS_POLICY_VIOLATION = 1008; +export const WS_RECONNECT_DELAY = 1_000; // ms before the first reconnect; doubles each retry +export const WS_RECONNECT_DELAY_MAX = 30_000; // cap on the reconnect delay +export const WS_OPEN_TIMEOUT = 10_000; // ms to wait for the opening handshake + // Endpoints that require both from_date and to_date. export const REQUIRES_DATE_RANGE = Object.freeze(['count', 'crypto_count', 'market_count']); @@ -102,6 +128,19 @@ export const FILTERS = Object.freeze({ 'market_id', 'prioritydomain', 'page', 'sentiment', 'removeduplicate', 'size', 'sort', 'tag', 'interval', 'creator', 'datatype', 'sentiment_score', ], + // Real-time query registration. No date/paging filters — a registered query + // matches news as it is published. `news_type` is set by websocketRegister, + // not by the caller. + websocket_register: [ + 'q', 'qintitle', 'qinmeta', 'country', 'excludecountry', 'category', + 'excludecategory', 'language', 'excludelanguage', 'domain', 'domainurl', + 'excludedomain', 'prioritydomain', 'timezone', 'full_content', 'image', + 'video', 'removeduplicate', 'tag', 'sentiment', 'sentiment_score', + 'region', 'organization', 'creator', 'datatype', 'excludefield', + 'news_type', + ], + websocket_fetch: [], + websocket_delete: ['registration_id'], }); // Control/meta keys accepted on endpoint methods but not sent as API params. diff --git a/src/core/errors.js b/src/core/errors.js index fa109d8..5268d3e 100644 --- a/src/core/errors.js +++ b/src/core/errors.js @@ -72,6 +72,31 @@ export class NewsdataServerError extends NewsdataApiError { } } +/** A real-time WebSocket stream failure (NewsDataApiWebSocket). */ +export class NewsdataWebSocketError extends NewsdataError { + /** + * @param {string} message + * @param {Error|null} [cause] The underlying error. + */ + constructor(message, cause = null) { + super(message); + this.name = 'NewsdataWebSocketError'; + if (cause) this.cause = cause; + } +} + +/** + * The server rejected the WebSocket connection — bad API key, missing + * WebSocket entitlement, unknown registration_id, device limit reached, or + * exhausted quota. Never retried, regardless of the `reconnect` setting. + */ +export class NewsdataWebSocketAuthError extends NewsdataWebSocketError { + constructor(message, cause = null) { + super(message, cause); + this.name = 'NewsdataWebSocketAuthError'; + } +} + /** A network-level failure (DNS, TLS, timeout, abort) prevented the request. */ export class NewsdataNetworkError extends NewsdataError { /** diff --git a/src/core/websocket.js b/src/core/websocket.js new file mode 100644 index 0000000..53c649d --- /dev/null +++ b/src/core/websocket.js @@ -0,0 +1,322 @@ +// Real-time WebSocket support for NewsData.io. +// +// Uses the browser's global `WebSocket`. No dependency is required; pass +// `options.WebSocket` to supply your own implementation when running somewhere +// it is missing (Node < 22, jsdom without a WebSocket, tests). + +import { + WS_BASE_URL, + WS_POLICY_VIOLATION, + WS_RECONNECT_DELAY, + WS_RECONNECT_DELAY_MAX, + WS_OPEN_TIMEOUT, +} from './constants.js'; +import { + NewsdataError, + NewsdataValidationError, + NewsdataWebSocketError, + NewsdataWebSocketAuthError, +} from './errors.js'; +import { redactApiKey } from './client.js'; + +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * NewsData.io real-time WebSocket service. + * + * Registers, lists, and deletes the account's real-time queries and streams + * the responses for a registered query: + * + * ```js + * const client = new NewsDataApiClient(apiKey); + * const ws = new NewsDataApiWebSocket(client); + * + * const { results } = await ws.websocketRegister({ q: 'bitcoin' }); + * + * for await (const response of ws.stream(results.registration_id)) { + * for (const article of response.results) console.log(article.title); + * } + * ``` + * + * Transient drops are reconnected automatically with a capped exponential + * backoff; pass `reconnect: false` to stop on the first disconnect. A + * permanent rejection throws `NewsdataWebSocketAuthError` and is never + * retried. + * + * Break out of the loop (or call `close()`) to stop; the connection is closed + * either way. + */ +export class NewsDataApiWebSocket { + #client; + + #baseUrl; + + #reconnect; + + #reconnectDelay; + + #reconnectDelayMax; + + #openTimeout; + + #WebSocketImpl; + + #socket = null; + + #closed = false; + + /** + * @param {import('./client.js').NewsDataApiClient} client + * Supplies the API key and performs the management HTTP calls. Not closed + * by this class. + * @param {object} [options] + * @param {string} [options.baseUrl] WebSocket endpoint. + * @param {boolean} [options.reconnect] Auto-reconnect; default true. + * @param {number} [options.reconnectDelay] ms before the first reconnect. + * @param {number} [options.reconnectDelayMax] Cap on the reconnect delay. + * @param {number} [options.openTimeout] ms to wait for the handshake. + * @param {Function} [options.WebSocket] WebSocket implementation. + */ + constructor(client, options = {}) { + if (!client) { + throw new NewsdataValidationError('client is required', 'client'); + } + this.#client = client; + this.#baseUrl = options.baseUrl ?? WS_BASE_URL; + this.#reconnect = options.reconnect ?? true; + this.#reconnectDelay = options.reconnectDelay ?? WS_RECONNECT_DELAY; + this.#reconnectDelayMax = options.reconnectDelayMax ?? WS_RECONNECT_DELAY_MAX; + this.#openTimeout = options.openTimeout ?? WS_OPEN_TIMEOUT; + this.#WebSocketImpl = options.WebSocket ?? globalThis.WebSocket; + + if (typeof this.#WebSocketImpl !== 'function') { + throw new NewsdataError( + 'No WebSocket implementation available; browsers provide one globally, ' + + 'or pass options.WebSocket.', + ); + } + } + + // ---- query management ------------------------------------------------- + + /** Register a real-time query. See NewsDataApiClient#websocketRegister. */ + websocketRegister(params = {}) { + return this.#client.websocketRegister(params); + } + + /** List registered queries. See NewsDataApiClient#websocketFetch. */ + websocketFetch() { + return this.#client.websocketFetch(); + } + + /** Delete a registered query. See NewsDataApiClient#websocketDelete. */ + websocketDelete(registrationId) { + return this.#client.websocketDelete(registrationId); + } + + // ---- streaming -------------------------------------------------------- + + #url(registrationId) { + const search = new URLSearchParams({ + apikey: this.#client.apiKeyForWebSocket, + registration_id: registrationId, + }); + return `${this.#baseUrl}?${search.toString()}`; + } + + #nextDelay(delay) { + return Math.min(delay * 2, this.#reconnectDelayMax); + } + + /** + * Connect and yield each response for `registrationId` as it arrives. + * Responses have the familiar status / totalResults / results shape. + * + * @param {string} registrationId + * @returns {AsyncGenerator} + */ + async* stream(registrationId) { + if (typeof registrationId !== 'string' || registrationId === '') { + throw new NewsdataValidationError( + 'registrationId must be a non-empty string', + 'registration_id', + ); + } + const url = this.#url(registrationId); + const logUrl = redactApiKey(url); + let delay = this.#reconnectDelay; + this.#closed = false; + + try { + while (!this.#closed) { + const session = this.#connect(url, logUrl); + + try { + for await (const message of session) { + delay = this.#reconnectDelay; // reset after a successful connect + let response; + try { + response = JSON.parse(message); + } catch { + continue; // skip malformed frames + } + if (response && typeof response === 'object' && !Array.isArray(response)) { + yield response; + } + } + } catch (err) { + if (this.#closed) return; + const permanent = this.#permanentAuthError(err); + if (permanent) throw permanent; + if (!this.#reconnect) throw toTransientError(err); + this.#client.logForWebSocket( + 'warn', + `connection to ${logUrl} failed (${err.message}); reconnecting in ${delay}ms`, + ); + } + + if (this.#closed) return; + // A clean close with reconnect disabled ends the stream. + if (!this.#reconnect) return; + await sleep(delay); + delay = this.#nextDelay(delay); + } + } finally { + this.close(); + } + } + + /** + * Bridge one WebSocket connection's events into an async iterable of raw + * message payloads. Throws a `WsClosed` when the socket drops. + */ + #connect(url, logUrl) { + const socket = new this.#WebSocketImpl(url); + this.#socket = socket; + + /** @type {string[]} */ + const queue = []; + /** @type {{resolve: Function, reject: Function}[]} */ + const waiters = []; + let failure = null; + let done = false; + + const settle = () => { + while (waiters.length) { + const waiter = waiters.shift(); + if (queue.length) waiter.resolve({ value: queue.shift(), done: false }); + else if (failure) waiter.reject(failure); + else if (done) waiter.resolve({ value: undefined, done: true }); + else { + waiters.unshift(waiter); + return; + } + } + }; + + let openTimer = null; + if (this.#openTimeout > 0) { + openTimer = setTimeout(() => { + failure = new WsClosed('handshake timed out', null, false); + try { socket.close(); } catch { /* already closing */ } + settle(); + }, this.#openTimeout); + } + + socket.addEventListener('open', () => { + if (openTimer) clearTimeout(openTimer); + this.#client.logForWebSocket('info', `connected to ${logUrl}`); + }); + + socket.addEventListener('message', (event) => { + const { data } = event; + queue.push(typeof data === 'string' ? data : String(data)); + settle(); + }); + + socket.addEventListener('error', () => { + // The close event that follows carries the code; record nothing here so + // the close handler classifies the failure. + }); + + socket.addEventListener('close', (event) => { + if (openTimer) clearTimeout(openTimer); + const code = event?.code ?? null; + const reason = event?.reason || ''; + if (code === 1000) { + done = true; // normal closure + } else { + failure = new WsClosed(reason || `connection closed (${code})`, code, true); + } + settle(); + }); + + return { + [Symbol.asyncIterator]() { + return { + next() { + if (queue.length) { + return Promise.resolve({ value: queue.shift(), done: false }); + } + if (failure) return Promise.reject(failure); + if (done) return Promise.resolve({ value: undefined, done: true }); + return new Promise((resolve, reject) => { + waiters.push({ resolve, reject }); + }); + }, + return() { + try { socket.close(); } catch { /* already closing */ } + return Promise.resolve({ value: undefined, done: true }); + }, + }; + }, + }; + } + + /** + * Decide whether a failure is a permanent rejection. + * + * The server always accepts the handshake and then closes with code 1008 + * on a permanent failure — bad apikey or unknown registration_id + * ("invalid credentials or registration not found"), exhausted credits + * ("api limit reached"), or too many simultaneous devices for the same + * registration_id ("device limit reached"). Every other close code, + * including 1013 ("send timeout", the client read too slowly), is + * transient and reconnects. + * + * @returns {NewsdataWebSocketAuthError|null} + */ + #permanentAuthError(err) { + if (err instanceof WsClosed && err.code === WS_POLICY_VIOLATION) { + return new NewsdataWebSocketAuthError(err.message || 'connection rejected', err); + } + return null; + } + + /** Close the active connection, ending any in-flight `stream()`. */ + close() { + this.#closed = true; + if (this.#socket) { + try { this.#socket.close(); } catch { /* already closing */ } + this.#socket = null; + } + } +} + +/** Internal marker for a dropped connection, carrying the close code. */ +class WsClosed extends Error { + constructor(message, code, wasOpen) { + super(message); + this.name = 'WsClosed'; + this.code = code; + this.wasOpen = wasOpen; + } +} + +/** Wrap a transient failure, used only when reconnect is disabled. */ +function toTransientError(err) { + if (err instanceof WsClosed) { + return new NewsdataWebSocketError(err.code === null ? err.message : 'connection closed', err); + } + return new NewsdataWebSocketError(`connection error: ${err.message}`, err); +} diff --git a/src/index.js b/src/index.js index a266d97..486e9b5 100644 --- a/src/index.js +++ b/src/index.js @@ -2,6 +2,7 @@ // Core (re-exported so users can also use the client directly outside hooks). export { NewsDataApiClient, redactApiKey } from './core/client.js'; +export { NewsDataApiWebSocket } from './core/websocket.js'; export { validateParams } from './core/validator.js'; export * as constants from './core/constants.js'; export { @@ -12,6 +13,8 @@ export { NewsdataRateLimitError, NewsdataServerError, NewsdataNetworkError, + NewsdataWebSocketError, + NewsdataWebSocketAuthError, } from './core/errors.js'; // React layer. @@ -28,3 +31,4 @@ export { useCryptoCount, useMarketCount, } from './react/hooks.js'; +export { useNewsStream } from './react/useNewsStream.js'; diff --git a/src/react/useNewsStream.js b/src/react/useNewsStream.js new file mode 100644 index 0000000..44c38b5 --- /dev/null +++ b/src/react/useNewsStream.js @@ -0,0 +1,105 @@ +// React hook for the real-time WebSocket stream. +// +// const { articles, latest, error, isConnected } = useNewsStream(registrationId); +// +// The hook owns the connection for the lifetime of the component: it opens on +// mount (once `registrationId` is set), closes on unmount, and reconnects +// through the same capped exponential backoff the core client uses. +// +// Articles accumulate newest-first and are capped at `maxArticles` so a +// long-lived stream cannot grow without bound. + +import { useEffect, useMemo, useRef, useState } from 'react'; +import { NewsDataApiWebSocket } from '../core/websocket.js'; +import { useNewsDataClient } from './context.js'; + +const DEFAULT_MAX_ARTICLES = 100; + +/** + * Stream the news matching a registered real-time query. + * + * Register the query first — with `useNewsDataClient().websocketRegister(...)` + * or out of band — and pass its `registration_id` here. + * + * @param {string|null|undefined} registrationId + * The registered query to stream. Falsy defers connecting, which is useful + * while the id is still being fetched. + * @param {object} [options] + * @param {boolean} [options.enabled=true] Defer connecting when false. + * @param {number} [options.maxArticles=100] + * Cap on retained articles; older ones are dropped. + * @param {boolean} [options.reconnect=true] Auto-reconnect on transient drops. + * @param {string} [options.baseUrl] WebSocket endpoint override. + * @param {Function} [options.WebSocket] WebSocket implementation override. + * @returns {{ + * articles: object[], + * latest: object|null, + * error: Error|null, + * isConnected: boolean, + * }} + */ +export function useNewsStream(registrationId, options) { + const client = useNewsDataClient(); + + const enabled = options?.enabled ?? true; + const maxArticles = options?.maxArticles ?? DEFAULT_MAX_ARTICLES; + const reconnect = options?.reconnect ?? true; + const baseUrl = options?.baseUrl; + const WebSocketImpl = options?.WebSocket; + + const [articles, setArticles] = useState([]); + const [error, setError] = useState(null); + const [isConnected, setIsConnected] = useState(false); + + // Read the cap from a ref so raising it doesn't tear down the connection. + const maxArticlesRef = useRef(maxArticles); + maxArticlesRef.current = maxArticles; + + // Identity of the socket config; changing any of these reconnects. + const socketOptions = useMemo( + () => ({ reconnect, baseUrl, WebSocket: WebSocketImpl }), + [reconnect, baseUrl, WebSocketImpl], + ); + + useEffect(() => { + if (!enabled || !registrationId) { + setIsConnected(false); + return undefined; + } + + let cancelled = false; + setError(null); + + const ws = new NewsDataApiWebSocket(client, socketOptions); + + (async () => { + try { + for await (const response of ws.stream(registrationId)) { + if (cancelled) break; + setIsConnected(true); + const incoming = Array.isArray(response.results) ? response.results : []; + if (incoming.length === 0) continue; + setArticles((prev) => [...incoming, ...prev].slice(0, maxArticlesRef.current)); + } + } catch (err) { + if (!cancelled) { + setError(err); + setIsConnected(false); + } + } + })(); + + return () => { + cancelled = true; + ws.close(); + setIsConnected(false); + }; + }, [client, registrationId, enabled, socketOptions]); + + return { + articles, + latest: articles[0] ?? null, + error, + isConnected, + }; +} diff --git a/test/exports.test.js b/test/exports.test.js index 924b9df..6d7031d 100644 --- a/test/exports.test.js +++ b/test/exports.test.js @@ -31,6 +31,11 @@ test('exposes one hook per endpoint', () => { } }); +test('exposes the real-time stream hook and WebSocket class', () => { + assert.equal(typeof pkg.useNewsStream, 'function'); + assert.equal(typeof pkg.NewsDataApiWebSocket, 'function'); +}); + test('re-exports the core client and error hierarchy', () => { assert.equal(typeof pkg.NewsDataApiClient, 'function'); for (const name of [ @@ -41,6 +46,8 @@ test('re-exports the core client and error hierarchy', () => { 'NewsdataRateLimitError', 'NewsdataServerError', 'NewsdataNetworkError', + 'NewsdataWebSocketError', + 'NewsdataWebSocketAuthError', ]) { assert.equal(typeof pkg[name], 'function', `${name} should be a class`); } diff --git a/test/websocket.test.js b/test/websocket.test.js new file mode 100644 index 0000000..9a58830 --- /dev/null +++ b/test/websocket.test.js @@ -0,0 +1,331 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { NewsDataApiClient } from '../src/core/client.js'; +import { NewsDataApiWebSocket } from '../src/core/websocket.js'; +import { + NewsdataValidationError, + NewsdataWebSocketAuthError, + NewsdataWebSocketError, +} from '../src/core/errors.js'; + +function mockResponse(status, body) { + return { + status, + headers: new Headers(), + text: async () => (typeof body === 'string' ? body : JSON.stringify(body)), + }; +} + +function stubFetch(responses) { + const calls = []; + const queue = [...responses]; + const fn = async (url, init = {}) => { + calls.push({ url, method: init.method ?? 'GET' }); + const next = queue.shift(); + if (typeof next === 'function') return next(); + return next ?? mockResponse(200, { status: 'success', results: {} }); + }; + fn.calls = calls; + return fn; +} + +/** + * A scriptable stand-in for the global WebSocket. `script` runs with the + * socket instance once listeners are attached, and drives the events. + */ +function fakeWebSocketFactory(script) { + const instances = []; + class FakeWebSocket { + constructor(url) { + this.url = url; + this.listeners = new Map(); + this.closed = false; + instances.push(this); + // Let the caller attach listeners before anything fires. + queueMicrotask(() => script(this, instances.length)); + } + + addEventListener(type, fn) { + if (!this.listeners.has(type)) this.listeners.set(type, []); + this.listeners.get(type).push(fn); + } + + emit(type, event) { + for (const fn of this.listeners.get(type) ?? []) fn(event); + } + + open() { this.emit('open', {}); } + + message(data) { + this.emit('message', { data: typeof data === 'string' ? data : JSON.stringify(data) }); + } + + drop(code = 1006, reason = '') { this.emit('close', { code, reason }); } + + close() { + if (this.closed) return; + this.closed = true; + this.emit('close', { code: 1000, reason: '' }); + } + } + FakeWebSocket.instances = instances; + return FakeWebSocket; +} + +const article = (id, title) => JSON.stringify({ + status: 'success', totalResults: 1, results: [{ article_id: id, title }], +}); + +function wsClient(fetchStub = stubFetch([])) { + return new NewsDataApiClient('key', { fetch: fetchStub }); +} + +test('stream yields each response as it arrives', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.message(article('a1', 'one')); + socket.message(article('a2', 'two')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + const titles = []; + for await (const response of ws.stream('reg-1')) { + titles.push(response.results[0].title); + if (titles.length === 2) break; + } + assert.deepEqual(titles, ['one', 'two']); +}); + +test('stream sends apikey and registration_id in the query', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.message(article('a1', 'one')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-42')) break; + + const { url } = FakeWebSocket.instances[0]; + assert.match(url, /apikey=key/); + assert.match(url, /registration_id=reg-42/); +}); + +test('stream skips malformed frames', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.message('not json at all'); + socket.message(article('a1', 'one')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + const seen = []; + for await (const response of ws.stream('reg-1')) { + seen.push(response.results[0].title); + break; + } + assert.deepEqual(seen, ['one'], 'the malformed frame should be skipped, not yielded'); +}); + +test('close code 1008 raises a permanent auth error and does not reconnect', async () => { + let connections = 0; + const FakeWebSocket = fakeWebSocketFactory((socket, n) => { + connections = n; + socket.drop(1008, 'quota exhausted'); + }); + // reconnect stays ON to prove a permanent rejection is not retried. + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + await assert.rejects( + async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-1')) { /* unreachable */ } + }, + (err) => { + assert.ok(err instanceof NewsdataWebSocketAuthError, `got ${err.name}`); + assert.match(err.message, /quota exhausted/); + return true; + }, + ); + assert.equal(connections, 1, 'a permanent rejection must not retry'); +}); + +// The server always accepts the handshake, then closes with 1008 carrying the +// reason. These are the three documented permanent rejections. +for (const reason of [ + 'invalid credentials or registration not found', + 'api limit reached', + 'device limit reached', +]) { + test(`close 1008 "${reason}" is permanent`, async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.drop(1008, reason); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + await assert.rejects( + async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-1')) { /* unreachable */ } + }, + (err) => { + assert.ok(err instanceof NewsdataWebSocketAuthError, `got ${err.name}`); + assert.match(err.message, new RegExp(reason)); + return true; + }, + ); + }); +} + +// 1013 ("send timeout" — the client read too slowly) is transient. +test('close 1013 is transient and reconnects', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket, n) => { + if (n === 1) { + socket.open(); + socket.drop(1013, 'send timeout'); + return; + } + socket.open(); + socket.message(article('a1', 'after-reconnect')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { + WebSocket: FakeWebSocket, + reconnectDelay: 1, + reconnectDelayMax: 2, + }); + + const titles = []; + for await (const response of ws.stream('reg-1')) { + titles.push(response.results[0].title); + break; + } + assert.deepEqual(titles, ['after-reconnect']); +}); + +test('a transient drop stops with a websocket error when reconnect is disabled', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.drop(1006); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { + WebSocket: FakeWebSocket, + reconnect: false, + }); + + await assert.rejects( + async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-1')) { /* unreachable */ } + }, + (err) => { + assert.ok(err instanceof NewsdataWebSocketError, `got ${err.name}`); + assert.ok(!(err instanceof NewsdataWebSocketAuthError), 'should not be an auth error'); + return true; + }, + ); +}); + +test('a transient drop reconnects when reconnect is enabled', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket, n) => { + if (n === 1) { + socket.open(); + socket.drop(1006); // transient + return; + } + socket.open(); + socket.message(article('a1', 'after-reconnect')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { + WebSocket: FakeWebSocket, + reconnectDelay: 1, + reconnectDelayMax: 2, + }); + + const titles = []; + for await (const response of ws.stream('reg-1')) { + titles.push(response.results[0].title); + break; + } + assert.deepEqual(titles, ['after-reconnect']); + assert.ok(FakeWebSocket.instances.length >= 2, 'should have reconnected'); +}); + +test('breaking out of the loop closes the socket', async () => { + const FakeWebSocket = fakeWebSocketFactory((socket) => { + socket.open(); + socket.message(article('a1', 'one')); + }); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('reg-1')) break; + + assert.equal(FakeWebSocket.instances[0].closed, true); +}); + +test('stream rejects an empty registration id', async () => { + const FakeWebSocket = fakeWebSocketFactory(() => {}); + const ws = new NewsDataApiWebSocket(wsClient(), { WebSocket: FakeWebSocket }); + await assert.rejects( + async () => { + // eslint-disable-next-line no-unused-vars + for await (const _ of ws.stream('')) { /* unreachable */ } + }, + NewsdataValidationError, + ); +}); + +// ---- query management --------------------------------------------------- + +test('websocketRegister POSTs and injects news_type=latest', async () => { + const fetchStub = stubFetch([ + mockResponse(200, { status: 'success', results: { registration_id: 'reg-9' } }), + ]); + const client = wsClient(fetchStub); + const res = await client.websocketRegister({ q: 'bitcoin' }); + + assert.equal(res.results.registration_id, 'reg-9'); + assert.equal(fetchStub.calls[0].method, 'POST'); + assert.match(fetchStub.calls[0].url, /news_type=latest/); + assert.match(fetchStub.calls[0].url, /q=bitcoin/); + assert.match(fetchStub.calls[0].url, /websocket\/register/); +}); + +test('websocketFetch GETs the fetch endpoint', async () => { + const fetchStub = stubFetch([ + mockResponse(200, { status: 'success', results: { queries: [] } }), + ]); + await wsClient(fetchStub).websocketFetch(); + assert.equal(fetchStub.calls[0].method, 'GET'); + assert.match(fetchStub.calls[0].url, /websocket\/fetch/); +}); + +test('websocketDelete uses DELETE and carries registration_id', async () => { + const fetchStub = stubFetch([ + mockResponse(200, { status: 'success', results: { deleted: true } }), + ]); + await wsClient(fetchStub).websocketDelete('reg-9'); + assert.equal(fetchStub.calls[0].method, 'DELETE'); + assert.match(fetchStub.calls[0].url, /registration_id=reg-9/); +}); + +test('websocketDelete rejects an empty id', () => { + assert.throws(() => wsClient().websocketDelete(''), NewsdataValidationError); +}); + +test('a resultless success envelope still succeeds on the websocket endpoints', async () => { + const fetchStub = stubFetch([mockResponse(200, { status: 'success' })]); + const res = await wsClient(fetchStub).websocketDelete('reg-9'); + assert.equal(res.status, 'success'); +}); + +test('the WebSocket class delegates management calls to the client', async () => { + const fetchStub = stubFetch([ + mockResponse(200, { status: 'success', results: { registration_id: 'reg-7' } }), + ]); + const FakeWebSocket = fakeWebSocketFactory(() => {}); + const ws = new NewsDataApiWebSocket(wsClient(fetchStub), { WebSocket: FakeWebSocket }); + const res = await ws.websocketRegister({ q: 'x' }); + assert.equal(res.results.registration_id, 'reg-7'); +}); diff --git a/types/index.d.ts b/types/index.d.ts index 45ddafd..d12f7a8 100644 --- a/types/index.d.ts +++ b/types/index.d.ts @@ -56,6 +56,41 @@ export class NewsDataApiClient { cryptoCountApi(params?: EndpointParams): EndpointResult; marketCountApi(params?: EndpointParams): EndpointResult; sourcesApi(params?: EndpointParams): Promise; + + /** Register a real-time WebSocket query. POST /1/websocket/register */ + websocketRegister(params?: EndpointParams): Promise; + /** List the account's registered real-time queries. GET /1/websocket/fetch */ + websocketFetch(): Promise; + /** Delete a registered real-time query. DELETE /1/websocket/delete */ + websocketDelete(registrationId: string): Promise; +} + +export interface WebSocketOptions { + /** WebSocket endpoint; defaults to wss://ws.newsdata.io/ws/event. */ + baseUrl?: string; + /** Reconnect automatically on transient drops. Default true. */ + reconnect?: boolean; + /** Milliseconds before the first reconnect; doubles after each failure. */ + reconnectDelay?: number; + /** Upper bound on the reconnect delay, in milliseconds. */ + reconnectDelayMax?: number; + /** Milliseconds to wait for the opening handshake. */ + openTimeout?: number; + /** WebSocket implementation; defaults to the browser's global. */ + WebSocket?: new (url: string) => unknown; +} + +/** + * NewsData.io real-time WebSocket service. Use `useNewsStream` inside + * components; this class is for imperative use outside React. + */ +export class NewsDataApiWebSocket { + constructor(client: NewsDataApiClient, options?: WebSocketOptions); + websocketRegister(params?: EndpointParams): Promise; + websocketFetch(): Promise; + websocketDelete(registrationId: string): Promise; + stream(registrationId: string): AsyncGenerator; + close(): void; } export function redactApiKey(url: string): string; @@ -83,6 +118,10 @@ export class NewsdataServerError extends NewsdataApiError {} export class NewsdataNetworkError extends NewsdataError { cause?: Error; } +export class NewsdataWebSocketError extends NewsdataError { + cause?: Error; +} +export class NewsdataWebSocketAuthError extends NewsdataWebSocketError {} // ---- React layer -------------------------------------------------------- @@ -131,6 +170,39 @@ export function useNewsCount(params?: EndpointParams, options?: UseQueryOptions) export function useCryptoCount(params?: EndpointParams, options?: UseQueryOptions): UseQueryResult; export function useMarketCount(params?: EndpointParams, options?: UseQueryOptions): UseQueryResult; +export interface UseNewsStreamOptions { + /** Defer connecting when false. Default true. */ + enabled?: boolean; + /** Cap on retained articles; older ones are dropped. Default 100. */ + maxArticles?: number; + /** Reconnect automatically on transient drops. Default true. */ + reconnect?: boolean; + /** WebSocket endpoint override. */ + baseUrl?: string; + /** WebSocket implementation override. */ + WebSocket?: new (url: string) => unknown; +} + +export interface UseNewsStreamResult { + /** Articles received so far, newest first, capped at `maxArticles`. */ + articles: object[]; + /** The most recent article, or null before the first arrives. */ + latest: object | null; + /** A permanent rejection or, with `reconnect: false`, a stream error. */ + error: Error | null; + /** True once the first response has arrived on the current connection. */ + isConnected: boolean; +} + +/** + * Stream the news matching a registered real-time query. Opens on mount, + * closes on unmount, reconnects on transient drops. + */ +export function useNewsStream( + registrationId: string | null | undefined, + options?: UseNewsStreamOptions, +): UseNewsStreamResult; + export const constants: { BASE_URL: string; ENDPOINTS: Record;