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(''); + } +}