Add real-time WebSocket support

README.md

@@ -74,6 +74,9 @@ for (Article article : resp.articles(client.objectMapper())) {
 | `client.count(params)` | `/1/count` | Aggregate counts (requires `from_date`, `to_date`) |
 | `client.cryptoCount(params)` | `/1/crypto/count` | Aggregate crypto counts |
 | `client.marketCount(params)` | `/1/market/count` | Aggregate market counts |
+| `client.websocketRegister(params)` | `/1/websocket/register` | Register a real-time query |
+| `client.websocketFetch()` | `/1/websocket/fetch` | List registered queries |
+| `client.websocketDelete(id)` | `/1/websocket/delete` | Delete a registered query |
 
 `params` is a `Map<String, Object>`. Use the bundled [`Params`](src/main/java/io/newsdata/api/Params.java) helper for fluent construction — `null` values are dropped automatically so you can chain optional fields without `if`-guards. Values may be `String`, `Number`, `Boolean`, or `Collection<String>` (sent comma-joined). Parameter names are case-insensitive — `qInTitle` and `qintitle` are equivalent.
 
@@ -115,6 +118,81 @@ A `NewsdataValidationException` is thrown — before any HTTP request — when:
 
 Booleans for `full_content`, `image`, `video`, and `removeduplicate` are coerced to `"1"`/`"0"`.
 
+## Real-time news (WebSocket)
+
+Register a query first — the returned `registration_id` identifies it from then on:
+
+```java
+NewsdataResponse registered = client.websocketRegister(
+        Params.of().with("q", "bitcoin").with("language", "en"));
+String registrationId = registered.results().path("registration_id").asText();
+```
+
+`websocketRegister` 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
+`NewsdataApiException` with status 409; the existing id is in its response body.
+`websocketFetch()` lists every registered query and `websocketDelete(id)`
+removes one.
+
+Then stream. The handler returns `true` to keep streaming and `false` to stop:
+
+```java
+try (NewsDataApiWebSocket ws = new NewsDataApiWebSocket(client)) {
+    ws.stream(registrationId, response -> {
+        for (Article a : response.articles(client.objectMapper())) {
+            System.out.println(a.title() + " - " + a.link());
+        }
+        return true;
+    });
+}
+```
+
+`stream` blocks until the handler returns `false`, the instance is closed, or
+the calling thread is interrupted. Closing from another thread — including via
+try-with-resources — ends an in-flight stream.
+
+Transient drops (network errors, server restarts, abnormal closes) are
+reconnected automatically with a capped exponential backoff. Build with
+`.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
+`NewsdataWebSocketAuthException` 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:
+
+```java
+try {
+    ws.stream(registrationId, response -> true);
+} catch (NewsdataWebSocketAuthException e) {
+    System.err.println("rejected: " + e.getMessage());
+} catch (NewsdataWebSocketException e) {
+    System.err.println("stream error: " + e.getMessage());
+}
+```
+
+All connection options go through the builder:
+
+```java
+NewsDataApiWebSocket ws = NewsDataApiWebSocket.builder(client)
+        .baseUrl("wss://ws.newsdata.io/ws/event")   // staging / self-hosted / proxied
+        .reconnect(true)                            // default true
+        .reconnectDelay(Duration.ofSeconds(1))      // first delay; doubles each retry
+        .reconnectDelayMax(Duration.ofSeconds(30))  // cap on the delay
+        .handshakeTimeout(Duration.ofSeconds(10))   // opening handshake bound
+        .headers(Map.of("X-Trace", "abc"))          // extra handshake headers
+        .build();
+```
+
 ## Error handling
 
 All SDK exceptions extend `RuntimeException`, so they don't pollute method signatures with `throws` clauses. Catch by type to react to specific failures:
@@ -148,7 +226,9 @@ NewsdataException (extends RuntimeException)
 │   ├── NewsdataAuthException               (401 / 403)
 │   ├── NewsdataRateLimitException          (429; .retryAfter())
 │   └── NewsdataServerException             (5xx)
-└── NewsdataNetworkException                (.getCause())
+├── NewsdataNetworkException                (.getCause())
+└── NewsdataWebSocketException              (real-time stream)
+    └── NewsdataWebSocketAuthException      (policy-violation close 1008)
 ```
 
 ## Configuration

src/main/java/io/newsdata/api/Constants.java

@@ -40,17 +40,51 @@ private Constants() {}
     public static final int SIZE_MAX = 50;
 
     /** Endpoint key → URL path appended to {@link #BASE_URL}. */
-    public static final Map<String, String> ENDPOINT_PATHS = Map.of(
-            "latest", "latest",
-            "crypto", "crypto",
-            "archive", "archive",
-            "sources", "sources",
-            "market", "market",
-            "count", "count",
-            "crypto_count", "crypto/count",
-            "market_count", "market/count"
+    public static final Map<String, String> ENDPOINT_PATHS = Map.ofEntries(
+            Map.entry("latest", "latest"),
+            Map.entry("crypto", "crypto"),
+            Map.entry("archive", "archive"),
+            Map.entry("sources", "sources"),
+            Map.entry("market", "market"),
+            Map.entry("count", "count"),
+            Map.entry("crypto_count", "crypto/count"),
+            Map.entry("market_count", "market/count"),
+            Map.entry("websocket_register", "websocket/register"),
+            Map.entry("websocket_fetch", "websocket/fetch"),
+            Map.entry("websocket_delete", "websocket/delete")
     );
 
+    /** HTTP method per endpoint; anything absent is a {@code GET}. */
+    public static final Map<String, String> ENDPOINT_METHODS = Map.of(
+            "websocket_register", "POST",
+            "websocket_delete", "DELETE"
+    );
+
+    /**
+     * Endpoints whose success envelope may carry no {@code results} field, so
+     * they are exempt from the results-present check applied elsewhere.
+     */
+    public static final Set<String> RESULTS_OPTIONAL =
+            Set.of("websocket_register", "websocket_fetch", "websocket_delete");
+
+    /** Real-time WebSocket endpoint. */
+    public static final String WS_BASE_URL = "wss://ws.newsdata.io/ws/event";
+
+    /** The feed a registered query matches against. */
+    public static final String WS_NEWS_TYPE = "latest";
+
+    /** Close code the server uses for a permanent connection rejection. */
+    public static final int WS_POLICY_VIOLATION = 1008;
+
+    /** Wait before the first reconnect; doubles after each consecutive failure. */
+    public static final Duration WS_RECONNECT_DELAY = Duration.ofSeconds(1);
+
+    /** Upper bound on the reconnect delay. */
+    public static final Duration WS_RECONNECT_DELAY_MAX = Duration.ofSeconds(30);
+
+    /** Bound on the opening handshake. */
+    public static final Duration WS_HANDSHAKE_TIMEOUT = Duration.ofSeconds(10);
+
     /** Endpoints that require both {@code from_date} and {@code to_date}. */
     public static final Set<String> REQUIRES_DATE_RANGE =
             Set.of("count", "crypto_count", "market_count");
@@ -132,6 +166,19 @@ private Constants() {}
                     "excludelanguage", "full_content", "image", "video", "organization",
                     "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.
+            Map.entry("websocket_register", Set.of(
+                    "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"
+            )),
+            Map.entry("websocket_fetch", Set.<String>of()),
+            Map.entry("websocket_delete", Set.of("registration_id"))
     );
 }

src/main/java/io/newsdata/api/Endpoint.java

@@ -23,7 +23,13 @@ public enum Endpoint {
     /** {@code /1/crypto/count} — aggregate crypto counts. */
     CRYPTO_COUNT("crypto_count"),
     /** {@code /1/market/count} — aggregate market counts. */
-    MARKET_COUNT("market_count");
+    MARKET_COUNT("market_count"),
+    /** {@code /1/websocket/register} — register a real-time query. */
+    WEBSOCKET_REGISTER("websocket_register"),
+    /** {@code /1/websocket/fetch} — list registered real-time queries. */
+    WEBSOCKET_FETCH("websocket_fetch"),
+    /** {@code /1/websocket/delete} — delete a registered real-time query. */
+    WEBSOCKET_DELETE("websocket_delete");
 
     private final String key;
 

src/main/java/io/newsdata/api/NewsDataApiClient.java

@@ -141,6 +141,55 @@ public NewsdataResponse marketCount(Map<String, Object> params) {
         return request(Endpoint.MARKET_COUNT.key(), params);
     }
 
+    // ---- real-time query management --------------------------------------
+
+    /**
+     * Register a real-time WebSocket query. POST /1/websocket/register.
+     *
+     * <p>Takes the familiar filter names ({@code q}, {@code country},
+     * {@code language}, {@code domain}, …); no date or paging filters apply,
+     * since a registered query matches news as it is published. The new
+     * query's id is at {@code results.registration_id} — pass it to
+     * {@link NewsDataApiWebSocket#stream}.
+     *
+     * <p>Registering an identical query twice throws
+     * {@link io.newsdata.api.exception.NewsdataApiException} with status 409;
+     * the existing id is in its response body.
+     */
+    public NewsdataResponse websocketRegister(Map<String, Object> params) {
+        Map<String, Object> withType = new LinkedHashMap<>(
+                params == null ? Map.of() : params);
+        withType.put("news_type", Constants.WS_NEWS_TYPE);
+        return request(Endpoint.WEBSOCKET_REGISTER.key(), withType);
+    }
+
+    /**
+     * List the account's registered real-time queries. GET /1/websocket/fetch.
+     * One entry per query at {@code results.queries}.
+     */
+    public NewsdataResponse websocketFetch() {
+        return request(Endpoint.WEBSOCKET_FETCH.key(), Map.of());
+    }
+
+    /** Delete a registered real-time query. DELETE /1/websocket/delete. */
+    public NewsdataResponse websocketDelete(String registrationId) {
+        if (registrationId == null || registrationId.isEmpty()) {
+            throw new NewsdataValidationException(
+                    "registrationId must be a non-empty string", "registration_id");
+        }
+        return request(Endpoint.WEBSOCKET_DELETE.key(),
+                Map.of("registration_id", registrationId));
+    }
+
+    /** The API key, for the WebSocket handshake URL. */
+    String apiKey() { return apiKey; }
+
+    /** The HTTP client, reused for the WebSocket handshake. */
+    HttpClient httpClient() { return httpClient; }
+
+    /** Forward a log line from the WebSocket layer. */
+    void logFromWebSocket(String level, String message) { log(level, message); }
+
     // ---- pagination ------------------------------------------------------
 
     /**
@@ -249,17 +298,18 @@ NewsdataResponse request(String endpoint, Map<String, Object> params) {
         String path = Constants.ENDPOINT_PATHS.get(endpoint);
         String url = baseUrl + path + "?" + buildQuery(encoded);
         String logUrl = redactApiKey(url);
+        String method = Constants.ENDPOINT_METHODS.getOrDefault(endpoint, "GET");
 
         Exception last = null;
         for (int attempt = 1; attempt <= maxRetries; attempt++) {
-            log("info", "GET " + logUrl + " (attempt " + attempt + "/" + maxRetries + ")");
+            log("info", method + " " + logUrl + " (attempt " + attempt + "/" + maxRetries + ")");
             HttpResponse<String> resp;
             try {
                 HttpRequest req = HttpRequest.newBuilder()
                         .uri(URI.create(url))
                         .timeout(timeout)
                         .header("Accept", "application/json")
-                        .GET()
+                        .method(method, HttpRequest.BodyPublishers.noBody())
                         .build();
                 resp = httpClient.send(req, BodyHandlers.ofString());
             } catch (IOException e) {
@@ -290,8 +340,10 @@ NewsdataResponse request(String endpoint, Map<String, Object> params) {
                         "non-JSON response from API (status " + status + ")", status, body);
             }
 
+            boolean hasResults = parsed != null
+                    && !parsed.path("results").isNull() && !parsed.path("results").isMissingNode();
             if (status == 200 && parsed != null && parsed.path("status").asText("").equals("success")
-                    && !parsed.path("results").isNull() && !parsed.path("results").isMissingNode()) {
+                    && (hasResults || Constants.RESULTS_OPTIONAL.contains(endpoint))) {
                 return new NewsdataResponse(
                         "success",
                         parsed.path("totalResults").asInt(0),