diff --git a/README.md b/README.md index 5356400..f7f8e26 100644 --- a/README.md +++ b/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`. 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` (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 diff --git a/src/main/java/io/newsdata/api/Constants.java b/src/main/java/io/newsdata/api/Constants.java index 87b7988..99d266b 100644 --- a/src/main/java/io/newsdata/api/Constants.java +++ b/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 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 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 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 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 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.of()), + Map.entry("websocket_delete", Set.of("registration_id")) ); } diff --git a/src/main/java/io/newsdata/api/Endpoint.java b/src/main/java/io/newsdata/api/Endpoint.java index a0263db..45d50c0 100644 --- a/src/main/java/io/newsdata/api/Endpoint.java +++ b/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; diff --git a/src/main/java/io/newsdata/api/NewsDataApiClient.java b/src/main/java/io/newsdata/api/NewsDataApiClient.java index 287347b..216513d 100644 --- a/src/main/java/io/newsdata/api/NewsDataApiClient.java +++ b/src/main/java/io/newsdata/api/NewsDataApiClient.java @@ -141,6 +141,55 @@ public NewsdataResponse marketCount(Map params) { return request(Endpoint.MARKET_COUNT.key(), params); } + // ---- real-time query management -------------------------------------- + + /** + * Register a real-time WebSocket query. POST /1/websocket/register. + * + *

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}. + * + *

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 params) { + Map 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 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 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 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), diff --git a/src/main/java/io/newsdata/api/NewsDataApiWebSocket.java b/src/main/java/io/newsdata/api/NewsDataApiWebSocket.java new file mode 100644 index 0000000..32fbc80 --- /dev/null +++ b/src/main/java/io/newsdata/api/NewsDataApiWebSocket.java @@ -0,0 +1,427 @@ +package io.newsdata.api; + +import java.io.IOException; +import java.net.URI; +import java.net.URLEncoder; +import java.net.http.HttpClient; +import java.net.http.WebSocket; +import java.net.http.WebSocketHandshakeException; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Predicate; + +import com.fasterxml.jackson.databind.JsonNode; + +import io.newsdata.api.exception.NewsdataValidationException; +import io.newsdata.api.exception.NewsdataWebSocketAuthException; +import io.newsdata.api.exception.NewsdataWebSocketException; + +/** + * NewsData.io real-time WebSocket service. + * + *

Registers, lists, and deletes the account's real-time queries and streams + * the responses for a registered query. The management calls go through the + * wrapped {@link NewsDataApiClient}: + * + * 

{@code
+ * NewsDataApiClient client = NewsDataApiClient.builder(apiKey).build();
+ * try (NewsDataApiWebSocket ws = new NewsDataApiWebSocket(client)) {
+ *     NewsdataResponse registered = ws.websocketRegister(Params.of().with("q", "bitcoin"));
+ *     String id = registered.results().path("registration_id").asText();
+ *
+ *     ws.stream(id, response -> {
+ *         for (Article a : response.articles(client.objectMapper())) {
+ *             System.out.println(a.title());
+ *         }
+ *         return true;   // keep streaming; return false to stop
+ *     });
+ * }
+ * }
+ * + *

Transient drops (network errors, server restarts, abnormal closes) are + * reconnected automatically with a capped exponential backoff; build with + * {@code reconnect(false)} to stop on the first disconnect. A permanent + * rejection always throws {@link NewsdataWebSocketAuthException} and is never + * retried. + * + *

Closing the instance (or leaving the try-with-resources block) stops an + * in-flight {@link #stream}. + * + *

A single {@code stream} call must not be shared between threads. + */ +public final class NewsDataApiWebSocket implements AutoCloseable { + + /** Sentinel queued when the connection closes normally. */ + private static final String NORMAL_CLOSE = new String("__normal_close__"); + + private final NewsDataApiClient client; + private final String baseUrl; + private final boolean reconnect; + private final Duration reconnectDelay; + private final Duration reconnectDelayMax; + private final Duration handshakeTimeout; + private final Map headers; + + private final AtomicBoolean closed = new AtomicBoolean(false); + private volatile WebSocket active; + + /** Construct with the defaults; equivalent to {@code builder(client).build()}. */ + public NewsDataApiWebSocket(NewsDataApiClient client) { + this(builder(client)); + } + + private NewsDataApiWebSocket(Builder b) { + if (b.client == null) { + throw new NewsdataValidationException("client is required", "client"); + } + this.client = b.client; + this.baseUrl = b.baseUrl; + this.reconnect = b.reconnect; + this.reconnectDelay = b.reconnectDelay; + this.reconnectDelayMax = b.reconnectDelayMax; + this.handshakeTimeout = b.handshakeTimeout; + this.headers = Map.copyOf(b.headers); + } + + /** A builder for the connection options. */ + public static Builder builder(NewsDataApiClient client) { + return new Builder(client); + } + + /** Fluent options for {@link NewsDataApiWebSocket}. */ + public static final class Builder { + private final NewsDataApiClient client; + private String baseUrl = Constants.WS_BASE_URL; + private boolean reconnect = true; + private Duration reconnectDelay = Constants.WS_RECONNECT_DELAY; + private Duration reconnectDelayMax = Constants.WS_RECONNECT_DELAY_MAX; + private Duration handshakeTimeout = Constants.WS_HANDSHAKE_TIMEOUT; + private Map headers = Map.of(); + + private Builder(NewsDataApiClient client) { + this.client = client; + } + + /** WebSocket endpoint; override for staging / self-hosted / proxied. */ + public Builder baseUrl(String v) { this.baseUrl = v; return this; } + + /** Reconnect on transient drops. Default {@code true}. */ + public Builder reconnect(boolean v) { this.reconnect = v; return this; } + + /** Wait before the first reconnect; doubles after each failure. */ + public Builder reconnectDelay(Duration v) { this.reconnectDelay = v; return this; } + + /** Upper bound on the reconnect delay. */ + public Builder reconnectDelayMax(Duration v) { this.reconnectDelayMax = v; return this; } + + /** Bound on the opening handshake. */ + public Builder handshakeTimeout(Duration v) { this.handshakeTimeout = v; return this; } + + /** Extra HTTP headers for the opening handshake. */ + public Builder headers(Map v) { this.headers = v; return this; } + + public NewsDataApiWebSocket build() { return new NewsDataApiWebSocket(this); } + } + + // ---- query management ------------------------------------------------- + + /** Register a real-time query. See {@link NewsDataApiClient#websocketRegister}. */ + public NewsdataResponse websocketRegister(Map params) { + return client.websocketRegister(params); + } + + /** List registered queries. See {@link NewsDataApiClient#websocketFetch}. */ + public NewsdataResponse websocketFetch() { + return client.websocketFetch(); + } + + /** Delete a registered query. See {@link NewsDataApiClient#websocketDelete}. */ + public NewsdataResponse websocketDelete(String registrationId) { + return client.websocketDelete(registrationId); + } + + // ---- streaming -------------------------------------------------------- + + private String url(String registrationId) { + return baseUrl + + "?apikey=" + URLEncoder.encode(client.apiKey(), StandardCharsets.UTF_8) + + "®istration_id=" + URLEncoder.encode(registrationId, StandardCharsets.UTF_8); + } + + private Duration nextDelay(Duration delay) { + Duration doubled = delay.multipliedBy(2); + return doubled.compareTo(reconnectDelayMax) > 0 ? reconnectDelayMax : doubled; + } + + /** + * Connect and hand each response to {@code handler} as it arrives. + * Responses have the familiar {@code status} / {@code totalResults} / + * {@code results} shape. + * + *

Blocks until the handler returns {@code false}, the instance is + * closed, the calling thread is interrupted, or — with reconnect + * disabled — the connection drops. + * + * @param registrationId the registered query to stream + * @param handler returns {@code true} to keep streaming, + * {@code false} to stop + */ + public void stream(String registrationId, Predicate handler) { + if (registrationId == null || registrationId.isEmpty()) { + throw new NewsdataValidationException( + "registrationId must be a non-empty string", "registration_id"); + } + if (handler == null) { + throw new NewsdataValidationException("handler must not be null", "handler"); + } + + String url = url(registrationId); + String logUrl = NewsDataApiClient.redactApiKey(url); + Duration delay = reconnectDelay; + closed.set(false); + + try { + while (!closed.get()) { + boolean closedNormally; + try { + closedNormally = runOnce(url, logUrl, handler); + } catch (StopStream stop) { + return; + } + if (closed.get()) return; + if (closedNormally && !reconnect) return; + if (!reconnect) return; + + // Transient failure, or a normal close with reconnect enabled: + // wait (capped exponential backoff) and reconnect. + try { + Thread.sleep(delay.toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + delay = nextDelay(delay); + } + } finally { + closeActive(); + } + } + + /** + * Hold a single connection open until it drops or the handler stops it. + * + * @return {@code true} when the connection closed normally + * @throws StopStream when the handler asked to stop + */ + private boolean runOnce(String url, String logUrl, Predicate handler) { + LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); + WebSocket socket; + try { + HttpClient http = client.httpClient(); + WebSocket.Builder wsBuilder = http.newWebSocketBuilder() + .connectTimeout(handshakeTimeout); + headers.forEach(wsBuilder::header); + socket = wsBuilder.buildAsync(URI.create(url), new Listener(queue)) + .get(handshakeTimeout.toMillis() + 1_000, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new StopStream(); + } catch (Exception e) { + handleFailure(unwrap(e), logUrl); + return false; + } + + active = socket; + client.logFromWebSocket("info", "connected to " + logUrl); + socket.request(1); + + while (true) { + Object item; + try { + item = queue.take(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new StopStream(); + } + if (closed.get()) throw new StopStream(); + + if (item == NORMAL_CLOSE) { + return true; + } + if (item instanceof Throwable) { + handleFailure((Throwable) item, logUrl); + return false; + } + + NewsdataResponse response = parse((String) item); + if (response == null) continue; // skip malformed frames + if (!handler.test(response)) { + throw new StopStream(); + } + } + } + + /** Parse one frame, returning null when it isn't a JSON object. */ + private NewsdataResponse parse(String message) { + try { + JsonNode node = client.objectMapper().readTree(message); + if (node == null || !node.isObject()) return null; + return new NewsdataResponse( + node.path("status").asText(null), + node.path("totalResults").asInt(0), + node.get("results"), + node.hasNonNull("nextPage") ? node.get("nextPage").asText() : null, + null); + } catch (IOException e) { + return null; + } + } + + /** + * Classify a connection failure. + * + *

The server always accepts the handshake and then closes with code + * 1008 on a permanent failure — {@code invalid credentials or registration + * not found}, {@code api limit reached}, or {@code device limit reached}. + * Those always throw. Every other close code, including 1013 + * ({@code send timeout} — this client read too slowly), is transient: it + * throws only when reconnect is disabled, otherwise it is logged so the + * caller backs off and retries. The handshake-status check is defensive, + * for proxies in front of the documented server. + */ + private void handleFailure(Throwable err, String logUrl) { + NewsdataWebSocketAuthException auth = permanentAuthError(err); + if (auth != null) throw auth; + if (!reconnect) throw transientError(err); + client.logFromWebSocket("warn", + "connection to " + logUrl + " failed (" + err + "); reconnecting"); + } + + /** + * The auth error to throw if {@code err} is permanent, else null. Close + * code 1008 is the documented permanent signal; the handshake check is + * defensive. + */ + private static NewsdataWebSocketAuthException permanentAuthError(Throwable err) { + if (err instanceof WebSocketHandshakeException) { + int status = ((WebSocketHandshakeException) err).getResponse().statusCode(); + if (status == 401 || status == 403) { + return new NewsdataWebSocketAuthException("connection rejected", err); + } + return null; + } + if (err instanceof PolicyViolation) { + return new NewsdataWebSocketAuthException(err.getMessage(), err); + } + return null; + } + + /** Wrap a transient failure; used only when reconnect is disabled. */ + private static NewsdataWebSocketException transientError(Throwable err) { + if (err instanceof WebSocketHandshakeException) { + int status = ((WebSocketHandshakeException) err).getResponse().statusCode(); + return new NewsdataWebSocketException("handshake failed (HTTP " + status + ")", err); + } + if (err instanceof AbnormalClose) { + return new NewsdataWebSocketException("connection closed", err); + } + return new NewsdataWebSocketException("connection error: " + err, err); + } + + private static Throwable unwrap(Throwable e) { + Throwable t = e; + while ((t instanceof CompletionException || t instanceof java.util.concurrent.ExecutionException) + && t.getCause() != null) { + t = t.getCause(); + } + return t; + } + + private void closeActive() { + WebSocket socket = active; + active = null; + if (socket != null && !socket.isOutputClosed()) { + try { + socket.sendClose(WebSocket.NORMAL_CLOSURE, ""); + } catch (Exception ignored) { + socket.abort(); + } + } + } + + /** Close the active connection, ending any in-flight {@link #stream}. */ + @Override + public void close() { + closed.set(true); + WebSocket socket = active; + if (socket != null) socket.abort(); + active = null; + } + + /** Bridges the callback listener onto the blocking queue the stream reads. */ + private final class Listener implements WebSocket.Listener { + private final LinkedBlockingQueue queue; + private final StringBuilder buffer = new StringBuilder(); + + Listener(LinkedBlockingQueue queue) { + this.queue = queue; + } + + @Override + public CompletionStage onText(WebSocket webSocket, CharSequence data, boolean last) { + buffer.append(data); + if (last) { + queue.offer(buffer.toString()); + buffer.setLength(0); + } + webSocket.request(1); + return null; + } + + @Override + public CompletionStage onClose(WebSocket webSocket, int statusCode, String reason) { + if (statusCode == WebSocket.NORMAL_CLOSURE) { + queue.offer(NORMAL_CLOSE); + } else if (statusCode == Constants.WS_POLICY_VIOLATION) { + queue.offer(new PolicyViolation( + reason == null || reason.isEmpty() ? "connection rejected" : reason)); + } else { + queue.offer(new AbnormalClose("connection closed (" + statusCode + ")")); + } + return CompletableFuture.completedFuture(null); + } + + @Override + public void onError(WebSocket webSocket, Throwable error) { + queue.offer(error); + } + } + + /** Close code 1008 — a permanent rejection. */ + private static final class PolicyViolation extends RuntimeException { + private static final long serialVersionUID = 1L; + + PolicyViolation(String message) { super(message); } + } + + /** Any other non-normal close — transient. */ + private static final class AbnormalClose extends RuntimeException { + private static final long serialVersionUID = 1L; + + AbnormalClose(String message) { super(message); } + } + + /** Internal control-flow marker: the handler (or close()) stopped the stream. */ + private static final class StopStream extends RuntimeException { + private static final long serialVersionUID = 1L; + + StopStream() { super(null, null, false, false); } + } +} diff --git a/src/main/java/io/newsdata/api/exception/NewsdataWebSocketAuthException.java b/src/main/java/io/newsdata/api/exception/NewsdataWebSocketAuthException.java new file mode 100644 index 0000000..86513d1 --- /dev/null +++ b/src/main/java/io/newsdata/api/exception/NewsdataWebSocketAuthException.java @@ -0,0 +1,19 @@ +package io.newsdata.api.exception; + +/** + * The server rejected the WebSocket connection — bad API key, missing + * WebSocket entitlement, unknown {@code registration_id}, device limit + * reached, or exhausted quota. Never retried, regardless of the + * {@code reconnect} setting. + */ +public class NewsdataWebSocketAuthException extends NewsdataWebSocketException { + private static final long serialVersionUID = 1L; + + public NewsdataWebSocketAuthException(String message) { + super(message); + } + + public NewsdataWebSocketAuthException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/src/main/java/io/newsdata/api/exception/NewsdataWebSocketException.java b/src/main/java/io/newsdata/api/exception/NewsdataWebSocketException.java new file mode 100644 index 0000000..8dca957 --- /dev/null +++ b/src/main/java/io/newsdata/api/exception/NewsdataWebSocketException.java @@ -0,0 +1,17 @@ +package io.newsdata.api.exception; + +/** + * A real-time WebSocket stream failure + * (see {@code io.newsdata.api.NewsDataApiWebSocket}). + */ +public class NewsdataWebSocketException extends NewsdataException { + private static final long serialVersionUID = 1L; + + public NewsdataWebSocketException(String message) { + super(message); + } + + public NewsdataWebSocketException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/src/test/java/io/newsdata/api/MockWebSocketServer.java b/src/test/java/io/newsdata/api/MockWebSocketServer.java new file mode 100644 index 0000000..12b5283 --- /dev/null +++ b/src/test/java/io/newsdata/api/MockWebSocketServer.java @@ -0,0 +1,194 @@ +package io.newsdata.api; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Base64; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; + +/** + * A minimal RFC 6455 server for tests — enough to accept the handshake and + * push text / close frames at the client. Server-to-client frames are unmasked, + * which keeps the writer trivial; inbound frames are drained and ignored apart + * from noticing the socket closing. + * + *

Not a general-purpose implementation: no fragmentation, no extensions, no + * payloads over 65535 bytes. + */ +final class MockWebSocketServer implements AutoCloseable { + + private static final String GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; + + private final ServerSocket serverSocket; + private final Thread acceptor; + private final AtomicInteger connections = new AtomicInteger(); + private final List queries = new CopyOnWriteArrayList<>(); + private volatile boolean running = true; + + /** Handshake status to answer with; 101 performs the upgrade. */ + private final int handshakeStatus; + + /** + * @param handshakeStatus 101 to accept, or an error status to reject with + * @param onConnect runs per accepted connection: (session, connectionNumber) + */ + MockWebSocketServer(int handshakeStatus, BiConsumer onConnect) + throws IOException { + this.handshakeStatus = handshakeStatus; + this.serverSocket = new ServerSocket(); + this.serverSocket.bind(new InetSocketAddress("127.0.0.1", 0)); + this.acceptor = new Thread(() -> { + while (running) { + try { + Socket socket = serverSocket.accept(); + int n = connections.incrementAndGet(); + Thread worker = new Thread(() -> serve(socket, onConnect, n)); + worker.setDaemon(true); + worker.start(); + } catch (IOException e) { + return; // socket closed + } + } + }); + this.acceptor.setDaemon(true); + this.acceptor.start(); + } + + String url() { + return "ws://127.0.0.1:" + serverSocket.getLocalPort() + "/ws/event"; + } + + int connectionCount() { + return connections.get(); + } + + /** Query strings seen on each handshake, in order. */ + List queries() { + return new ArrayList<>(queries); + } + + private void serve(Socket socket, BiConsumer onConnect, int n) { + try (Socket s = socket) { + InputStream in = s.getInputStream(); + OutputStream out = s.getOutputStream(); + + BufferedReader reader = new BufferedReader( + new InputStreamReader(in, StandardCharsets.UTF_8)); + String requestLine = reader.readLine(); + if (requestLine != null) { + String[] parts = requestLine.split(" "); + if (parts.length > 1) { + int q = parts[1].indexOf('?'); + queries.add(q >= 0 ? parts[1].substring(q + 1) : ""); + } + } + String key = null; + String line; + while ((line = reader.readLine()) != null && !line.isEmpty()) { + if (line.toLowerCase().startsWith("sec-websocket-key:")) { + key = line.substring(line.indexOf(':') + 1).trim(); + } + } + + if (handshakeStatus != 101) { + byte[] body = "{\"status\":\"error\"}".getBytes(StandardCharsets.UTF_8); + out.write(("HTTP/1.1 " + handshakeStatus + " NO\r\n" + + "Content-Length: " + body.length + "\r\n" + + "Connection: close\r\n\r\n").getBytes(StandardCharsets.UTF_8)); + out.write(body); + out.flush(); + return; + } + + out.write(("HTTP/1.1 101 Switching Protocols\r\n" + + "Upgrade: websocket\r\n" + + "Connection: Upgrade\r\n" + + "Sec-WebSocket-Accept: " + accept(key) + "\r\n\r\n") + .getBytes(StandardCharsets.UTF_8)); + out.flush(); + + onConnect.accept(new Session(out), n); + } catch (IOException ignored) { + // client went away + } + } + + private static String accept(String key) { + try { + MessageDigest sha1 = MessageDigest.getInstance("SHA-1"); + byte[] digest = sha1.digest((key + GUID).getBytes(StandardCharsets.UTF_8)); + return Base64.getEncoder().encodeToString(digest); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } + + @Override + public void close() { + running = false; + try { + serverSocket.close(); + } catch (IOException ignored) { + // already closed + } + } + + /** Write side of one accepted connection. */ + static final class Session { + private final OutputStream out; + + Session(OutputStream out) { + this.out = out; + } + + /** Send one unmasked text frame. */ + void sendText(String payload) throws IOException { + byte[] data = payload.getBytes(StandardCharsets.UTF_8); + synchronized (out) { + out.write(0x81); // FIN + text + if (data.length < 126) { + out.write(data.length); + } else { + out.write(126); + out.write((data.length >>> 8) & 0xFF); + out.write(data.length & 0xFF); + } + out.write(data); + out.flush(); + } + } + + /** Send a close frame with the given code and reason. */ + void sendClose(int code, String reason) throws IOException { + byte[] r = reason.getBytes(StandardCharsets.UTF_8); + synchronized (out) { + out.write(0x88); // FIN + close + out.write(2 + r.length); + out.write((code >>> 8) & 0xFF); + out.write(code & 0xFF); + out.write(r); + out.flush(); + } + } + + /** Sleep, keeping the connection open. */ + void hold(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + } +} diff --git a/src/test/java/io/newsdata/api/NewsDataApiWebSocketTest.java b/src/test/java/io/newsdata/api/NewsDataApiWebSocketTest.java new file mode 100644 index 0000000..db0c94e --- /dev/null +++ b/src/test/java/io/newsdata/api/NewsDataApiWebSocketTest.java @@ -0,0 +1,314 @@ +package io.newsdata.api; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicReference; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpHandler; +import com.sun.net.httpserver.HttpServer; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import io.newsdata.api.exception.NewsdataValidationException; +import io.newsdata.api.exception.NewsdataWebSocketAuthException; +import io.newsdata.api.exception.NewsdataWebSocketException; + +/** Real-time WebSocket tests, against a local RFC 6455 mock. */ +class NewsDataApiWebSocketTest { + + private HttpServer httpServer; + private String baseUrl; + + @BeforeEach + void startHttp() throws IOException { + httpServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + httpServer.start(); + baseUrl = "http://127.0.0.1:" + httpServer.getAddress().getPort() + "/api/1/"; + } + + @AfterEach + void stopHttp() { + if (httpServer != null) httpServer.stop(0); + } + + private void handle(String path, HttpHandler handler) { + httpServer.createContext("/api/1/" + path, handler); + } + + private static void respond(HttpExchange exchange, int status, String body) throws IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(status, bytes.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(bytes); + } + } + + private NewsDataApiClient client() { + return NewsDataApiClient.builder().apiKey("key").baseUrl(baseUrl).build(); + } + + private static String articleFrame(String id, String title) { + return "{\"status\":\"success\",\"totalResults\":1,\"results\":" + + "[{\"article_id\":\"" + id + "\",\"title\":\"" + title + "\"}]}"; + } + + @Test + @Timeout(20) + void streamsResponsesAsTheyArrive() throws Exception { + try (MockWebSocketServer server = new MockWebSocketServer(101, (session, n) -> { + try { + session.sendText(articleFrame("a1", "one")); + session.sendText(articleFrame("a2", "two")); + session.hold(500); + } catch (IOException ignored) { + // client closed + } + })) { + var ws = NewsDataApiWebSocket.builder(client()) + .baseUrl(server.url()).reconnect(false).build(); + + List titles = new CopyOnWriteArrayList<>(); + ws.stream("reg-1", response -> { + titles.add(response.results().get(0).path("title").asText()); + return titles.size() < 2; // stop after the second + }); + + assertEquals(List.of("one", "two"), titles); + } + } + + @Test + @Timeout(20) + void sendsApiKeyAndRegistrationIdInQuery() throws Exception { + try (MockWebSocketServer server = new MockWebSocketServer(101, (session, n) -> { + try { + session.sendText(articleFrame("a1", "one")); + session.hold(300); + } catch (IOException ignored) { + // client closed + } + })) { + var ws = NewsDataApiWebSocket.builder(client()) + .baseUrl(server.url()).reconnect(false).build(); + ws.stream("reg-42", response -> false); + + String query = server.queries().get(0); + assertTrue(query.contains("apikey=key"), "query missing apikey: " + query); + assertTrue(query.contains("registration_id=reg-42"), "query missing id: " + query); + } + } + + @Test + @Timeout(20) + void skipsMalformedFrames() throws Exception { + try (MockWebSocketServer server = new MockWebSocketServer(101, (session, n) -> { + try { + session.sendText("not json at all"); + session.sendText(articleFrame("a1", "one")); + session.hold(300); + } catch (IOException ignored) { + // client closed + } + })) { + var ws = NewsDataApiWebSocket.builder(client()) + .baseUrl(server.url()).reconnect(false).build(); + + List seen = new CopyOnWriteArrayList<>(); + ws.stream("reg-1", response -> { + seen.add(response.results().get(0).path("title").asText()); + return false; + }); + + assertEquals(List.of("one"), seen, "the malformed frame should be skipped"); + } + } + + @Test + @Timeout(20) + void handshake401IsPermanentAndNotRetried() throws Exception { + try (MockWebSocketServer server = new MockWebSocketServer(401, (session, n) -> { })) { + // reconnect stays ON to prove a permanent rejection is not retried. + var ws = NewsDataApiWebSocket.builder(client()) + .baseUrl(server.url()) + .reconnectDelay(Duration.ofMillis(5)) + .build(); + + assertThrows(NewsdataWebSocketAuthException.class, + () -> ws.stream("reg-1", response -> true)); + assertEquals(1, server.connectionCount(), + "a permanent rejection must not retry"); + } + } + + @Test + @Timeout(20) + void policyViolationCloseIsPermanent() throws Exception { + try (MockWebSocketServer server = new MockWebSocketServer(101, (session, n) -> { + try { + session.sendClose(1008, "quota exhausted"); + session.hold(200); + } catch (IOException ignored) { + // client closed + } + })) { + var ws = NewsDataApiWebSocket.builder(client()) + .baseUrl(server.url()) + .reconnectDelay(Duration.ofMillis(5)) + .build(); + + var err = assertThrows(NewsdataWebSocketAuthException.class, + () -> ws.stream("reg-1", response -> true)); + assertTrue(err.getMessage().contains("quota exhausted"), + "should carry the close reason, got: " + err.getMessage()); + assertEquals(1, server.connectionCount()); + } + } + + @Test + @Timeout(20) + void transientHandshakeStopsWhenReconnectDisabled() throws Exception { + try (MockWebSocketServer server = new MockWebSocketServer(500, (session, n) -> { })) { + var ws = NewsDataApiWebSocket.builder(client()) + .baseUrl(server.url()).reconnect(false).build(); + + var err = assertThrows(NewsdataWebSocketException.class, + () -> ws.stream("reg-1", response -> true)); + assertTrue(!(err instanceof NewsdataWebSocketAuthException), + "a 500 handshake is transient, not an auth error"); + } + } + + @Test + @Timeout(30) + void reconnectsAfterTransientDrop() throws Exception { + try (MockWebSocketServer server = new MockWebSocketServer(101, (session, n) -> { + try { + if (n == 1) { + session.sendClose(1011, "server restart"); // transient + return; + } + session.sendText(articleFrame("a1", "after-reconnect")); + session.hold(300); + } catch (IOException ignored) { + // client closed + } + })) { + var ws = NewsDataApiWebSocket.builder(client()) + .baseUrl(server.url()) + .reconnectDelay(Duration.ofMillis(10)) + .reconnectDelayMax(Duration.ofMillis(50)) + .build(); + + AtomicReference got = new AtomicReference<>(); + ws.stream("reg-1", response -> { + got.set(response.results().get(0).path("title").asText()); + return false; + }); + + assertEquals("after-reconnect", got.get()); + assertTrue(server.connectionCount() >= 2, + "should have reconnected, connections=" + server.connectionCount()); + } + } + + @Test + void rejectsEmptyRegistrationId() { + var ws = new NewsDataApiWebSocket(client()); + assertThrows(NewsdataValidationException.class, + () -> ws.stream("", response -> true)); + } + + @Test + void rejectsNullHandler() { + var ws = new NewsDataApiWebSocket(client()); + assertThrows(NewsdataValidationException.class, () -> ws.stream("reg-1", null)); + } + + // ---- query management ------------------------------------------------- + + @Test + void websocketRegisterPostsWithNewsType() { + AtomicReference method = new AtomicReference<>(); + AtomicReference query = new AtomicReference<>(); + handle("websocket/register", exchange -> { + method.set(exchange.getRequestMethod()); + query.set(exchange.getRequestURI().getQuery()); + respond(exchange, 200, + "{\"status\":\"success\",\"results\":{\"registration_id\":\"reg-9\"}}"); + }); + + var response = client().websocketRegister(Params.of().with("q", "bitcoin")); + + assertEquals("POST", method.get()); + assertTrue(query.get().contains("news_type=latest"), query.get()); + assertTrue(query.get().contains("q=bitcoin"), query.get()); + assertEquals("reg-9", response.results().path("registration_id").asText()); + } + + @Test + void websocketRegisterDoesNotMutateCallerParams() { + handle("websocket/register", exchange -> + respond(exchange, 200, "{\"status\":\"success\",\"results\":{}}")); + + var params = Params.of().with("q", "bitcoin"); + client().websocketRegister(params); + + assertTrue(!params.containsKey("news_type"), + "websocketRegister leaked news_type into the caller's params"); + } + + @Test + void websocketFetchUsesGet() { + AtomicReference method = new AtomicReference<>(); + handle("websocket/fetch", exchange -> { + method.set(exchange.getRequestMethod()); + respond(exchange, 200, "{\"status\":\"success\",\"results\":{\"queries\":[]}}"); + }); + + client().websocketFetch(); + assertEquals("GET", method.get()); + } + + @Test + void websocketDeleteUsesDelete() { + AtomicReference method = new AtomicReference<>(); + AtomicReference query = new AtomicReference<>(); + handle("websocket/delete", exchange -> { + method.set(exchange.getRequestMethod()); + query.set(exchange.getRequestURI().getQuery()); + respond(exchange, 200, "{\"status\":\"success\",\"results\":{\"deleted\":true}}"); + }); + + client().websocketDelete("reg-9"); + + assertEquals("DELETE", method.get()); + assertTrue(query.get().contains("registration_id=reg-9"), query.get()); + } + + @Test + void websocketDeleteRejectsEmptyId() { + assertThrows(NewsdataValidationException.class, () -> client().websocketDelete("")); + } + + @Test + void resultlessSuccessStillSucceedsOnWebsocketEndpoints() { + handle("websocket/delete", exchange -> + respond(exchange, 200, "{\"status\":\"success\"}")); + + var response = client().websocketDelete("reg-9"); + assertEquals("success", response.status()); + } +}