From caa8d23c9999a860f417e70cc206baf51210cfb3 Mon Sep 17 00:00:00 2001 From: arjunjain Date: Wed, 26 Aug 2026 19:28:13 +0530 Subject: [PATCH] Add real-time WebSocket support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sync with python-client 0.3.0. Adds the three query-management endpoints (get_websocket_register / _fetch / _delete) and NewsdataWebSocket::stream, a generator over the news matching a registered query. phrity/websocket is an OPTIONAL dependency (suggest only), not a hard require: it needs PHP 8.1 while this package supports 7.3+, and CI runs 7.4 through 8.4 — requiring it would break composer install on the older jobs for REST users who may never stream. Everything except stream() works without it; stream() throws a NewsdataWebSocketError naming the package when it is missing. The class is referenced dynamically so the file loads and analyses without it installed. The server accepts every handshake and then closes with code 1008 on a permanent rejection (invalid credentials, api limit, device limit); those throw NewsdataWebSocketAuthError and are never retried. execute() now carries a per-endpoint HTTP method, and the websocket endpoints are exempt from the results-present success check. --- README.md | 105 +++++- composer.json | 19 +- examples/websocket.php | 78 ++++ src/Constants.php | 50 +++ src/Exception/NewsdataWebSocketAuthError.php | 14 + src/Exception/NewsdataWebSocketError.php | 14 + src/NewsdataApi.php | 70 ++++ src/NewsdataApiBase.php | 30 +- src/NewsdataWebSocket.php | 352 +++++++++++++++++++ tests/NewsdataWebSocketTest.php | 279 +++++++++++++++ 10 files changed, 994 insertions(+), 17 deletions(-) create mode 100644 examples/websocket.php create mode 100644 src/Exception/NewsdataWebSocketAuthError.php create mode 100644 src/Exception/NewsdataWebSocketError.php create mode 100644 src/NewsdataWebSocket.php create mode 100644 tests/NewsdataWebSocketTest.php diff --git a/README.md b/README.md index 2a7d860..eb3604e 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,9 @@ The official PHP client for the [Newsdata.io](https://newsdata.io) REST API. It wraps every endpoint (`latest`, `archive`, `sources`, `crypto`, `market`, `count`, `crypto/count`, `market/count`) with client-side parameter validation, -automatic retries with exponential backoff, and a typed exception hierarchy. +automatic retries with exponential backoff, and a typed exception hierarchy. It +also covers the real-time WebSocket service: register, list, and delete queries, +and stream the matching news as it is published. ## Requirements @@ -75,6 +77,9 @@ By default the response is decoded to objects; call | `get_news_count($data)` | `/1/count` | Aggregate counts (requires `from_date`, `to_date`) | | `get_crypto_count($data)` | `/1/crypto/count` | Aggregate crypto counts (requires dates) | | `get_market_count($data)` | `/1/market/count` | Aggregate market counts (requires dates) | +| `get_websocket_register($data)` | `/1/websocket/register` | Register a real-time query | +| `get_websocket_fetch()` | `/1/websocket/fetch` | List registered queries | +| `get_websocket_delete($id)` | `/1/websocket/delete` | Delete a registered query | Each `$data` value may be a single string or an array of strings. Parameter names are case-insensitive. See the @@ -118,6 +123,100 @@ Before any request is sent, parameters are validated and normalized. A Booleans (`full_content`, `image`, `video`, `removeduplicate`) are coerced to `1` / `0`. +## Real-time news (WebSocket) + +Register a query first — the returned `registration_id` identifies it from then on: + +```php +use NewsdataIO\NewsdataApi; +use NewsdataIO\NewsdataWebSocket; + +$api = new NewsdataApi('YOUR_API_KEY'); +$ws = new NewsdataWebSocket($api); + +$registered = $ws->register(['q' => 'bitcoin', 'language' => 'en']); +$registrationId = $registered->results->registration_id; +``` + +`register()` takes the familiar filter names (`q`, `country`, `language`, +`domain`, …) — no date or paging filters, since a registered query matches news +as it is published. Registering an identical query twice throws +`NewsdataAPIError` with status 409; the existing id is in the response body. +`fetch()` lists every registered query and `delete($id)` removes one. All three +also exist directly on the API object as `get_websocket_register()`, +`get_websocket_fetch()` and `get_websocket_delete()`. + +Then stream. `stream()` is a generator — `break` out of the loop to stop, and +the connection closes for you: + +```php +foreach ($ws->stream($registrationId) as $response) { + foreach ($response->results as $article) { + echo $article->title, ' - ', $article->link, PHP_EOL; + } +} +``` + +Transient drops (network errors, server restarts, abnormal closes) are +reconnected automatically with a capped exponential backoff. Pass +`'reconnect' => false` to stop on the first disconnect instead. A permanent +rejection — bad API key or unknown +`registration_id`, exhausted API credits, or too many simultaneous devices — throws +`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.** + +Catch it like any other client error: + +```php +use NewsdataIO\Exception\NewsdataWebSocketAuthError; +use NewsdataIO\Exception\NewsdataWebSocketError; + +try { + foreach ($ws->stream($registrationId) as $response) { + // ... + } +} catch (NewsdataWebSocketAuthError $e) { + echo 'rejected: ', $e->getMessage(), PHP_EOL; +} catch (NewsdataWebSocketError $e) { + echo 'stream error: ', $e->getMessage(), PHP_EOL; +} +``` + +All connection options are optional: + +```php +$ws = new NewsdataWebSocket($api, [ + 'baseUrl' => 'wss://ws.newsdata.io/ws/event', // staging / self-hosted + 'reconnect' => true, // auto-reconnect on transient drops; default true + 'reconnectDelay' => 1.0, // seconds before the first reconnect (doubles each retry) + 'reconnectDelayMax' => 30.0, // cap on the reconnect delay + 'handshakeTimeout' => 10, // seconds to wait for the opening handshake +]); +``` + +> **Streaming needs one extra package.** PHP has no WebSocket client in core, so +> `stream()` requires [`phrity/websocket`](https://packagist.org/packages/phrity/websocket) +> (PHP 8.1+): +> +> ```bash +> composer require phrity/websocket +> ``` +> +> It is an **optional** dependency — everything else in this SDK, including the +> three `websocket/*` management endpoints above, works without it on every +> supported PHP version. `stream()` throws a `NewsdataWebSocketError` telling +> you to install it if it is missing. + +Runnable example: [`examples/websocket.php`](examples/websocket.php). + ## Error handling ```php @@ -151,7 +250,9 @@ NewsdataException (catch-all base) │ ├── NewsdataAuthError (401 / 403) │ ├── NewsdataRateLimitError (429; getRetryAfter()) │ └── NewsdataServerError (5xx) -└── NewsdataNetworkError (cURL / connectivity) +├── NewsdataNetworkError (cURL / connectivity) +└── NewsdataWebSocketError (real-time stream) + └── NewsdataWebSocketAuthError (policy-violation close 1008) ``` ## Configuration diff --git a/composer.json b/composer.json index 1d61654..2cf9a8c 100644 --- a/composer.json +++ b/composer.json @@ -1,8 +1,20 @@ { "name": "newsdataio/newsdataapi", "type": "library", - "description": "Official PHP client (SDK) for the Newsdata.io News API — fetch real-time, historical, crypto, and stock-market news via REST with validation, retries, and error handling.", - "keywords": ["news", "newsdata", "latest", "archive", "crypto", "market", "sources", "count", "historical", "api", "sdk"], + "description": "Official PHP client (SDK) for the Newsdata.io News API \u2014 fetch real-time, historical, crypto, and stock-market news via REST with validation, retries, and error handling.", + "keywords": [ + "news", + "newsdata", + "latest", + "archive", + "crypto", + "market", + "sources", + "count", + "historical", + "api", + "sdk" + ], "license": "MIT", "homepage": "https://newsdata.io", "authors": [ @@ -28,7 +40,8 @@ "phpunit/phpunit": "^9 || ^10" }, "suggest": { - "psr/log": "Allows attaching a PSR-3 logger via NewsdataApi::setLogger()" + "psr/log": "Allows attaching a PSR-3 logger via NewsdataApi::setLogger()", + "phrity/websocket": "Required for real-time streaming via NewsdataWebSocket::stream() (PHP 8.1+)" }, "autoload": { "psr-4": { diff --git a/examples/websocket.php b/examples/websocket.php new file mode 100644 index 0000000..dc661ce --- /dev/null +++ b/examples/websocket.php @@ -0,0 +1,78 @@ + php examples/websocket.php + * + * Streaming needs the optional phrity/websocket package (PHP 8.1+): + * + * composer require phrity/websocket + * + * Articles are matched by a registered query. If NEWSDATA_REGISTRATION_ID is + * set, that query is streamed directly; otherwise this registers a demo query + * (q="pizza") first and prints the resulting registration_id so you can reuse + * it on the next run — or remove it later with $ws->delete($id). + */ + +declare(strict_types=1); + +require_once __DIR__ . '/../vendor/autoload.php'; + +use NewsdataIO\Exception\NewsdataAPIError; +use NewsdataIO\Exception\NewsdataWebSocketAuthError; +use NewsdataIO\Exception\NewsdataWebSocketError; +use NewsdataIO\NewsdataApi; +use NewsdataIO\NewsdataWebSocket; + +$apiKey = getenv('NEWSDATA_API_KEY'); +if ($apiKey === false || $apiKey === '') { + fwrite(STDERR, "Set NEWSDATA_API_KEY in your environment before running this example.\n"); + exit(1); +} + +$api = new NewsdataApi($apiKey); +$ws = new NewsdataWebSocket($api); + +/** + * Register q="pizza" and return its registration_id. Registering an identical + * query again answers HTTP 409 with the existing id in the response body — + * reuse it instead of failing. + */ +function registerDemoQuery(NewsdataWebSocket $ws): string +{ + try { + $response = $ws->register(['q' => 'pizza']); + $id = $response->results->registration_id; + echo "registered demo query q=\"pizza\" -> {$id}\n"; + return $id; + } catch (NewsdataAPIError $e) { + // getResponseBody() decodes to an array regardless of the client's + // setDecodeJsonAsArray() setting. + $body = $e->getResponseBody(); + $existing = $body['results']['registration_id'] ?? null; + if ($e->getStatusCode() === 409 && $existing !== null) { + echo "query already registered; reusing {$existing}\n"; + return $existing; + } + throw $e; + } +} + +$registrationId = getenv('NEWSDATA_REGISTRATION_ID') ?: registerDemoQuery($ws); + +echo "streaming {$registrationId} — Ctrl-C to stop\n"; + +try { + foreach ($ws->stream($registrationId) as $response) { + foreach ($response->results as $article) { + echo $article->title, ' - ', $article->link, PHP_EOL; + } + } +} catch (NewsdataWebSocketAuthError $e) { + fwrite(STDERR, 'rejected: ' . $e->getMessage() . PHP_EOL); + exit(1); +} catch (NewsdataWebSocketError $e) { + fwrite(STDERR, 'stream error: ' . $e->getMessage() . PHP_EOL); + exit(1); +} diff --git a/src/Constants.php b/src/Constants.php index a608113..d7ee9ae 100644 --- a/src/Constants.php +++ b/src/Constants.php @@ -60,8 +60,45 @@ final class Constants '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. */ + public const ENDPOINT_METHODS = [ + 'websocket_register' => 'POST', + 'websocket_delete' => 'DELETE', + ]; + + /** + * Endpoints whose success envelope may carry no `results` field, so they + * are exempt from the results-present check applied to the news endpoints. + */ + public const RESULTS_OPTIONAL = [ + 'websocket_register', + 'websocket_fetch', + 'websocket_delete', + ]; + + /** Real-time WebSocket endpoint. */ + public const WS_BASE_URL = 'wss://ws.newsdata.io/ws/event'; + + /** The feed a registered query matches against. */ + public const WS_NEWS_TYPE = 'latest'; + + /** Close code the server uses for a permanent connection rejection. */ + public const WS_POLICY_VIOLATION = 1008; + + /** Seconds before the first reconnect; doubles after each failure. */ + public const WS_RECONNECT_DELAY = 1.0; + + /** Upper bound on the reconnect delay, in seconds. */ + public const WS_RECONNECT_DELAY_MAX = 30.0; + + /** Bound on the opening handshake, in seconds. */ + public const WS_HANDSHAKE_TIMEOUT = 10; + /** Endpoints that require both `from_date` and `to_date`. */ public const REQUIRES_DATE_RANGE = ['count', 'crypto_count', 'market_count']; @@ -145,5 +182,18 @@ final class Constants '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 + // get_websocket_register(), 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'], ]; } diff --git a/src/Exception/NewsdataWebSocketAuthError.php b/src/Exception/NewsdataWebSocketAuthError.php new file mode 100644 index 0000000..f526df3 --- /dev/null +++ b/src/Exception/NewsdataWebSocketAuthError.php @@ -0,0 +1,14 @@ +request('market_count', $data); } + + // ---- 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 {@see NewsdataWebSocket::stream()}. + * + * Registering an identical query twice throws + * {@see Exception\NewsdataAPIError} with status 409; the existing id is in + * the response body. + * + * @param array $data + * + * @return array|object + */ + public function get_websocket_register(array $data = []) + { + $data['news_type'] = Constants::WS_NEWS_TYPE; + return $this->request('websocket_register', $data); + } + + /** + * List the account's registered real-time queries. + * GET /1/websocket/fetch. One entry per query at `results.queries`. + * + * @return array|object + */ + public function get_websocket_fetch() + { + return $this->request('websocket_fetch', []); + } + + /** + * Delete a registered real-time query. DELETE /1/websocket/delete. + * + * @param string $registrationId + * + * @return array|object + */ + public function get_websocket_delete(string $registrationId) + { + if ($registrationId === '') { + throw new Exception\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 */ + public function apiKeyForWebSocket(): string + { + return $this->apiKey; + } + + /** + * Whether responses decode to arrays rather than objects; the WebSocket + * layer decodes frames the same way. + * + * @internal + */ + public function isDecodingJsonAsArray(): bool + { + return $this->decodeJsonAsArray; + } } diff --git a/src/NewsdataApiBase.php b/src/NewsdataApiBase.php index 4d84cbb..b049eed 100644 --- a/src/NewsdataApiBase.php +++ b/src/NewsdataApiBase.php @@ -86,7 +86,7 @@ protected function request(string $endpoint, array $data) } $params = ParamValidator::validate($endpoint, $data); $baseUrl = Constants::BASE_URL . Constants::ENDPOINTS[$endpoint]; - return $this->execute($baseUrl, $params); + return $this->execute($baseUrl, $params, $endpoint); } /** @@ -97,7 +97,7 @@ protected function request(string $endpoint, array $data) * * @return array|object */ - private function execute(string $baseUrl, array $params) + private function execute(string $baseUrl, array $params, string $endpoint = '') { $this->response = new Response(); $this->response->setApiPath($baseUrl); @@ -108,12 +108,13 @@ private function execute(string $baseUrl, array $params) $logUrl = Util::redactApiKey($fullUrl); $attempts = max(1, $this->maxRetries); + $method = Constants::ENDPOINT_METHODS[$endpoint] ?? 'GET'; for ($attempt = 1; $attempt <= $attempts; $attempt++) { - $this->log('info', "GET {$logUrl} (attempt {$attempt}/{$attempts})"); + $this->log('info', "{$method} {$logUrl} (attempt {$attempt}/{$attempts})"); try { - [$status, $headers, $rawBody] = $this->httpGet($fullUrl); + [$status, $headers, $rawBody] = $this->httpGet($fullUrl, $method); } catch (NewsdataNetworkError $e) { if ($attempt >= $attempts) { throw $e; @@ -140,7 +141,7 @@ private function execute(string $baseUrl, array $params) } $this->response->setBody($body); - if ($status === 200 && $this->isSuccessBody($body)) { + if ($status === 200 && $this->isSuccessBody($body, $endpoint)) { return $body; } @@ -206,10 +207,10 @@ private function execute(string $baseUrl, array $params) * * @throws NewsdataNetworkError */ - private function httpGet(string $url): array + private function httpGet(string $url, string $method = 'GET'): array { $ch = curl_init(); - curl_setopt_array($ch, $this->curlOptions($url)); + curl_setopt_array($ch, $this->curlOptions($url, $method)); $raw = curl_exec($ch); @@ -238,10 +239,11 @@ private function httpGet(string $url): array * * @return array */ - private function curlOptions(string $url): array + private function curlOptions(string $url, string $method = 'GET'): array { $options = [ CURLOPT_URL => $url, + CURLOPT_CUSTOMREQUEST => $method, CURLOPT_HTTPHEADER => ['Accept: application/json'], CURLOPT_CONNECTTIMEOUT => $this->connectionTimeout, CURLOPT_TIMEOUT => $this->timeout, @@ -305,16 +307,20 @@ private function jsonDecode(string $json) * * @param mixed $body */ - private function isSuccessBody($body): bool + private function isSuccessBody($body, string $endpoint = ''): bool { + // The websocket management endpoints may answer without `results`. + $resultsOptional = in_array($endpoint, Constants::RESULTS_OPTIONAL, true); + if (is_array($body)) { return ($body['status'] ?? null) === 'success' - && array_key_exists('results', $body) - && $body['results'] !== null; + && ($resultsOptional + || (array_key_exists('results', $body) && $body['results'] !== null)); } if ($body instanceof \stdClass) { return isset($body->status) && $body->status === 'success' - && property_exists($body, 'results') && $body->results !== null; + && ($resultsOptional + || (property_exists($body, 'results') && $body->results !== null)); } return false; } diff --git a/src/NewsdataWebSocket.php b/src/NewsdataWebSocket.php new file mode 100644 index 0000000..fe523e9 --- /dev/null +++ b/src/NewsdataWebSocket.php @@ -0,0 +1,352 @@ +register(['q' => 'bitcoin']); + * $id = $registered['results']['registration_id']; + * + * foreach ($ws->stream($id) as $response) { + * foreach ($response['results'] as $article) { + * echo $article['title'], PHP_EOL; + * } + * } + * + * `stream()` is a generator: `break` out of the loop to stop, and the + * connection is closed for you. + * + * Transient drops (network errors, server restarts, abnormal closes) are + * reconnected automatically with a capped exponential backoff; pass + * `'reconnect' => false` to stop on the first disconnect. A permanent + * rejection always throws {@see NewsdataWebSocketAuthError} and is never + * retried. + * + * Streaming needs the optional `phrity/websocket` package (PHP 8.1+): + * + * composer require phrity/websocket + * + * The REST management calls below work without it, on every supported PHP. + */ +class NewsdataWebSocket +{ + /** @var NewsdataApi */ + private $api; + + /** @var string */ + private $baseUrl; + + /** @var bool */ + private $reconnect; + + /** @var float */ + private $reconnectDelay; + + /** @var float */ + private $reconnectDelayMax; + + /** @var int */ + private $handshakeTimeout; + + /** @var callable|null Injection point for tests. */ + private $connector; + + /** @var bool */ + private $closed = false; + + /** @var mixed The live connection, while one is open. */ + private $connection; + + /** + * @param NewsdataApi $api Supplies the API key and performs the + * management HTTP calls. Not closed by this + * class. + * @param array $options baseUrl, reconnect, reconnectDelay, + * reconnectDelayMax, handshakeTimeout, + * connector (callable, for tests). + */ + public function __construct(NewsdataApi $api, array $options = []) + { + $this->api = $api; + $this->baseUrl = $options['baseUrl'] ?? Constants::WS_BASE_URL; + $this->reconnect = $options['reconnect'] ?? true; + $this->reconnectDelay = (float) ($options['reconnectDelay'] ?? Constants::WS_RECONNECT_DELAY); + $this->reconnectDelayMax = (float) ($options['reconnectDelayMax'] ?? Constants::WS_RECONNECT_DELAY_MAX); + $this->handshakeTimeout = (int) ($options['handshakeTimeout'] ?? Constants::WS_HANDSHAKE_TIMEOUT); + $this->connector = $options['connector'] ?? null; + } + + // ---- query management ------------------------------------------------- + + /** + * Register a real-time query. + * + * @see NewsdataApi::get_websocket_register() + * + * @param array $data + * + * @return array|object + */ + public function register(array $data = []) + { + return $this->api->get_websocket_register($data); + } + + /** + * List the account's registered real-time queries. + * + * @see NewsdataApi::get_websocket_fetch() + * + * @return array|object + */ + public function fetch() + { + return $this->api->get_websocket_fetch(); + } + + /** + * Delete a registered real-time query. + * + * @see NewsdataApi::get_websocket_delete() + * + * @param string $registrationId + * + * @return array|object + */ + public function delete(string $registrationId) + { + return $this->api->get_websocket_delete($registrationId); + } + + // ---- streaming -------------------------------------------------------- + + /** + * Build the handshake URL. + */ + private function url(string $registrationId): string + { + return $this->baseUrl + . '?apikey=' . rawurlencode($this->api->apiKeyForWebSocket()) + . '®istration_id=' . rawurlencode($registrationId); + } + + private function nextDelay(float $delay): float + { + return min($delay * 2, $this->reconnectDelayMax); + } + + /** + * Connect and yield each response for $registrationId as it arrives. + * + * Responses have the familiar `status` / `totalResults` / `results` shape, + * decoded the same way the REST methods decode theirs. + * + * @param string $registrationId + * + * @return \Generator + */ + public function stream(string $registrationId): \Generator + { + if ($registrationId === '') { + throw new NewsdataValidationError( + 'registrationId must be a non-empty string', + 'registration_id' + ); + } + + $url = $this->url($registrationId); + $delay = $this->reconnectDelay; + $this->closed = false; + + try { + while (!$this->isClosed()) { + $client = null; + + try { + $client = $this->connect($url); + $this->connection = $client; + + while (!$this->isClosed()) { + $message = $client->receive(); + if ($message === null) { + break; // connection closed + } + $payload = is_string($message) + ? $message + : (is_object($message) && method_exists($message, 'getContent') + ? (string) $message->getContent() + : null); + if ($payload === null) { + continue; // ignore binary / control frames + } + $decoded = $this->decode($payload); + if ($decoded !== null) { + yield $decoded; + } + $delay = $this->reconnectDelay; // reset after real traffic + } + } catch (\Throwable $e) { + if ($this->isClosed()) { + return; + } + $auth = $this->permanentAuthError($e); + if ($auth !== null) { + throw $auth; + } + if (!$this->reconnect) { + throw $this->transientError($e); + } + } finally { + $this->closeConnection($client); + } + + if ($this->isClosed()) { + return; + } + if (!$this->reconnect) { + return; + } + usleep((int) round($delay * 1000000)); + $delay = $this->nextDelay($delay); + } + } finally { + $this->close(); + } + } + + /** + * Open one connection. Uses the injected connector when present (tests), + * otherwise the phrity/websocket client. + * + * @return mixed + */ + private function connect(string $url) + { + if ($this->connector !== null) { + return ($this->connector)($url); + } + + // Referenced dynamically: phrity/websocket is an optional dependency, + // so the class need not exist for this file to load or analyse. + $clientClass = 'WebSocket\\Client'; + $uriClass = 'Phrity\\Net\\Uri'; + + if (!class_exists($clientClass) || !class_exists($uriClass)) { + throw new NewsdataWebSocketError( + 'Real-time streaming needs the phrity/websocket package: ' + . 'composer require phrity/websocket' + ); + } + + $client = new $clientClass(new $uriClass($url)); + $client->setTimeout($this->handshakeTimeout); + $client->connect(); + return $client; + } + + /** + * Decode one frame, returning null when it isn't a JSON object. + * + * @return array|object|null + */ + private function decode(string $payload) + { + $decoded = json_decode($payload, $this->api->isDecodingJsonAsArray()); + if (json_last_error() !== JSON_ERROR_NONE) { + return null; // skip malformed frames + } + if (!is_array($decoded) && !($decoded instanceof \stdClass)) { + return null; + } + return $decoded; + } + + /** + * The auth error to throw if the failure is a permanent rejection, + * else null (the caller then treats it as transient and reconnects). + * + * Close code 1008 and handshake 401 / 403 are permanent; everything else + * is transient. + */ + private function permanentAuthError(\Throwable $e): ?NewsdataWebSocketAuthError + { + if ($e instanceof NewsdataWebSocketAuthError) { + return $e; + } + $code = $e->getCode(); + if ($code === Constants::WS_POLICY_VIOLATION) { + $reason = $e->getMessage(); + return new NewsdataWebSocketAuthError( + $reason !== '' ? $reason : 'connection rejected' + ); + } + $message = $e->getMessage(); + if (strpos($message, '401') !== false || strpos($message, '403') !== false) { + return new NewsdataWebSocketAuthError('connection rejected'); + } + return null; + } + + /** Wrap a transient failure; used only when reconnect is disabled. */ + private function transientError(\Throwable $e): NewsdataWebSocketError + { + if ($e instanceof NewsdataWebSocketError) { + return $e; + } + return new NewsdataWebSocketError('connection error: ' . $e->getMessage()); + } + + /** + * @param mixed $client + */ + private function closeConnection($client): void + { + $this->connection = null; + if ($client === null) { + return; + } + try { + if (method_exists($client, 'close')) { + $client->close(); + } elseif (method_exists($client, 'disconnect')) { + $client->disconnect(); + } + } catch (\Throwable $ignored) { + // already closing + } + } + + /** + * Whether {@see close()} has been called. + * + * Marked impure so static analysis does not fold the flag to its last + * assigned value: the stream loop yields, and the caller may close the + * connection from outside between yields. + * + * @phpstan-impure + */ + private function isClosed(): bool + { + return $this->closed; + } + + /** Close the active connection, ending any in-flight {@see stream()}. */ + public function close(): void + { + $this->closed = true; + $this->closeConnection($this->connection); + } +} diff --git a/tests/NewsdataWebSocketTest.php b/tests/NewsdataWebSocketTest.php new file mode 100644 index 0000000..53f22b3 --- /dev/null +++ b/tests/NewsdataWebSocketTest.php @@ -0,0 +1,279 @@ + */ + private $script; + + /** @var int */ + public $connectionNumber; + + /** @var bool */ + public $closed = false; + + public function __construct(array $script, int $connectionNumber) + { + $this->script = $script; + $this->connectionNumber = $connectionNumber; + } + + /** + * @return string|null + */ + public function receive() + { + if ($this->script === []) { + return null; // nothing left: behave as a closed connection + } + $next = array_shift($this->script); + if ($next instanceof \Throwable) { + throw $next; + } + return $next; // string frame, or null to close + } + + public function close(): void + { + $this->closed = true; + } +} + +/** + * Real-time WebSocket tests. The transport is injected via the `connector` + * option, so these run without a live server or the phrity/websocket package. + */ +class NewsdataWebSocketTest extends TestCase +{ + /** @var array */ + public $connections = []; + + /** @var array */ + public $urls = []; + + private function api(): NewsdataApi + { + return new NewsdataApi('key'); + } + + /** + * Build a websocket whose connector hands out scripted fake connections. + * + * @param callable $scriptFor fn(int $connectionNumber): array + */ + private function ws(callable $scriptFor, array $options = []): NewsdataWebSocket + { + $this->connections = []; + $this->urls = []; + + $options['connector'] = function (string $url) use ($scriptFor) { + $this->urls[] = $url; + $n = count($this->connections) + 1; + $conn = new FakeConnection($scriptFor($n), $n); + $this->connections[] = $conn; + return $conn; + }; + $options['reconnectDelay'] = $options['reconnectDelay'] ?? 0.001; + $options['reconnectDelayMax'] = $options['reconnectDelayMax'] ?? 0.002; + + return new NewsdataWebSocket($this->api(), $options); + } + + private static function articleFrame(string $id, string $title): string + { + return json_encode([ + 'status' => 'success', + 'totalResults' => 1, + 'results' => [['article_id' => $id, 'title' => $title]], + ]); + } + + public function testStreamYieldsEachResponse(): void + { + $ws = $this->ws(fn (int $n) => [ + self::articleFrame('a1', 'one'), + self::articleFrame('a2', 'two'), + ], ['reconnect' => false]); + + $titles = []; + foreach ($ws->stream('reg-1') as $response) { + $titles[] = $response->results[0]->title; + if (count($titles) === 2) { + break; + } + } + + $this->assertSame(['one', 'two'], $titles); + } + + public function testStreamSendsApiKeyAndRegistrationIdInQuery(): void + { + $ws = $this->ws(fn (int $n) => [self::articleFrame('a1', 'one')], ['reconnect' => false]); + + foreach ($ws->stream('reg-42') as $response) { + break; + } + + $this->assertStringContainsString('apikey=key', $this->urls[0]); + $this->assertStringContainsString('registration_id=reg-42', $this->urls[0]); + } + + public function testStreamSkipsMalformedFrames(): void + { + $ws = $this->ws(fn (int $n) => [ + 'not json at all', + self::articleFrame('a1', 'one'), + ], ['reconnect' => false]); + + $seen = []; + foreach ($ws->stream('reg-1') as $response) { + $seen[] = $response->results[0]->title; + break; + } + + $this->assertSame(['one'], $seen, 'the malformed frame should be skipped'); + } + + public function testPolicyViolationCloseIsPermanentAndNotRetried(): void + { + // reconnect stays ON to prove a permanent rejection is not retried. + $ws = $this->ws(fn (int $n) => [ + new \RuntimeException('quota exhausted', Constants::WS_POLICY_VIOLATION), + ]); + + try { + foreach ($ws->stream('reg-1') as $response) { + $this->fail('should not yield'); + } + $this->fail('expected a NewsdataWebSocketAuthError'); + } catch (NewsdataWebSocketAuthError $e) { + $this->assertStringContainsString('quota exhausted', $e->getMessage()); + } + + $this->assertCount(1, $this->connections, 'a permanent rejection must not retry'); + } + + public function testHandshake401IsPermanent(): void + { + $ws = $this->ws(fn (int $n) => [ + new \RuntimeException('Could not connect: server responded 401'), + ]); + + $this->expectException(NewsdataWebSocketAuthError::class); + foreach ($ws->stream('reg-1') as $response) { + $this->fail('should not yield'); + } + } + + public function testTransientFailureStopsWhenReconnectDisabled(): void + { + $ws = $this->ws(fn (int $n) => [ + new \RuntimeException('connection refused'), + ], ['reconnect' => false]); + + try { + foreach ($ws->stream('reg-1') as $response) { + $this->fail('should not yield'); + } + $this->fail('expected a NewsdataWebSocketError'); + } catch (NewsdataWebSocketAuthError $e) { + $this->fail('a plain connection error should not be an auth error'); + } catch (NewsdataWebSocketError $e) { + $this->assertStringContainsString('connection refused', $e->getMessage()); + } + } + + public function testReconnectsAfterTransientDrop(): void + { + $ws = $this->ws(function (int $n) { + if ($n === 1) { + return [new \RuntimeException('connection reset')]; // transient + } + return [self::articleFrame('a1', 'after-reconnect')]; + }); + + $titles = []; + foreach ($ws->stream('reg-1') as $response) { + $titles[] = $response->results[0]->title; + break; + } + + $this->assertSame(['after-reconnect'], $titles); + $this->assertGreaterThanOrEqual(2, count($this->connections)); + } + + public function testStreamRejectsEmptyRegistrationId(): void + { + $ws = $this->ws(fn (int $n) => []); + + $this->expectException(NewsdataValidationError::class); + foreach ($ws->stream('') as $response) { + $this->fail('should not yield'); + } + } + + public function testBreakingOutOfTheLoopClosesTheConnection(): void + { + $ws = $this->ws(fn (int $n) => [ + self::articleFrame('a1', 'one'), + self::articleFrame('a2', 'two'), + ], ['reconnect' => false]); + + foreach ($ws->stream('reg-1') as $response) { + break; + } + + $this->assertTrue($this->connections[0]->closed); + } + + // ---- query management ------------------------------------------------- + + public function testRegisterInjectsNewsTypeIntoFilters(): void + { + // websocket_register accepts news_type; validation proves it is set. + $filters = Constants::FILTERS['websocket_register']; + $this->assertContains('news_type', $filters); + $this->assertContains('q', $filters); + // No date or paging filters on a registered query. + $this->assertNotContains('from_date', $filters); + $this->assertNotContains('page', $filters); + $this->assertNotContains('size', $filters); + } + + public function testWebsocketEndpointsUseTheRightHttpMethods(): void + { + $this->assertSame('POST', Constants::ENDPOINT_METHODS['websocket_register']); + $this->assertSame('DELETE', Constants::ENDPOINT_METHODS['websocket_delete']); + // fetch is absent, so it falls through to GET. + $this->assertArrayNotHasKey('websocket_fetch', Constants::ENDPOINT_METHODS); + } + + public function testWebsocketEndpointPathsAreRegistered(): void + { + $this->assertSame('websocket/register', Constants::ENDPOINTS['websocket_register']); + $this->assertSame('websocket/fetch', Constants::ENDPOINTS['websocket_fetch']); + $this->assertSame('websocket/delete', Constants::ENDPOINTS['websocket_delete']); + } + + public function testDeleteRejectsEmptyRegistrationId(): void + { + $this->expectException(NewsdataValidationError::class); + $this->api()->get_websocket_delete(''); + } +}