diff --git a/.fern/metadata.json b/.fern/metadata.json
index 104acf4..2dc8f1c 100644
--- a/.fern/metadata.json
+++ b/.fern/metadata.json
@@ -1,16 +1,16 @@
{
- "cliVersion": "5.143.0",
+ "cliVersion": "5.170.5",
"generatorName": "fernapi/fern-java-sdk",
- "generatorVersion": "4.23.1",
+ "generatorVersion": "4.32.4",
"generatorConfig": {
"publish-to": "central",
"client-class-name": "CatchAllApi",
"package-prefix": "com.newscatcher.catchall"
},
- "originGitCommit": "6ef7448ab601e5b372caec45717b0390a4150d2c",
+ "originGitCommit": "b7764dca3eb740a480b120edb87932f928b9210e",
"originGitCommitIsDirty": true,
"invokedBy": "ci",
"requestedVersion": "AUTO",
"ciProvider": "unknown",
- "sdkVersion": "3.0.5"
+ "sdkVersion": "4.0.0"
}
\ No newline at end of file
diff --git a/.fern/replay.lock b/.fern/replay.lock
index 39d42b2..c909f6c 100644
--- a/.fern/replay.lock
+++ b/.fern/replay.lock
@@ -18,5 +18,11 @@ generations:
cli_version: unknown
generator_versions:
fernapi/fern-java-sdk: 4.23.1
-current_generation: b2be01153a63a951ba3db45c87ebd0d44d6b696d
+ - commit_sha: 799bb58627612f83a19eb44e11e4552d5186d6f0
+ tree_hash: c186d73dcee3e42a9f475589faf75c5260d64f30
+ timestamp: 2026-10-22T15:37:56.670Z
+ cli_version: unknown
+ generator_versions:
+ fernapi/fern-java-sdk: 4.32.4
+current_generation: 799bb58627612f83a19eb44e11e4552d5186d6f0
patches: []
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 1f32646..7b513f7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -62,7 +62,7 @@ jobs:
env:
MAVEN_USERNAME: ${{ secrets.MAVEN_USERNAME }}
MAVEN_PASSWORD: ${{ secrets.MAVEN_PASSWORD }}
- MAVEN_PUBLISH_REGISTRY_URL: "https://oss.sonatype.org/service/local/staging/deploy/maven2/"
+ MAVEN_PUBLISH_REGISTRY_URL: "https://ossrh-staging-api.central.sonatype.com/service/local/staging/deploy/maven2/"
MAVEN_SIGNATURE_KID: ${{ secrets.MAVEN_SIGNATURE_KID }}
MAVEN_SIGNATURE_SECRET_KEY: ${{ secrets.MAVEN_SIGNATURE_SECRET_KEY }}
MAVEN_SIGNATURE_PASSWORD: ${{ secrets.MAVEN_SIGNATURE_PASSWORD }}
diff --git a/README.md b/README.md
index a90bbfd..ade4769 100644
--- a/README.md
+++ b/README.md
@@ -19,6 +19,7 @@ The Newscatcher Java library provides convenient access to the Newscatcher APIs
- [Retries](#retries)
- [Timeouts](#timeouts)
- [Custom Headers](#custom-headers)
+ - [Additional Body Properties](#additional-body-properties)
- [Access Raw Response Data](#access-raw-response-data)
- [Contributing](#contributing)
- [Support](#support)
@@ -35,7 +36,7 @@ Add the dependency in your `build.gradle` file:
```groovy
dependencies {
- implementation 'com.newscatcherapi:newscatcher-catchall-sdk:3.0.5'
+ implementation 'com.newscatcherapi:newscatcher-catchall-sdk:4.0.0'
}
```
@@ -47,7 +48,7 @@ Add the dependency in your `pom.xml` file:
com.newscatcherapi
newscatcher-catchall-sdk
- 3.0.5
+ 4.0.0
```
@@ -99,7 +100,7 @@ import com.newscatcher.catchall.core.Environment;
CatchAllApi client = CatchAllApi
.builder()
- .environment(Environment.Default)
+ .environment(Environment.DEFAULT)
.build();
```
@@ -231,6 +232,25 @@ client.jobs().createJob(
);
```
+### Additional Body Properties
+
+If you need to send a request body property that is not yet part of the SDK (for example, a beta field),
+use the `addBodyProperty` request option. Properties are sent with the key exactly as given and override
+any property with the same key; for endpoints without a request body, they are sent as a JSON body.
+This is supported for JSON and form-urlencoded request bodies, but not for multipart (file upload) requests.
+
+```java
+import com.newscatcher.catchall.core.RequestOptions;
+
+client.jobs().createJob(
+ ...,
+ RequestOptions
+ .builder()
+ .addBodyProperty("extra_field", "extra-value")
+ .build()
+);
+```
+
### Access Raw Response Data
The SDK provides access to raw response data, including headers, through the `withRawResponse()` method.
@@ -268,3 +288,4 @@ On the other hand, contributions to the README are always very welcome!
+
diff --git a/build.gradle b/build.gradle
index 20f4a1b..e3ff372 100644
--- a/build.gradle
+++ b/build.gradle
@@ -57,7 +57,7 @@ java {
group = 'com.newscatcherapi'
-version = '3.0.5'
+version = '4.0.0'
jar {
dependsOn(":generatePomFileForMavenPublication")
@@ -88,7 +88,7 @@ publishing {
maven(MavenPublication) {
groupId = 'com.newscatcherapi'
artifactId = 'newscatcher-catchall-sdk'
- version = '3.0.5'
+ version = '4.0.0'
from components.java
pom {
name = 'newscatcher'
diff --git a/changelog.md b/changelog.md
index 7d99624..334fd90 100644
--- a/changelog.md
+++ b/changelog.md
@@ -1,3 +1,21 @@
+## [4.0.0] - 2026-10-22
+### Breaking Changes
+- **`DatasetsClient.createDatasetFromCsv()`** — all InputStream overloads now require a `CreateDatasetFromCsvRequest` as the first argument; replace `createDatasetFromCsv(stream, filename, ...)` with `createDatasetFromCsv(request, stream, filename, ...)`.
+- **`AsyncDatasetsClient.createDatasetFromCsv()`** — same signature change as the synchronous client; prepend a `CreateDatasetFromCsvRequest` argument before `stream` and `filename` at every call site.
+- **`RawDatasetsClient.createDatasetFromCsv()`** — all four overloads now require a leading `CreateDatasetFromCsvRequest` parameter so that `name` and `description` are sent as multipart form fields; add a `CreateDatasetFromCsvRequest` as the first argument to every call site.
+- **`AsyncRawDatasetsClient.createDatasetFromCsv()`** — all four overloads now require a leading `CreateDatasetFromCsvRequest` parameter; update every call site to prepend a `CreateDatasetFromCsvRequest` argument.
+
+### Added
+- **`RequestOptions.Builder.addBodyProperty()` / `bodyProperties()`** — new builder methods that inject arbitrary key/value pairs into the JSON or form-urlencoded request body, overriding any existing property with the same key.
+- **`RetryInterceptor.newAsyncCall()`** — new static factory that creates a non-blocking async call with built-in retry support, preventing OkHttp dispatcher slot exhaustion and enabling immediate cancellation.
+- **`Stream.fromSseWithEventDiscrimination()`** — new static factory for SSE streams that supports envelope-level event discrimination, deserializing specified event names from the full SSE envelope rather than just the `data` payload.
+- **`PathSegments.validate()`** — new core utility that rejects `"."` and `".."` as path parameter values, preventing silent URL path traversal.
+
+### Changed
+- **`AsyncDatasetsClient`** — all async methods now propagate `CompletableFuture` cancellation to the underlying HTTP call, ensuring in-flight requests are properly cancelled.
+- **Request builder classes** — `@JsonAnySetter` now filters out known property keys on all request types (e.g., `CreateDatasetRequest`, `UpdateDatasetRequest`, `ListDatasetsRequest`, and others), preventing typed fields from being silently overwritten during deserialization.
+- **`RawDatasetsClient`** — path segment values for `datasetId` are now validated via `PathSegments.validate()` and extra body properties from `RequestOptions` are merged into every JSON request body.
+
## [3.0.5] - 2026-09-30
## [3.0.4] - 2026-08-04
diff --git a/src/main/java/com/newscatcher/catchall/core/BodyProperties.java b/src/main/java/com/newscatcher/catchall/core/BodyProperties.java
new file mode 100644
index 0000000..50ee2eb
--- /dev/null
+++ b/src/main/java/com/newscatcher/catchall/core/BodyProperties.java
@@ -0,0 +1,89 @@
+/**
+ * This file was auto-generated by Fern from our API Definition.
+ */
+package com.newscatcher.catchall.core;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import okhttp3.RequestBody;
+
+/**
+ * Merges the additional body properties set through {@code RequestOptions.Builder#addBodyProperty} into request
+ * bodies. Properties are sent with the keys exactly as given (no casing transform) and override any generated body
+ * property with the same key.
+ *
+ *
Supported for JSON and {@code application/x-www-form-urlencoded} bodies. Multipart (file upload) and raw bytes
+ * bodies are sent unchanged.
+ */
+public final class BodyProperties {
+
+ private BodyProperties() {}
+
+ public static boolean isPresent(Map bodyProperties) {
+ return bodyProperties != null && !bodyProperties.isEmpty();
+ }
+
+ /**
+ * Returns the value to serialize as the JSON request body. Without body properties the body is returned unchanged.
+ * Otherwise the body is converted to a JSON object (an empty one when the body is null or absent) and the body
+ * properties are set on it, overriding existing keys. Bodies that are not JSON objects (arrays, primitives) are
+ * returned unchanged.
+ */
+ public static Object merge(Object body, Map bodyProperties) throws JsonProcessingException {
+ if (!isPresent(bodyProperties)) {
+ return body;
+ }
+ JsonNode node;
+ try {
+ node = ObjectMappers.JSON_MAPPER.valueToTree(body);
+ } catch (IllegalArgumentException e) {
+ throw JsonMappingException.from((com.fasterxml.jackson.core.JsonParser) null, e.getMessage(), e);
+ }
+ ObjectNode merged;
+ if (node == null || node.isNull() || node.isMissingNode()) {
+ merged = ObjectMappers.JSON_MAPPER.createObjectNode();
+ } else if (node.isObject()) {
+ merged = (ObjectNode) node;
+ } else {
+ return body;
+ }
+ for (Map.Entry entry : bodyProperties.entrySet()) {
+ merged.putPOJO(entry.getKey(), entry.getValue());
+ }
+ return merged;
+ }
+
+ /**
+ * Returns a JSON request body made of the body properties for an endpoint without a request body, or the fallback
+ * when there are no body properties.
+ */
+ public static RequestBody toRequestBody(Map bodyProperties, RequestBody fallback) {
+ if (!isPresent(bodyProperties)) {
+ return fallback;
+ }
+ try {
+ return RequestBody.create(
+ ObjectMappers.JSON_MAPPER.writeValueAsBytes(merge(null, bodyProperties)),
+ MediaTypes.APPLICATION_JSON);
+ } catch (JsonProcessingException e) {
+ throw new IllegalArgumentException("Failed to serialize body properties", e);
+ }
+ }
+
+ /** Returns the form parameters with the body properties added, overriding existing keys. */
+ public static Map mergeFormParams(
+ Map formParams, Map bodyProperties) {
+ Map merged = new LinkedHashMap<>();
+ if (formParams != null) {
+ merged.putAll(formParams);
+ }
+ if (isPresent(bodyProperties)) {
+ merged.putAll(bodyProperties);
+ }
+ return merged;
+ }
+}
diff --git a/src/main/java/com/newscatcher/catchall/core/ClientOptions.java b/src/main/java/com/newscatcher/catchall/core/ClientOptions.java
index a831b43..a3a1655 100644
--- a/src/main/java/com/newscatcher/catchall/core/ClientOptions.java
+++ b/src/main/java/com/newscatcher/catchall/core/ClientOptions.java
@@ -60,10 +60,10 @@ private ClientOptions(
this.headers.putAll(headers);
this.headers.putAll(new HashMap() {
{
- put("User-Agent", "com.newscatcherapi.newscatcher-catchall-sdk/3.0.5");
+ put("User-Agent", "com.newscatcherapi.newscatcher-catchall-sdk/4.0.0");
put("X-Fern-Language", "JAVA");
put("X-Fern-SDK-Name", "com.newscatcherapi:newscatcher-catchall-sdk");
- put("X-Fern-SDK-Version", "3.0.5");
+ put("X-Fern-SDK-Version", "4.0.0");
}
});
this.headerSuppliers = headerSuppliers;
diff --git a/src/main/java/com/newscatcher/catchall/core/LogLevel.java b/src/main/java/com/newscatcher/catchall/core/LogLevel.java
index 488b552..1cdc38a 100644
--- a/src/main/java/com/newscatcher/catchall/core/LogLevel.java
+++ b/src/main/java/com/newscatcher/catchall/core/LogLevel.java
@@ -3,6 +3,8 @@
*/
package com.newscatcher.catchall.core;
+import java.util.Locale;
+
/**
* Log levels for SDK logging configuration.
* Silent by default — no log output unless explicitly configured.
@@ -31,6 +33,6 @@ public int getValue() {
* @throws IllegalArgumentException if the string does not match any level
*/
public static LogLevel fromString(String level) {
- return LogLevel.valueOf(level.toUpperCase());
+ return LogLevel.valueOf(level.toUpperCase(Locale.ROOT));
}
}
diff --git a/src/main/java/com/newscatcher/catchall/core/LoggingInterceptor.java b/src/main/java/com/newscatcher/catchall/core/LoggingInterceptor.java
index 4cf746e..eef921a 100644
--- a/src/main/java/com/newscatcher/catchall/core/LoggingInterceptor.java
+++ b/src/main/java/com/newscatcher/catchall/core/LoggingInterceptor.java
@@ -6,6 +6,7 @@
import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
+import java.util.Locale;
import java.util.Set;
import okhttp3.Interceptor;
import okhttp3.Request;
@@ -58,7 +59,7 @@ public Response intercept(Chain chain) throws IOException {
sb.append(", ");
}
sb.append(name).append("=");
- if (SENSITIVE_HEADERS.contains(name.toLowerCase())) {
+ if (SENSITIVE_HEADERS.contains(name.toLowerCase(Locale.ROOT))) {
sb.append("[REDACTED]");
} else {
sb.append(request.header(name));
@@ -83,7 +84,7 @@ public Response intercept(Chain chain) throws IOException {
sb.append(", ");
}
sb.append(name).append("=");
- if (SENSITIVE_HEADERS.contains(name.toLowerCase())) {
+ if (SENSITIVE_HEADERS.contains(name.toLowerCase(Locale.ROOT))) {
sb.append("[REDACTED]");
} else {
sb.append(response.header(name));
diff --git a/src/main/java/com/newscatcher/catchall/core/PathSegments.java b/src/main/java/com/newscatcher/catchall/core/PathSegments.java
new file mode 100644
index 0000000..01726ea
--- /dev/null
+++ b/src/main/java/com/newscatcher/catchall/core/PathSegments.java
@@ -0,0 +1,21 @@
+/**
+ * This file was auto-generated by Fern from our API Definition.
+ */
+package com.newscatcher.catchall.core;
+
+public final class PathSegments {
+
+ private PathSegments() {}
+
+ /**
+ * Returns the value unchanged, rejecting "." and "..". These are dot-segments that URL builders resolve,
+ * which would otherwise change the endpoint a request is sent to.
+ */
+ public static String validate(String value) {
+ if (".".equals(value) || "..".equals(value)) {
+ throw new IllegalArgumentException(
+ "Invalid path parameter value \"" + value + "\": \".\" and \"..\" are not allowed.");
+ }
+ return value;
+ }
+}
diff --git a/src/main/java/com/newscatcher/catchall/core/RequestOptions.java b/src/main/java/com/newscatcher/catchall/core/RequestOptions.java
index a556072..1e0dedb 100644
--- a/src/main/java/com/newscatcher/catchall/core/RequestOptions.java
+++ b/src/main/java/com/newscatcher/catchall/core/RequestOptions.java
@@ -4,6 +4,7 @@
package com.newscatcher.catchall.core;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
@@ -26,6 +27,8 @@ public final class RequestOptions {
private final Map> queryParameterSuppliers;
+ private final Map bodyProperties;
+
private RequestOptions(
String apiKey,
Optional timeout,
@@ -34,7 +37,8 @@ private RequestOptions(
Map headers,
Map> headerSuppliers,
Map queryParameters,
- Map> queryParameterSuppliers) {
+ Map> queryParameterSuppliers,
+ Map bodyProperties) {
this.apiKey = apiKey;
this.timeout = timeout;
this.timeoutTimeUnit = timeoutTimeUnit;
@@ -43,6 +47,7 @@ private RequestOptions(
this.headerSuppliers = headerSuppliers;
this.queryParameters = queryParameters;
this.queryParameterSuppliers = queryParameterSuppliers;
+ this.bodyProperties = bodyProperties;
}
public Optional getTimeout() {
@@ -77,6 +82,13 @@ public Map getQueryParameters() {
return queryParameters;
}
+ /**
+ * Additional properties merged into the request body, keyed by their wire names.
+ */
+ public Map getBodyProperties() {
+ return new LinkedHashMap<>(this.bodyProperties);
+ }
+
public static Builder builder() {
return new Builder();
}
@@ -98,6 +110,8 @@ public static class Builder {
private final Map> queryParameterSuppliers = new HashMap<>();
+ private final Map bodyProperties = new LinkedHashMap<>();
+
public Builder apiKey(String apiKey) {
this.apiKey = apiKey;
return this;
@@ -139,6 +153,22 @@ public Builder addQueryParameter(String key, Supplier value) {
return this;
}
+ /**
+ * Adds a property to the JSON or form-urlencoded request body, keyed by its wire name. It overrides any request body property with the same key, and is sent as the whole body for endpoints without one. Multipart (file upload) bodies are not supported.
+ */
+ public Builder addBodyProperty(String key, Object value) {
+ this.bodyProperties.put(key, value);
+ return this;
+ }
+
+ /**
+ * Adds each entry as a request body property. See {@link #addBodyProperty}.
+ */
+ public Builder bodyProperties(Map bodyProperties) {
+ this.bodyProperties.putAll(bodyProperties);
+ return this;
+ }
+
public RequestOptions build() {
return new RequestOptions(
apiKey,
@@ -148,7 +178,8 @@ public RequestOptions build() {
headers,
headerSuppliers,
queryParameters,
- queryParameterSuppliers);
+ queryParameterSuppliers,
+ bodyProperties);
}
}
}
diff --git a/src/main/java/com/newscatcher/catchall/core/RetryInterceptor.java b/src/main/java/com/newscatcher/catchall/core/RetryInterceptor.java
index 783accd..bf208f2 100644
--- a/src/main/java/com/newscatcher/catchall/core/RetryInterceptor.java
+++ b/src/main/java/com/newscatcher/catchall/core/RetryInterceptor.java
@@ -4,14 +4,20 @@
package com.newscatcher.catchall.core;
import java.io.IOException;
+import java.io.InterruptedIOException;
import java.time.Duration;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
import java.util.Optional;
import java.util.Random;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import okhttp3.Call;
+import okhttp3.Callback;
import okhttp3.Interceptor;
+import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
@@ -63,6 +69,10 @@ public RetryInterceptor(
@Override
public Response intercept(Chain chain) throws IOException {
Request request = chain.request();
+ if (request.tag(AsyncAttempt.class) != null) {
+ // Retries of calls created by newAsyncCall are scheduled off OkHttp's dispatcher threads.
+ return chain.proceed(request);
+ }
int effectiveMaxRetries = resolveMaxRetries(request);
Response response = chain.proceed(request);
@@ -88,7 +98,11 @@ private Response retryChain(Response response, Chain chain, int maxRetries) thro
Thread.sleep(nextBackoff.get().toMillis());
} catch (InterruptedException e) {
response.close();
- throw new IOException("Interrupted while trying request", e);
+ Thread.currentThread().interrupt();
+ InterruptedIOException interrupted =
+ new InterruptedIOException("Interrupted while waiting to retry request");
+ interrupted.initCause(e);
+ throw interrupted;
} finally {
callTimeout.ifPresent(AsyncTimeout::enter);
}
@@ -114,6 +128,229 @@ private Response retryChain(Response response, Chain chain, int maxRetries) thro
return response;
}
+ /**
+ * Creates a call for the async client. If {@code client} has a {@link RetryInterceptor}, retries are scheduled
+ * after the backoff delay instead of sleeping inside the interceptor, so a call that is waiting to retry does not
+ * hold one of OkHttp's dispatcher slots (5 per host by default) and stops waiting as soon as it is cancelled.
+ */
+ public static AsyncCall newAsyncCall(OkHttpClient client, Request request) {
+ for (Interceptor interceptor : client.interceptors()) {
+ if (interceptor instanceof RetryInterceptor) {
+ RetryInterceptor retryInterceptor = (RetryInterceptor) interceptor;
+ return new AsyncRetryingCall(
+ retryInterceptor, client, request, retryInterceptor.resolveMaxRetries(request));
+ }
+ }
+ return new AsyncRetryingCall(null, client, request, 0);
+ }
+
+ /** An async OkHttp call that retries retryable responses without blocking a dispatcher thread. */
+ public interface AsyncCall {
+ void enqueue(Callback callback);
+
+ void cancel();
+ }
+
+ /** Marks an attempt of an {@link AsyncCall} so that {@link #intercept(Chain)} does not retry it in place. */
+ private static final class AsyncAttempt {
+ private static final AsyncAttempt INSTANCE = new AsyncAttempt();
+ }
+
+ private static final class RetryScheduler {
+ private static final ScheduledThreadPoolExecutor INSTANCE = create();
+
+ private static ScheduledThreadPoolExecutor create() {
+ ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, runnable -> {
+ Thread thread = new Thread(runnable, "RetryInterceptor-scheduler");
+ thread.setDaemon(true);
+ return thread;
+ });
+ executor.setRemoveOnCancelPolicy(true);
+ return executor;
+ }
+ }
+
+ private static final class AsyncRetryingCall implements AsyncCall, Callback {
+ private final OkHttpClient client;
+ private final Request request;
+ private final ExponentialBackoff backoff;
+ private final Object lock = new Object();
+
+ // All fields below are guarded by lock.
+ private Callback callback;
+ private Call currentCall;
+ private ScheduledFuture> pendingRetry;
+ private Response previousResponse;
+ private boolean executed;
+ private boolean canceled;
+ private boolean completed;
+
+ AsyncRetryingCall(RetryInterceptor retryInterceptor, OkHttpClient client, Request request, int maxRetries) {
+ this.client = client;
+ this.request = request.newBuilder()
+ .tag(AsyncAttempt.class, AsyncAttempt.INSTANCE)
+ .build();
+ this.backoff = retryInterceptor != null ? retryInterceptor.new ExponentialBackoff(maxRetries) : null;
+ }
+
+ @Override
+ public void enqueue(Callback callback) {
+ Call call;
+ boolean wasCanceled;
+ synchronized (lock) {
+ if (executed) {
+ throw new IllegalStateException("Already Executed");
+ }
+ executed = true;
+ this.callback = callback;
+ call = client.newCall(request);
+ currentCall = call;
+ wasCanceled = canceled;
+ }
+ if (wasCanceled) {
+ failCanceled();
+ return;
+ }
+ call.enqueue(this);
+ }
+
+ private void startRetry() {
+ Call call;
+ synchronized (lock) {
+ pendingRetry = null;
+ if (completed) {
+ return;
+ }
+ if (canceled) {
+ call = null;
+ } else {
+ call = client.newCall(request);
+ currentCall = call;
+ }
+ }
+ if (call == null) {
+ failCanceled();
+ return;
+ }
+ try {
+ call.enqueue(this);
+ } catch (RuntimeException e) {
+ // Runs on the retry scheduler, where an uncaught exception would leave the callback never invoked.
+ onFailure(call, new IOException("Failed to enqueue retry attempt", e));
+ }
+ }
+
+ @Override
+ public void onResponse(Call call, Response response) throws IOException {
+ if (backoff != null && shouldRetry(response.code())) {
+ Optional delay = backoff.nextBackoff(response);
+ if (delay.isPresent()) {
+ Response buffered = bufferResponse(response);
+ synchronized (lock) {
+ if (!canceled && !completed) {
+ closePreviousResponse();
+ previousResponse = buffered;
+ pendingRetry = RetryScheduler.INSTANCE.schedule(
+ this::startRetry, delay.get().toMillis(), TimeUnit.MILLISECONDS);
+ return;
+ }
+ }
+ buffered.close();
+ failCanceled();
+ return;
+ }
+ }
+ boolean deliver;
+ synchronized (lock) {
+ deliver = !canceled && !completed;
+ if (deliver) {
+ completed = true;
+ closePreviousResponse();
+ }
+ }
+ if (!deliver) {
+ response.close();
+ failCanceled();
+ return;
+ }
+ callback.onResponse(call, response);
+ }
+
+ @Override
+ public void onFailure(Call call, IOException e) {
+ Response previous;
+ synchronized (lock) {
+ if (completed) {
+ return;
+ }
+ completed = true;
+ // As in the synchronous retry loop, a cancelled attempt (cancel() or the call timeout firing)
+ // reports its own failure rather than the response received before it.
+ previous = canceled || call.isCanceled() ? null : previousResponse;
+ previousResponse = null;
+ closePreviousResponse();
+ }
+ if (previous == null) {
+ callback.onFailure(call, e);
+ return;
+ }
+ // A retry attempt failed with a transport error; the response already received from the API is more
+ // actionable, as in the synchronous retry loop. As with OkHttp, the callback owns the response once it is
+ // delivered, and is not called again if onResponse throws.
+ try {
+ callback.onResponse(call, previous);
+ } catch (IOException | RuntimeException callbackError) {
+ previous.close();
+ if (callbackError instanceof RuntimeException) {
+ throw (RuntimeException) callbackError;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ Call call;
+ boolean wasWaiting;
+ synchronized (lock) {
+ if (canceled) {
+ return;
+ }
+ canceled = true;
+ wasWaiting = pendingRetry != null && pendingRetry.cancel(false);
+ pendingRetry = null;
+ closePreviousResponse();
+ call = currentCall;
+ }
+ if (call != null) {
+ call.cancel();
+ }
+ if (wasWaiting) {
+ failCanceled();
+ }
+ }
+
+ /** Completes the call with a cancellation failure, unless the callback has already been invoked. */
+ private void failCanceled() {
+ Call call;
+ synchronized (lock) {
+ if (completed) {
+ return;
+ }
+ completed = true;
+ closePreviousResponse();
+ call = currentCall;
+ }
+ callback.onFailure(call, new IOException("Canceled"));
+ }
+
+ private void closePreviousResponse() {
+ if (previousResponse != null) {
+ previousResponse.close();
+ previousResponse = null;
+ }
+ }
+ }
+
/**
* Pauses the OkHttp call timeout so that it applies to each attempt individually rather than to the whole
* retry loop: the backoff wait (e.g. a {@code Retry-After} the size of the call timeout) does not consume the
diff --git a/src/main/java/com/newscatcher/catchall/core/SseEventParser.java b/src/main/java/com/newscatcher/catchall/core/SseEventParser.java
index 4118d9d..641fc8c 100644
--- a/src/main/java/com/newscatcher/catchall/core/SseEventParser.java
+++ b/src/main/java/com/newscatcher/catchall/core/SseEventParser.java
@@ -51,6 +51,27 @@ private SseEventParser() {
*/
public static T parseEventLevelUnion(
String eventType, String data, String id, Long retry, Class unionClass, String discriminatorProperty) {
+ return parseEventLevelUnion(eventType, data, id, retry, unionClass, discriminatorProperty, null);
+ }
+
+ /**
+ * Parse an SSE event using event-level discrimination.
+ *
+ * Events in {@code envelopeEvents} (or every event, if {@code envelopeEvents} is null) are deserialized from the
+ * SSE envelope ({@code event}, {@code data}, {@code id}, {@code retry}). Other events are deserialized from the
+ * JSON {@code data} payload with the discriminator injected.
+ */
+ public static T parseEventLevelUnion(
+ String eventType,
+ String data,
+ String id,
+ Long retry,
+ Class unionClass,
+ String discriminatorProperty,
+ Set envelopeEvents) {
+ if (envelopeEvents != null && eventType != null && !envelopeEvents.contains(eventType)) {
+ return parsePayloadWithInjectedDiscriminator(eventType, data, unionClass, discriminatorProperty);
+ }
try {
// Determine if data should be parsed as JSON based on the variant's expected type
Object parsedData = parseDataForVariant(eventType, data, unionClass, discriminatorProperty);
@@ -74,6 +95,21 @@ public static T parseEventLevelUnion(
}
}
+ private static T parsePayloadWithInjectedDiscriminator(
+ String eventType, String data, Class unionClass, String discriminatorProperty) {
+ try {
+ Map payload = new HashMap<>();
+ if (data != null && !data.isEmpty()) {
+ payload.putAll(ObjectMappers.JSON_MAPPER.readValue(data, new TypeReference