Add real-time WebSocket support

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

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": {

examples/websocket.php

@@ -0,0 +1,78 @@
+<?php
+
+/**
+ * Real-time news streaming.
+ *
+ *   NEWSDATA_API_KEY=<your key> 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);
+}

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'],
     ];
 }

src/Exception/NewsdataWebSocketAuthError.php

@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace NewsdataIO\Exception;
+
+/**
+ * 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.
+ */
+class NewsdataWebSocketAuthError extends NewsdataWebSocketError
+{
+}

src/Exception/NewsdataWebSocketError.php

@@ -0,0 +1,14 @@
+<?php
+
+declare(strict_types=1);
+
+namespace NewsdataIO\Exception;
+
+/**
+ * A real-time WebSocket stream failure.
+ *
+ * @see \NewsdataIO\NewsdataWebSocket
+ */
+class NewsdataWebSocketError extends NewsdataException
+{
+}