`). |
| `getStatistics()` | Dataset statistics. Completes with `JsonNode`; throws `NotFoundError` if the dataset is gone (no separate "statistics absent" state). |
-| `createItemsPublicUrl(DatasetListItemsOptions, Long expiresInSecs)` | A public (optionally signed) items URL, served as `json`. Completes with `String`. |
+| `createItemsPublicUrl(DatasetListItemsOptions, Long expiresInSecs)` | A public (optionally signed) items URL, served as `json`. `expiresInSecs` must be positive and at most `4611686018427387` when set. Completes with `String`. |
| `createItemsPublicUrl(DatasetListItemsOptions, Long expiresInSecs, DownloadItemsFormat format)` | As above, with the URL's serialization `format` set explicitly. |
> **Server-side item filters, `unwind`, and iteration.** The dataset-items endpoint applies
> `offset`/`limit` to the raw rows and then transforms them: a filter (`skipEmpty`, `skipHidden`,
> `clean`, `simplified`) can drop rows (fewer items returned than scanned), and `unwind` can split a
-> row's array field into several items (more items returned than scanned). Where the API reports the
-> number of rows it scanned (`X-Apify-Pagination-Count`), `iterateItems` advances by that number
-> rather than by the number of items returned, so neither case repeats already-seen items, skips
-> rows, nor ends iteration early. (A reported `0` is trusted only when the page also returned
-> nothing, since scanning zero rows can never produce items; any other combination falls back to the
-> returned count, as if the header were absent.)
+> row's array field into several items (more items returned than scanned). On a page from
+> `listItems`, `PaginationList.getCount()` is the number of rows the API scanned for it
+> (`X-Apify-Pagination-Count`), or the number of items returned when that is larger, since the header
+> is derived from the dataset's item count, which lags a fresh push. `iterateItems` advances by that
+> count, capped at the rows the request asked for, rather than by the number of items returned, so
+> neither case repeats already-seen items, skips rows, nor ends iteration early. The `limit` counts
+> rows too: a filter can leave fewer items than `limit`, and `unwind` more. To page through a
+> filtered dataset by hand, advance `offset` by `min(count, limit)`; the cap matters only with
+> `unwind`. `PaginationList.getScannedCount()` still returns the raw header value, and is
+> deprecated in favor of `getCount()`.
```java
Dataset ds = client.datasets().getOrCreate("my-dataset").join();
@@ -106,8 +110,11 @@ adds `attachment` (`Boolean`), `bom` (`Boolean`), `delimiter` (`String`), `skipH
`createItemsPublicUrl(DatasetListItemsOptions, Long expiresInSecs[, DownloadItemsFormat format])`
completes with a `String` URL. If the dataset is private, the client fetches it, reads its
URL-signing secret, and appends an HMAC-SHA256 signature (bounded by `expiresInSecs`, or
-non-expiring when `null`); for public datasets the URL is unsigned. The optional `format` sets the
-serialization the URL serves items in (default `json` when omitted/`null`).
+non-expiring when `null`); for public datasets the URL is unsigned. `expiresInSecs` must be positive
+and at most `4611686018427387`, otherwise the method throws `IllegalArgumentException` before any
+request is sent. The expiry counts from the moment the URL is signed, after the dataset lookup. The
+optional `format` sets the serialization the URL serves items in (default `json` when
+omitted/`null`).
> **Public-URL limitation (matches the JavaScript reference client).** The public-URL builders
> (`createItemsPublicUrl`, and the key-value-store `getRecordPublicUrl` / `createKeysPublicUrl`)
@@ -139,7 +146,7 @@ datasets.
| `setRecordJson(String key, Object value)` | Store JSON. Completes with no value. |
| `deleteRecord(String key)` | Delete a record. Completes with no value. |
| `getRecordPublicUrl(String key)` | A public (optionally signed) record URL. Completes with `String`. |
-| `createKeysPublicUrl(Long expiresInSecs)` | A public (optionally signed) key-list URL. Completes with `String`. |
+| `createKeysPublicUrl(Long expiresInSecs)` | A public (optionally signed) key-list URL. `expiresInSecs` must be positive and at most `4611686018427387` when set. Completes with `String`. |
| `createKeysPublicUrl(ListKeysOptions, Long expiresInSecs)` | As above, forwarding key-listing filters (`limit`, `prefix`, `collection`, `exclusiveStartKey`). |
`ListKeysOptions` fields (all optional): `limit(Long)`, `exclusiveStartKey(String)`,
@@ -186,7 +193,8 @@ Both `getRecordPublicUrl` and `createKeysPublicUrl` complete with a `String` URL
and unsigned for public ones. They differ because they sign different things: a single record URL
signs the record key directly (there is no expiry to bound), while a key-list URL uses an
expiry-aware storage-content signature — hence only `createKeysPublicUrl` takes an `expiresInSecs`
-(`null` = non-expiring).
+(`null` = non-expiring, otherwise positive and at most `4611686018427387`; anything else throws
+`IllegalArgumentException` before any request is sent).
## Request queues
diff --git a/pom.xml b/pom.xml
index bbb34ab..b8e90ab 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,7 @@
com.apify
apify-client
- 0.12.0
+ 0.13.0
jar
Apify Java Client
diff --git a/src/main/java/com/apify/client/PaginationList.java b/src/main/java/com/apify/client/PaginationList.java
index 6caf693..77b7a21 100644
--- a/src/main/java/com/apify/client/PaginationList.java
+++ b/src/main/java/com/apify/client/PaginationList.java
@@ -39,7 +39,17 @@ public long getLimit() {
return limit;
}
- /** Number of items actually returned in this page. */
+ /**
+ * Number of entries on this page. On a page from {@link
+ * com.apify.client.dataset.DatasetClient#listItems}, the number of dataset rows the API scanned
+ * for it ({@code X-Apify-Pagination-Count}), or the number of items returned when that is larger,
+ * since the header is derived from the dataset's item count, which lags a fresh push. Filters
+ * ({@code clean}, {@code skipEmpty}) can leave it above the number of items returned. The next
+ * page starts at {@code offset + Math.min(count, limit)}; the cap matters only with {@code
+ * unwind}, which can return more items than the rows scanned. On any other list it is usually the
+ * number of items returned; when the API drops entries after counting them, as the Store list
+ * can, it is the number counted.
+ */
public long getCount() {
return count;
}
@@ -50,24 +60,17 @@ public boolean isDesc() {
}
/**
- * The number of rows the API scanned to produce this page, before any item-level transform (e.g.
- * dataset items' {@code clean}/{@code skipEmpty}/{@code skipHidden} filters, which can make this
- * larger than {@link #getCount()}, or {@code unwind}, which can make it smaller), as reported by
- * the {@code X-Apify-Pagination-Count} response header. {@code null} when the endpoint does not
- * send that header, in which case {@link #getCount()} is the right value to advance pagination by
- * - the two always agree outside the dataset items endpoint.
- *
- * A reported {@code 0} together with a nonzero {@link #getCount()} cannot be a genuine answer
- * (scanning zero rows can never produce items), so that specific combination is treated as the
- * header not actually being populated for the request rather than as real data. See {@code
- * AsyncPaginatedPublisher.applyPage} for the exact fallback callers reading this field directly
- * should mirror.
+ * The raw value of the {@code X-Apify-Pagination-Count} response header of a dataset items page:
+ * the number of rows the API scanned to produce it. {@code null} when the header is absent, which
+ * is always the case outside the dataset items endpoint.
*
- *
Internal pagination-engine detail, exposed here (rather than hidden) only because {@link
- * #getItems()} and this metadata necessarily travel together on the same page object; most
- * callers never need it directly - {@link com.apify.client.dataset.DatasetClient#iterateItems}
- * already accounts for it.
+ * @deprecated {@link #getCount()} reports the scanned rows, or the items returned when that is
+ * larger, which is what pagination has to advance by: the header is derived from the
+ * dataset's item count, which lags a fresh push, so it can fall below the items returned.
+ * Page by {@code getCount()}, as {@link com.apify.client.dataset.DatasetClient#iterateItems}
+ * does.
*/
+ @Deprecated
public Long getScannedCount() {
return scannedCount;
}
@@ -104,6 +107,12 @@ public void setDesc(boolean desc) {
this.desc = desc;
}
+ /**
+ * Sets the raw value of the scanned-rows header.
+ *
+ * @deprecated see {@link #getScannedCount()}
+ */
+ @Deprecated
public void setScannedCount(Long scannedCount) {
this.scannedCount = scannedCount;
}
diff --git a/src/main/java/com/apify/client/Version.java b/src/main/java/com/apify/client/Version.java
index 2e3d187..8c1d9fe 100644
--- a/src/main/java/com/apify/client/Version.java
+++ b/src/main/java/com/apify/client/Version.java
@@ -13,13 +13,13 @@ public final class Version {
* The semantic version of this client library (see SemVer).
* Changes to the public interface other than additive ones are considered breaking changes.
*/
- public static final String CLIENT_VERSION = "0.12.0";
+ public static final String CLIENT_VERSION = "0.13.0";
/**
* The version of the Apify OpenAPI specification this client was generated and verified against.
* Corresponds to the {@code info.version} field of the Apify OpenAPI document.
*/
- public static final String API_SPEC_VERSION = "v2-2026-10-29T100357Z";
+ public static final String API_SPEC_VERSION = "v2-2026-11-04T151257Z";
private Version() {}
}
diff --git a/src/main/java/com/apify/client/actor/Actor.java b/src/main/java/com/apify/client/actor/Actor.java
index 52bc6be..a5778ee 100644
--- a/src/main/java/com/apify/client/actor/Actor.java
+++ b/src/main/java/com/apify/client/actor/Actor.java
@@ -28,6 +28,7 @@ public final class Actor extends ApifyResource {
private String seoDescription;
private List categories = List.of();
private ActorStandby actorStandby;
+ private String standbyUrl;
private String actorPermissionLevel;
/** The unique Actor ID. */
@@ -142,6 +143,14 @@ public ActorStandby getActorStandby() {
return actorStandby;
}
+ /**
+ * The URL for sending requests to the Actor in Standby mode; {@code null} if Standby mode is not
+ * enabled. See {@link ActorClient#server()} for sending them.
+ */
+ public String getStandbyUrl() {
+ return standbyUrl;
+ }
+
/** The permission level of the Actor on the Apify platform (e.g. {@code "OWNER"}). */
public String getActorPermissionLevel() {
return actorPermissionLevel;
diff --git a/src/main/java/com/apify/client/actor/ActorClient.java b/src/main/java/com/apify/client/actor/ActorClient.java
index 36666dc..8228ae3 100644
--- a/src/main/java/com/apify/client/actor/ActorClient.java
+++ b/src/main/java/com/apify/client/actor/ActorClient.java
@@ -18,6 +18,9 @@
import com.apify.client.webhook.NestedWebhookCollectionClient;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import tools.jackson.databind.JsonNode;
/**
@@ -27,11 +30,17 @@
* access its runs, builds, versions and webhooks.
*/
public final class ActorClient {
+ private static final Logger LOG = LoggerFactory.getLogger(ActorClient.class);
+
+ /** Set once the deprecation of {@link #standby()} was logged, which is done once per process. */
+ private static final AtomicBoolean STANDBY_DEPRECATION_LOGGED = new AtomicBoolean();
+
private final ApifyClient root;
private final HttpClientCore http;
private final ResourceContext ctx;
private final String baseUrl;
private final String id;
+ private final ActorServerClient server;
public ActorClient(ApifyClient root, HttpClientCore http, String baseUrl, String id) {
this.root = root;
@@ -39,6 +48,7 @@ public ActorClient(ApifyClient root, HttpClientCore http, String baseUrl, String
this.ctx = ResourceContext.single(http, baseUrl, ApiPaths.ACTORS, id);
this.baseUrl = baseUrl;
this.id = id;
+ this.server = new ActorServerClient(http, baseUrl, id);
}
/** The Actor's ID (or {@code username~name}) as provided. */
@@ -196,6 +206,36 @@ public RunClient lastRun(LastRunOptions options) {
return RunClient.lastRun(root, http, ctx.subUrl(""), options);
}
+ /**
+ * Returns a client for the server interface of this Actor.
+ *
+ * A Server Actor runs a web server, its server interface, that is called over HTTP without
+ * starting a run for each call. Server Actors were formerly called Actor Standby, and {@link
+ * #standby()} is a deprecated alias of this method. Every call returns the same client, so the
+ * server URL it fetches is reused.
+ *
+ * @see Standby mode
+ */
+ public ActorServerClient server() {
+ return server;
+ }
+
+ /**
+ * Returns a client for the server interface of this Actor in Standby mode, named after Actor
+ * Standby, the former name of Server Actors.
+ *
+ * @deprecated Use {@link #server()} instead. Will be removed in a future major version.
+ */
+ @Deprecated
+ public ActorServerClient standby() {
+ if (STANDBY_DEPRECATION_LOGGED.compareAndSet(false, true)) {
+ LOG.warn(
+ "`standby()` is deprecated and will be removed in a future major version. Use"
+ + " `server()` instead.");
+ }
+ return server();
+ }
+
/** A client for this Actor's build collection. */
public BuildCollectionClient builds() {
return new BuildCollectionClient(http, ctx.subUrl(""), "builds");
diff --git a/src/main/java/com/apify/client/actor/ActorServerClient.java b/src/main/java/com/apify/client/actor/ActorServerClient.java
new file mode 100644
index 0000000..b16c200
--- /dev/null
+++ b/src/main/java/com/apify/client/actor/ActorServerClient.java
@@ -0,0 +1,400 @@
+package com.apify.client.actor;
+
+import com.apify.client.build.Build;
+import com.apify.client.http.ApiResponse;
+import com.apify.client.http.ServerActorException;
+import com.apify.client.http.Timeout;
+import com.apify.client.internal.ApiPaths;
+import com.apify.client.internal.CancellableFuture;
+import com.apify.client.internal.HttpClientCore;
+import com.apify.client.internal.Json;
+import com.apify.client.internal.QueryParams;
+import com.apify.client.internal.RequestPolicy;
+import com.apify.client.internal.ResourceContext;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import tools.jackson.databind.JsonNode;
+
+/**
+ * A client for the server interface of a specific Actor, from {@link ActorClient#server()}.
+ *
+ *
A Server Actor (formerly Actor Standby) runs a web server. You call it over HTTP, and the
+ * platform keeps a run ready to answer, so each call skips the run startup. This client sends
+ * requests to the Actor's web server and gets its URL and OpenAPI specification.
+ *
+ *
{@code
+ * ActorServerClient server = client.actor("username/actor-name").server();
+ * ApiResponse response =
+ * server.request("/search", new ActorServerRequestOptions().param("q", "apify")).join();
+ * System.out.println(response.bodyAsString());
+ * }
+ *
+ * @see Standby mode
+ */
+public final class ActorServerClient {
+
+ /** The tag whose build serves the server interface when the Actor sets no build. */
+ private static final String DEFAULT_BUILD_TAG = "latest";
+
+ /** A build number like {@code 0.1.23}: the version {@code 0.1} and the sequence number 23. */
+ private static final Pattern BUILD_NUMBER_PATTERN = Pattern.compile("^(\\d+)\\.(\\d+)\\.(\\d+)$");
+
+ /** The path of an Actor's builds, relative to the Actor. */
+ private static final String BUILDS_PATH = "builds";
+
+ /** The path of a request that names none: the root of the server. */
+ private static final String ROOT_PATH = "/";
+
+ private static final String FRAGMENT_START = "#";
+
+ private final HttpClientCore http;
+ private final ResourceContext ctx;
+ private final String baseUrl;
+ private final String id;
+
+ /**
+ * The server URL. Getting it takes an API call for the Actor, so it is cached for {@link #getUrl}
+ * and the requests. It is pending while the first lookup runs, so concurrent first calls share
+ * one API request, and it is cleared when the lookup fails, so the next call tries again.
+ */
+ private final AtomicReference> url = new AtomicReference<>();
+
+ ActorServerClient(HttpClientCore http, String baseUrl, String id) {
+ this.http = http;
+ this.ctx = ResourceContext.single(http, baseUrl, ApiPaths.ACTORS, id);
+ this.baseUrl = baseUrl;
+ this.id = id;
+ }
+
+ /**
+ * Returns the URL of the Actor's server.
+ *
+ * The URL is the Actor's {@code standbyUrl}, fetched from the API by the first call of this
+ * method or {@link #getOpenApiSpec()} and reused by later calls on this client. Use it to call
+ * the server with another HTTP client, for example one generated from the server's OpenAPI
+ * specification. The API request takes the {@link Timeout#SHORT} timeout.
+ *
+ * @return the URL of the Actor's server; fails with a {@link ServerActorException} when the Actor
+ * is not a Server Actor, and with a {@link com.apify.client.http.NotFoundError} when the
+ * Actor does not exist
+ */
+ public CompletableFuture getUrl() {
+ return getUrl(Timeout.SHORT);
+ }
+
+ /**
+ * As {@link #getUrl()}, with {@code timeout} for the API request that fetches the URL. It is not
+ * used when the URL is already cached, nor while another caller's lookup is running, since the
+ * timeout of the caller that started it applies.
+ *
+ * @throws IllegalArgumentException if {@code timeout} is {@code null}
+ */
+ public CompletableFuture getUrl(Timeout timeout) {
+ requireTimeout(timeout);
+ // A copy, so that cancelling the returned future leaves the shared lookup alone.
+ return sharedUrl(timeout).copy();
+ }
+
+ private static void requireTimeout(Timeout timeout) {
+ if (timeout == null) {
+ throw new IllegalArgumentException("timeout must not be null");
+ }
+ }
+
+ /** The cached lookup of the server URL, started if there is none. */
+ private CompletableFuture sharedUrl(Timeout timeout) {
+ while (true) {
+ CompletableFuture cached = url.get();
+ if (cached != null) {
+ return cached;
+ }
+ CancellableFuture lookup = new CancellableFuture<>();
+ if (url.compareAndSet(null, lookup)) {
+ CompletableFuture fetch;
+ try {
+ fetch = fetchServerActor(timeout);
+ } catch (RuntimeException e) {
+ // A lookup that cannot even start must not stay cached, or every later call would wait on
+ // it.
+ fetch = CompletableFuture.failedFuture(e);
+ }
+ fetch.whenComplete(
+ (actor, error) -> {
+ if (error == null) {
+ lookup.complete(actor.getStandbyUrl());
+ } else {
+ url.compareAndSet(lookup, null);
+ lookup.completeExceptionally(HttpClientCore.unwrapCompletion(error));
+ }
+ });
+ return lookup;
+ }
+ }
+ }
+
+ /**
+ * Returns the OpenAPI specification of the Actor's server.
+ *
+ * The specification is the web server schema from the Actor's definition, taken from the build
+ * that serves the server interface: the build set in {@code actorStandby.build} (a build tag or a
+ * build number), or {@code latest} if none is set. Use it to generate an HTTP client for the
+ * server, and point that client at the URL returned by {@link #getUrl()}.
+ *
+ * @return the OpenAPI specification, or empty if the Actor's definition does not include one;
+ * fails with a {@link ServerActorException} when the Actor is not a Server Actor or the build
+ * that serves its server interface does not exist, and with a {@link
+ * com.apify.client.http.NotFoundError} when the Actor does not exist
+ * @see Web
+ * server schema
+ */
+ public CompletableFuture> getOpenApiSpec() {
+ return getOpenApiSpec(Timeout.SHORT);
+ }
+
+ /** As {@link #getOpenApiSpec()}, with {@code timeout} for each API request it makes. */
+ public CompletableFuture> getOpenApiSpec(Timeout timeout) {
+ requireTimeout(timeout);
+ return fetchServerActor(timeout)
+ .thenCompose(
+ actor -> {
+ url.set(CancellableFuture.completed(actor.getStandbyUrl()));
+ String buildRef = buildRefOf(actor);
+ Optional taggedBuildId = taggedBuildId(actor, buildRef);
+ CompletableFuture> buildId =
+ taggedBuildId.isPresent()
+ ? CompletableFuture.completedFuture(taggedBuildId)
+ // Not a tag, so look `buildRef` up as a build number, which the API cannot
+ // resolve directly.
+ : findBuildIdByNumber(buildRef, timeout);
+ return buildId.thenCompose(found -> fetchBuild(found, buildRef, timeout));
+ })
+ .thenApply(
+ build ->
+ Optional.ofNullable(build.getActorDefinition())
+ .map(definition -> definition.getWebServerSchema())
+ .filter(schema -> !schema.isNull()));
+ }
+
+ /**
+ * Sends an HTTP request to the Actor's server and completes with its response, the body buffered.
+ *
+ * The request goes to the URL returned by {@link #getUrl()} and carries the client's token in
+ * the {@code X-Apify-Authorization} header, which the platform consumes, so it never reaches the
+ * Actor's code. Headers you set, {@code Authorization} included, reach the Actor's server as they
+ * are. The first request to an idle Actor waits for the platform to start a server run, which is
+ * why the timeout defaults to {@link Timeout#LONG}.
+ *
+ *
The future completes with any response of the server, error statuses included, so check
+ * {@link ApiResponse#statusCode()}. A rate-limit (429) or Service Unavailable (503) response is
+ * retried with the client's retry settings, and the last one is returned once the retries run
+ * out. The Actor's code can return these statuses too, and then it gets the request again. A
+ * redirect is returned as it is, since following it would send the token to a host the Actor's
+ * code picked. A transport error, a timeout included, is not retried, since the Actor may have
+ * handled the request before it failed. Request bodies are sent uncompressed unless {@link
+ * ActorServerRequestOptions#compression} is set, since the Actor's server may not decode a
+ * compressed body.
+ *
+ * @param path the path on the Actor's server, for example {@code /search}. It is always resolved
+ * against the server URL, so the request, and the token with it, cannot be sent to another
+ * host. Pass the query as {@link ActorServerRequestOptions#param params}. A fragment ({@code
+ * #...}) is dropped, since it is never sent to the server.
+ * @param options the method, query, headers, body, compression and timeout; {@code null} sends a
+ * plain {@code GET}
+ * @return the response; fails with a {@link ServerActorException} when the Actor is not a Server
+ * Actor, and with a {@link com.apify.client.http.NotFoundError} when the Actor does not exist
+ */
+ public CompletableFuture request(String path, ActorServerRequestOptions options) {
+ ActorServerRequestOptions opts = options != null ? options : new ActorServerRequestOptions();
+ byte[] body = Json.toRequestBody(opts.bodyValue());
+ return getUrl()
+ .thenCompose(
+ serverUrl ->
+ http.call(
+ opts.methodValue(),
+ requestUrl(serverUrl, path, opts),
+ body,
+ contentTypeOf(opts),
+ opts.headersValue(),
+ opts.timeoutValue(),
+ false,
+ RequestPolicy.server(opts.compressionValue())));
+ }
+
+ /** A {@code GET} request to {@code path} on the Actor's server; see {@link #request}. */
+ public CompletableFuture request(String path) {
+ return request(path, null);
+ }
+
+ /** A {@code GET} request to the root of the Actor's server; see {@link #request}. */
+ public CompletableFuture request() {
+ return request(ROOT_PATH, null);
+ }
+
+ /**
+ * As {@link #request}, but completes with the response as soon as its headers arrive, with the
+ * body unread. A response whose body is large, or never ends, can be processed in chunks this
+ * way. The body holds its connection open until it is read to the end or closed, so close the
+ * returned {@link ActorServerStreamResponse}. A rate-limit (429) or Service Unavailable (503)
+ * response is retried like for {@link #request}, with the discarded responses closed; a transport
+ * error is not. {@link ActorServerRequestOptions#timeout} bounds the wait for the server to start
+ * responding.
+ */
+ public CompletableFuture requestStream(
+ String path, ActorServerRequestOptions options) {
+ ActorServerRequestOptions opts = options != null ? options : new ActorServerRequestOptions();
+ byte[] body = Json.toRequestBody(opts.bodyValue());
+ return getUrl()
+ .thenCompose(
+ serverUrl ->
+ http.streamRequest(
+ opts.methodValue(),
+ requestUrl(serverUrl, path, opts),
+ body,
+ contentTypeOf(opts),
+ opts.headersValue(),
+ opts.timeoutValue(),
+ RequestPolicy.server(opts.compressionValue())))
+ .thenApply(
+ response ->
+ new ActorServerStreamResponse(
+ response.statusCode(), response.headers(), response.body()));
+ }
+
+ /** Fetches the Actor and checks that it is a Server Actor. */
+ private CompletableFuture fetchServerActor(Timeout timeout) {
+ return ctx.getResourceRequired("", new QueryParams(), Actor.class, timeout)
+ .thenApply(
+ actor -> {
+ String standbyUrl = actor == null ? null : actor.getStandbyUrl();
+ if (standbyUrl == null || standbyUrl.isEmpty()) {
+ throw new ServerActorException("Actor " + id + " is not a Server Actor.");
+ }
+ return actor;
+ });
+ }
+
+ /** The tag or number of the build that serves the server interface. */
+ private static String buildRefOf(Actor actor) {
+ ActorStandby standby = actor.getActorStandby();
+ String build = standby == null ? null : standby.getBuild();
+ return build == null || build.isEmpty() ? DEFAULT_BUILD_TAG : build;
+ }
+
+ /** The id of the build the tag {@code buildRef} points to, or empty if it is not a tag. */
+ private static Optional taggedBuildId(Actor actor, String buildRef) {
+ JsonNode tagged = actor.getTaggedBuilds();
+ JsonNode buildId = tagged == null ? null : tagged.path(buildRef).path("buildId");
+ return buildId != null && buildId.isTextual()
+ ? Optional.of(buildId.asText())
+ : Optional.empty();
+ }
+
+ /**
+ * Looks {@code buildNumber} up in the Actor's builds, which are listed newest first and whose
+ * sequence numbers within a version grow over time, so meeting an older build of the same version
+ * means the build does not exist. Completes with empty for anything that is not a build number.
+ */
+ private CompletableFuture> findBuildIdByNumber(
+ String buildNumber, Timeout timeout) {
+ Optional target = parseBuildNumber(buildNumber);
+ if (target.isEmpty()) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+ ResourceContext builds = ResourceContext.collection(http, ctx.subUrl(""), BUILDS_PATH);
+ return findBuildIdByNumber(builds, buildNumber, target.get(), 0, timeout);
+ }
+
+ private CompletableFuture> findBuildIdByNumber(
+ ResourceContext builds, String buildNumber, int[] target, long offset, Timeout timeout) {
+ QueryParams params = new QueryParams().addBool("desc", true).addLong("offset", offset);
+ return builds
+ .listResource("", params, Build.class, timeout)
+ .thenCompose(
+ page -> {
+ for (Build build : page.getItems()) {
+ if (buildNumber.equals(build.getBuildNumber())) {
+ return CompletableFuture.completedFuture(Optional.of(build.getId()));
+ }
+ if (isOlderBuild(build.getBuildNumber(), target)) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+ }
+ if (page.getItems().isEmpty()) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+ return findBuildIdByNumber(
+ builds, buildNumber, target, offset + page.getItems().size(), timeout);
+ });
+ }
+
+ private CompletableFuture fetchBuild(
+ Optional buildId, String buildRef, Timeout timeout) {
+ if (buildId.isEmpty()) {
+ return CompletableFuture.failedFuture(buildDoesNotExist(buildRef));
+ }
+ return ResourceContext.single(http, baseUrl, ApiPaths.ACTOR_BUILDS, buildId.get())
+ .getResource("", new QueryParams(), Build.class, timeout)
+ .thenApply(build -> build.orElseThrow(() -> buildDoesNotExist(buildRef)));
+ }
+
+ private ServerActorException buildDoesNotExist(String buildRef) {
+ return new ServerActorException("Build " + buildRef + " of Actor " + id + " does not exist.");
+ }
+
+ /** Splits a build number like {@code 0.1.23} into its parts, or empty for anything else. */
+ private static Optional parseBuildNumber(String buildNumber) {
+ if (buildNumber == null) {
+ return Optional.empty();
+ }
+ Matcher match = BUILD_NUMBER_PATTERN.matcher(buildNumber);
+ if (!match.matches()) {
+ return Optional.empty();
+ }
+ return Optional.of(
+ new int[] {
+ Integer.parseInt(match.group(1)),
+ Integer.parseInt(match.group(2)),
+ Integer.parseInt(match.group(3))
+ });
+ }
+
+ /**
+ * Whether {@code buildNumber} belongs to the same version as {@code target} and comes before it.
+ */
+ private static boolean isOlderBuild(String buildNumber, int[] target) {
+ return parseBuildNumber(buildNumber)
+ .filter(parsed -> parsed[0] == target[0] && parsed[1] == target[1] && parsed[2] < target[2])
+ .isPresent();
+ }
+
+ /**
+ * The URL of a request: {@code path} resolved against the server URL, without its fragment and
+ * with the query parameters appended.
+ */
+ private static String requestUrl(
+ String serverUrl, String path, ActorServerRequestOptions options) {
+ String requested = path == null ? ROOT_PATH : path;
+ String withoutFragment = requested.split(FRAGMENT_START, 2)[0];
+ String serverPath = withoutFragment.replaceFirst("^/+", "");
+ String base = serverUrl.replaceFirst("/+$", "");
+ return options.queryParams().applyToUrl(base + "/" + serverPath);
+ }
+
+ /**
+ * The content type of the body: the one the caller set, JSON for a body that is serialized, and
+ * none for a body that is sent as it is.
+ */
+ private static String contentTypeOf(ActorServerRequestOptions options) {
+ String set = options.contentTypeHeader();
+ if (set != null) {
+ return set;
+ }
+ Object body = options.bodyValue();
+ boolean sentAsIs = body == null || body instanceof byte[] || body instanceof String;
+ return sentAsIs ? null : ResourceContext.CONTENT_TYPE_JSON;
+ }
+}
diff --git a/src/main/java/com/apify/client/actor/ActorServerRequestOptions.java b/src/main/java/com/apify/client/actor/ActorServerRequestOptions.java
new file mode 100644
index 0000000..3ca1306
--- /dev/null
+++ b/src/main/java/com/apify/client/actor/ActorServerRequestOptions.java
@@ -0,0 +1,164 @@
+package com.apify.client.actor;
+
+import com.apify.client.http.HttpCompressor;
+import com.apify.client.http.Timeout;
+import com.apify.client.internal.QueryParams;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Configures {@link ActorServerClient#request(String, ActorServerRequestOptions)} and {@link
+ * ActorServerClient#requestStream(String, ActorServerRequestOptions)}. All fields are optional: by
+ * default the request is a {@code GET} without query parameters, headers or body, sent with the
+ * {@link Timeout#LONG} timeout.
+ */
+public final class ActorServerRequestOptions {
+
+ /**
+ * The methods a request can use. {@code CONNECT} is left out, since the JDK HTTP client does not
+ * send it.
+ */
+ private static final Set METHODS =
+ Set.of("GET", "HEAD", "POST", "PUT", "DELETE", "OPTIONS", "TRACE", "PATCH");
+
+ private String method = "GET";
+ private final Map params = new LinkedHashMap<>();
+ private final Map headers = new LinkedHashMap<>();
+ private Object body;
+ private HttpCompressor compression;
+ private Timeout timeout = Timeout.LONG;
+
+ /**
+ * The HTTP method of the request, in upper case: {@code GET} (the default), {@code HEAD}, {@code
+ * POST}, {@code PUT}, {@code DELETE}, {@code OPTIONS}, {@code TRACE} or {@code PATCH}.
+ *
+ * @throws IllegalArgumentException for any other method
+ */
+ public ActorServerRequestOptions method(String method) {
+ if (method == null || !METHODS.contains(method)) {
+ throw new IllegalArgumentException(
+ "method must be one of " + METHODS.stream().sorted().toList() + ", was: " + method);
+ }
+ this.method = method;
+ return this;
+ }
+
+ /**
+ * Adds a query parameter. A {@code null} value is left out, a boolean is sent as {@code 1} or
+ * {@code 0} and a collection as its comma-joined elements, as for API calls; any other value is
+ * sent as its string form.
+ */
+ public ActorServerRequestOptions param(String name, Object value) {
+ params.put(name, value);
+ return this;
+ }
+
+ /** Adds all the entries of {@code params} as query parameters; see {@link #param}. */
+ public ActorServerRequestOptions params(Map params) {
+ this.params.putAll(params);
+ return this;
+ }
+
+ /** Adds a header to the request. */
+ public ActorServerRequestOptions header(String name, String value) {
+ headers.put(name, value);
+ return this;
+ }
+
+ /**
+ * Adds all the entries of {@code headers} to the request. They reach the Actor's server as they
+ * are, an {@code Authorization} header included: the client's token travels in the {@code
+ * X-Apify-Authorization} header, which the platform consumes, so it never reaches the Actor's
+ * code and the {@code Authorization} header stays free for the Actor's own authentication.
+ */
+ public ActorServerRequestOptions headers(Map headers) {
+ this.headers.putAll(headers);
+ return this;
+ }
+
+ /**
+ * The request body: a {@code byte[]} or a {@link String} is sent as it is (set its {@code
+ * Content-Type} with {@link #header}), and anything else, a {@link Map}, a list or a POJO, is
+ * serialized to JSON and sent as {@code application/json} unless a {@code Content-Type} header is
+ * set.
+ */
+ public ActorServerRequestOptions body(Object body) {
+ this.body = body;
+ return this;
+ }
+
+ /**
+ * Compresses the request body with {@code compression}, under the same conditions as for API
+ * calls: a body that is large enough and has a content type that is not already compressed, and
+ * no {@code Content-Encoding} header of your own. Set it only if the Actor's server decodes the
+ * resulting {@code Content-Encoding}: the body is sent uncompressed by default, since the server
+ * may not.
+ */
+ public ActorServerRequestOptions compression(HttpCompressor compression) {
+ this.compression = compression;
+ return this;
+ }
+
+ /**
+ * The timeout of each attempt. Defaults to {@link Timeout#LONG}, since the first request to an
+ * idle Actor waits for the platform to start a server run. For {@link
+ * ActorServerClient#requestStream} it bounds the wait for the server to start responding.
+ */
+ public ActorServerRequestOptions timeout(Timeout timeout) {
+ if (timeout == null) {
+ throw new IllegalArgumentException("timeout must not be null");
+ }
+ this.timeout = timeout;
+ return this;
+ }
+
+ String methodValue() {
+ return method;
+ }
+
+ Object bodyValue() {
+ return body;
+ }
+
+ Map headersValue() {
+ return Collections.unmodifiableMap(headers);
+ }
+
+ HttpCompressor compressionValue() {
+ return compression;
+ }
+
+ Timeout timeoutValue() {
+ return timeout;
+ }
+
+ /** The {@code Content-Type} header the caller set, ignoring the case of its name, or null. */
+ String contentTypeHeader() {
+ return headers.entrySet().stream()
+ .filter(entry -> "content-type".equals(entry.getKey().toLowerCase(Locale.ROOT)))
+ .map(Map.Entry::getValue)
+ .findFirst()
+ .orElse(null);
+ }
+
+ /** The query parameters, converted as for API calls. */
+ QueryParams queryParams() {
+ QueryParams query = new QueryParams();
+ params.forEach(
+ (name, value) -> {
+ if (value instanceof Boolean flag) {
+ query.addBool(name, flag);
+ } else if (value instanceof Collection values) {
+ query.addCsv(name, values.stream().map(String::valueOf).collect(Collectors.toList()));
+ } else if (value != null) {
+ query.addString(name, value.toString());
+ }
+ });
+ return query;
+ }
+}
diff --git a/src/main/java/com/apify/client/actor/ActorServerStreamResponse.java b/src/main/java/com/apify/client/actor/ActorServerStreamResponse.java
new file mode 100644
index 0000000..11a9355
--- /dev/null
+++ b/src/main/java/com/apify/client/actor/ActorServerStreamResponse.java
@@ -0,0 +1,26 @@
+package com.apify.client.actor;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpHeaders;
+
+/**
+ * A response of an Actor's server whose body is still unread, as returned by {@link
+ * ActorServerClient#requestStream}.
+ *
+ * The body holds its connection open until it is read to the end or closed, so close the
+ * response (it is {@link AutoCloseable}) or its {@link #body()} when done with it.
+ *
+ * @param statusCode the HTTP status code, of an error response too
+ * @param headers the response headers
+ * @param body the response body, unread
+ */
+public record ActorServerStreamResponse(int statusCode, HttpHeaders headers, InputStream body)
+ implements AutoCloseable {
+
+ /** Closes the body stream, releasing the connection. */
+ @Override
+ public void close() throws IOException {
+ body.close();
+ }
+}
diff --git a/src/main/java/com/apify/client/actor/ActorStandby.java b/src/main/java/com/apify/client/actor/ActorStandby.java
index d620b51..77cb864 100644
--- a/src/main/java/com/apify/client/actor/ActorStandby.java
+++ b/src/main/java/com/apify/client/actor/ActorStandby.java
@@ -62,19 +62,28 @@ public Boolean getShouldPassActorInput() {
}
/**
- * Who the Server Actor serves: {@code SINGLE_TENANT} (a private instance per caller) or a {@code
- * MULTI_TENANT_*} pool.
+ * Whether each user gets their own standby runs ({@code SINGLE_TENANT}) or the runs are shared
+ * across users ({@code MULTI_TENANT}). {@code null} when it is not set.
*/
public String getTenancy() {
return tenancy;
}
- /** Whether Console authentication is enabled for the standby URL. */
+ /**
+ * Whether standby requests can be authenticated with an Apify Console session. When creating an
+ * Actor, omit it or set it to {@code false}. When updating an Actor, only Apify administrators
+ * can change it; other users can only send the current value, or {@code false} if it is not set
+ * yet.
+ */
public Boolean getIsConsoleAuthEnabled() {
return isConsoleAuthEnabled;
}
- /** Whether requests to the standby URL may omit the API token. */
+ /**
+ * Whether standby requests can be made without an API token. When creating an Actor, omit it or
+ * set it to {@code false}. When updating an Actor, only Apify administrators can change it; other
+ * users can only send the current value, or {@code false} if it is not set yet.
+ */
public Boolean getIsTokenlessEnabled() {
return isTokenlessEnabled;
}
diff --git a/src/main/java/com/apify/client/actor/ActorVersion.java b/src/main/java/com/apify/client/actor/ActorVersion.java
index d3560c8..cb73510 100644
--- a/src/main/java/com/apify/client/actor/ActorVersion.java
+++ b/src/main/java/com/apify/client/actor/ActorVersion.java
@@ -8,6 +8,7 @@ public final class ActorVersion extends ApifyResource {
private String sourceType;
private String tarballUrl;
private String gitHubGistUrl;
+ private String actorPath;
private Integer githubWebhookId;
private Integer gitlabWebhookId;
private String bitbucketWebhookId;
@@ -22,7 +23,7 @@ public String getSourceType() {
return sourceType;
}
- /** The tarball URL the source is fetched from, for {@code TARBALL} versions. */
+ /** The URL of the zip archive the source is downloaded from, for {@code TARBALL} versions. */
public String getTarballUrl() {
return tarballUrl;
}
@@ -32,6 +33,16 @@ public String getGitHubGistUrl() {
return gitHubGistUrl;
}
+ /**
+ * The Actor's directory inside the uploaded source, as a path relative to the upload root with
+ * {@code /} separators (for example {@code "actors/typescript-actor"}). It applies to {@code
+ * SOURCE_FILES} and {@code TARBALL} versions. {@code null} when the upload root is the Actor's
+ * directory. On a {@code TARBALL} version, only users with write access to the Actor get it.
+ */
+ public String getActorPath() {
+ return actorPath;
+ }
+
/** The ID of the GitHub webhook that triggers builds on push, if configured. */
public Integer getGithubWebhookId() {
return githubWebhookId;
diff --git a/src/main/java/com/apify/client/build/ActorDefinition.java b/src/main/java/com/apify/client/build/ActorDefinition.java
index b96ea02..b6b2a08 100644
--- a/src/main/java/com/apify/client/build/ActorDefinition.java
+++ b/src/main/java/com/apify/client/build/ActorDefinition.java
@@ -77,17 +77,26 @@ public Map getEnvironmentVariables() {
: Collections.unmodifiableMap(environmentVariables);
}
- /** The path to the Dockerfile used for building the Actor on the platform. */
+ /**
+ * The path to the Dockerfile used for building the Actor on the platform, relative to the {@code
+ * .actor/} directory.
+ */
public String getDockerfile() {
return dockerfile;
}
- /** The path to the directory used as the Docker context when building the Actor. */
+ /**
+ * The path to the directory used as the Docker context when building the Actor, relative to the
+ * {@code .actor/} directory.
+ */
public String getDockerContextDir() {
return dockerContextDir;
}
- /** The path to the README file for the Actor; {@code null} if there is none. */
+ /**
+ * The path to the README file for the Actor, relative to the {@code .actor/} directory; {@code
+ * null} if there is none.
+ */
public String getReadme() {
return readme;
}
@@ -102,7 +111,10 @@ public JsonNode getOutput() {
return output;
}
- /** The path to the CHANGELOG file shown in the Actor's information tab; may be {@code null}. */
+ /**
+ * The path to the CHANGELOG file shown in the Actor's information tab, relative to the {@code
+ * .actor/} directory; may be {@code null}.
+ */
public String getChangelog() {
return changelog;
}
diff --git a/src/main/java/com/apify/client/build/BuildActVersion.java b/src/main/java/com/apify/client/build/BuildActVersion.java
index 2e66599..1397531 100644
--- a/src/main/java/com/apify/client/build/BuildActVersion.java
+++ b/src/main/java/com/apify/client/build/BuildActVersion.java
@@ -12,6 +12,7 @@ public final class BuildActVersion extends ApifyResource {
private String versionNumber;
private String gitRepoUrl;
private List sourceFiles;
+ private String actorPath;
/**
* How the version's source was provided (e.g. {@code "SOURCE_FILES"}, {@code "GIT_REPO"}, {@code
@@ -47,4 +48,13 @@ public String getGitRepoUrl() {
public List getSourceFiles() {
return sourceFiles == null ? List.of() : Collections.unmodifiableList(sourceFiles);
}
+
+ /**
+ * The Actor's directory inside the uploaded source, present when the version sets it. It applies
+ * when the source type is {@code SOURCE_FILES} or {@code TARBALL}. On a {@code TARBALL} version,
+ * only users with write access to the Actor get it.
+ */
+ public String getActorPath() {
+ return actorPath;
+ }
}
diff --git a/src/main/java/com/apify/client/dataset/DatasetClient.java b/src/main/java/com/apify/client/dataset/DatasetClient.java
index 9146602..01c3ea1 100644
--- a/src/main/java/com/apify/client/dataset/DatasetClient.java
+++ b/src/main/java/com/apify/client/dataset/DatasetClient.java
@@ -35,7 +35,7 @@ public final class DatasetClient {
/**
* Header reporting the number of rows the API scanned to produce this page, before any item-level
* filtering ({@code clean}/{@code skipEmpty}/{@code skipHidden}) is applied. Absent on older API
- * responses; see {@link PaginationList#getScannedCount()}.
+ * responses; see {@link PaginationList#getCount()}.
*/
private static final String HEADER_PAGINATION_COUNT = "X-Apify-Pagination-Count";
@@ -138,17 +138,18 @@ public Flow.Publisher iterateItems(DatasetListItemsOptions options) {
/**
* Returns a lazy, backpressure-aware publisher over the dataset's items, decoding each into
* {@code itemClass}, fetching pages on demand. The options' {@code limit} caps the total number
- * of items yielded ({@code null} or non-positive = all); {@code chunkSize} is the per-request
- * page size ({@code null} = server default).
+ * of dataset rows read ({@code null} or non-positive = all); {@code chunkSize} is the
+ * per-request page size ({@code null} = server default).
*
* A server-side item filter ({@code skipEmpty}, {@code skipHidden}, {@code clean}, {@code
- * simplified}) or {@code unwind} can make a page's returned item count diverge from the number of
- * rows actually scanned - a filter drops rows (returns fewer), {@code unwind} splits one row's
- * array field into several items (returns more). Where the API reports the scanned count ({@code
- * X-Apify-Pagination-Count}), this iterator advances by that number rather than by the number of
- * items returned, so neither case repeats already-seen items, skips rows, nor ends iteration
- * early. See {@link PaginationList#getScannedCount()} for the one case that header value is not
- * trusted (and why).
+ * simplified}) or {@code unwind} can make a page return fewer or more items than the rows it
+ * covers, so the iterator advances by {@link PaginationList#getCount()} rather than by the items
+ * returned: it never repeats rows, skips rows, or ends early.
+ *
+ *
Without {@code chunkSize}, each request gets the API's default page size, capped by the
+ * remaining {@code limit}. Dataset items have no default limit, so iterating without {@code
+ * chunkSize} loads the whole dataset, or all {@code limit} rows, in one response. Pass {@code
+ * chunkSize} to iterate a large dataset.
*/
public Flow.Publisher iterateItems(
DatasetListItemsOptions options, Long chunkSize, Class itemClass) {
@@ -182,15 +183,17 @@ private CompletableFuture> fetchItemsPage(
resp -> {
JavaType listType = Json.parametric(List.class, Json.type(itemClass));
List items = Json.parse(resp.body(), listType);
- long count = items.size();
+ long returned = items.size();
+ Long scanned = headerLongOrNull(resp, HEADER_PAGINATION_COUNT);
PaginationList result = new PaginationList<>();
result.setItems(items);
- result.setCount(count);
- result.setTotal(headerLong(resp, HEADER_PAGINATION_TOTAL, count));
+ // See PaginationList#getCount for why the header alone is not the count.
+ result.setCount(Math.max(scanned == null ? 0 : scanned, returned));
+ result.setTotal(headerLong(resp, HEADER_PAGINATION_TOTAL, returned));
result.setOffset(headerLong(resp, HEADER_PAGINATION_OFFSET, 0));
- result.setLimit(headerLong(resp, HEADER_PAGINATION_LIMIT, count));
- result.setScannedCount(headerLongOrNull(resp, HEADER_PAGINATION_COUNT));
+ result.setLimit(headerLong(resp, HEADER_PAGINATION_LIMIT, returned));
+ result.setScannedCount(scanned);
if (desc != null) {
result.setDesc(desc);
}
@@ -316,7 +319,11 @@ public CompletableFuture getStatistics() {
* It fetches the dataset, and if the dataset exposes a URL-signing secret key (i.e. it is
* private), appends an HMAC-SHA256 signature so the URL grants access without an API token.
* {@code expiresInSecs} optionally bounds the validity of a signed URL ({@code null} for
- * non-expiring). The URL is built from the configured public base URL.
+ * non-expiring); it must be positive and at most 4611686018427387. The URL is built from the
+ * configured public base URL.
+ *
+ * @throws IllegalArgumentException before any request is sent, if {@code expiresInSecs} is not
+ * {@code null} and is not positive or exceeds that maximum
*/
public CompletableFuture createItemsPublicUrl(
DatasetListItemsOptions options, Long expiresInSecs) {
@@ -330,6 +337,7 @@ public CompletableFuture createItemsPublicUrl(
*/
public CompletableFuture createItemsPublicUrl(
DatasetListItemsOptions options, Long expiresInSecs, DownloadItemsFormat format) {
+ Signatures.requireValidExpiresInSecs(expiresInSecs);
QueryParams params = new QueryParams();
options.apply(params);
if (format != null) {
@@ -362,9 +370,7 @@ private static long headerLong(ApiResponse resp, String name, long fallback) {
/**
* As {@link #headerLong}, but returns {@code null} (rather than a fallback value) when the header
- * is absent, so callers can distinguish "not reported" from any particular number - notably
- * {@code 0}, a value the header can legitimately carry (see {@link
- * PaginationList#getScannedCount()}).
+ * is absent, which {@link PaginationList#getScannedCount()} reports as {@code null}.
*/
private static Long headerLongOrNull(ApiResponse resp, String name) {
var value = resp.headers().firstValueAsLong(name);
diff --git a/src/main/java/com/apify/client/http/ApiResponse.java b/src/main/java/com/apify/client/http/ApiResponse.java
index ecc729d..a71d955 100644
--- a/src/main/java/com/apify/client/http/ApiResponse.java
+++ b/src/main/java/com/apify/client/http/ApiResponse.java
@@ -1,6 +1,8 @@
package com.apify.client.http;
+import com.apify.client.internal.Json;
import java.net.http.HttpHeaders;
+import java.nio.charset.StandardCharsets;
/**
* The parsed result of a single API call: the status code, headers and the fully-buffered response
@@ -17,4 +19,19 @@
* @param headers the response headers
* @param body the fully-buffered response body
*/
-public record ApiResponse(int statusCode, HttpHeaders headers, byte[] body) {}
+public record ApiResponse(int statusCode, HttpHeaders headers, byte[] body) {
+
+ /** The body decoded as UTF-8 text; an empty string for a response without a body. */
+ public String bodyAsString() {
+ return new String(body, StandardCharsets.UTF_8);
+ }
+
+ /**
+ * The body parsed as JSON into {@code type}, for example {@code JsonNode.class} or a POJO.
+ *
+ * @throws tools.jackson.core.JacksonException if the body is not valid JSON for {@code type}
+ */
+ public T bodyAs(Class type) {
+ return Json.parse(body, type);
+ }
+}
diff --git a/src/main/java/com/apify/client/http/ApifyClientException.java b/src/main/java/com/apify/client/http/ApifyClientException.java
index 0fa1d9b..d3b85b3 100644
--- a/src/main/java/com/apify/client/http/ApifyClientException.java
+++ b/src/main/java/com/apify/client/http/ApifyClientException.java
@@ -5,7 +5,7 @@
*
* It is unchecked (a {@link RuntimeException}), matching the reference JavaScript client's
* {@code Error}-based, unchecked error model, so callers are not forced to wrap every call in a
- * {@code try}/{@code catch}. Two concrete subtypes cover the two ways a call can fail:
+ * {@code try}/{@code catch}. Three concrete subtypes cover the ways a call can fail:
*
*
* - {@link ApifyApiException} — the request reached the API, which answered with a non-success
@@ -13,9 +13,11 @@
*
- {@link ApifyTransportException} — the request never produced an API response at all
* (connection failure, DNS, timeout, or a local failure while preparing the request/response,
* e.g. compression).
+ *
- {@link ServerActorException} — a call to an Actor's server cannot be made, since the Actor
+ * is not a Server Actor or the build that serves its server interface does not exist.
*
*
- * Catch this common type to handle both failure modes uniformly, or catch a specific subtype to
+ * Catch this common type to handle all the failure modes uniformly, or catch a specific subtype to
* handle one of them differently.
*/
public abstract class ApifyClientException extends RuntimeException {
diff --git a/src/main/java/com/apify/client/http/DefaultHttpTransport.java b/src/main/java/com/apify/client/http/DefaultHttpTransport.java
index 027ef37..cce9f2b 100644
--- a/src/main/java/com/apify/client/http/DefaultHttpTransport.java
+++ b/src/main/java/com/apify/client/http/DefaultHttpTransport.java
@@ -13,8 +13,9 @@
* native non-blocking {@code sendAsync}.
*
* The per-attempt timeout is applied to each {@link HttpRequest} by the orchestrating client, so
- * this transport sets only a connection timeout of its own. It follows normal redirects and reuses
- * connections from the underlying client's pool.
+ * this transport sets only a connection timeout of its own. It follows normal redirects, except for
+ * a request that asks not to (see {@link HttpTransport#sendAsync(HttpRequest, boolean)}), and
+ * reuses connections from the underlying client's pool.
*/
public final class DefaultHttpTransport implements HttpTransport {
@@ -28,6 +29,12 @@ public final class DefaultHttpTransport implements HttpTransport {
private final HttpClient client;
+ /**
+ * The client that serves a request which must not follow redirects: {@link #client} itself when
+ * it never follows them, otherwise a copy of it that does not. Created on first use.
+ */
+ private volatile HttpClient noRedirectClient;
+
/** Creates a transport with the {@link #DEFAULT_CONNECT_TIMEOUT}. */
public DefaultHttpTransport() {
this(DEFAULT_CONNECT_TIMEOUT);
@@ -62,6 +69,62 @@ public CompletableFuture> sendStreamingAsync(HttpReque
return unwrapTimeout(client.sendAsync(request, HttpResponse.BodyHandlers.ofInputStream()));
}
+ @Override
+ public CompletableFuture> sendAsync(
+ HttpRequest request, boolean followRedirects) {
+ return followRedirects
+ ? sendAsync(request)
+ : unwrapTimeout(
+ clientWithoutRedirects().sendAsync(request, HttpResponse.BodyHandlers.ofByteArray()));
+ }
+
+ @Override
+ public CompletableFuture> sendStreamingAsync(
+ HttpRequest request, boolean followRedirects) {
+ return followRedirects
+ ? sendStreamingAsync(request)
+ : unwrapTimeout(
+ clientWithoutRedirects().sendAsync(request, HttpResponse.BodyHandlers.ofInputStream()));
+ }
+
+ /**
+ * The JDK client's redirect policy is fixed when it is built, so a request that must not follow
+ * redirects goes through a client that was built without that policy. A client wrapped by {@link
+ * #DefaultHttpTransport(HttpClient)} is copied with the settings the JDK exposes (it does not
+ * share the original's connection pool).
+ */
+ private HttpClient clientWithoutRedirects() {
+ HttpClient result = noRedirectClient;
+ if (result == null) {
+ synchronized (this) {
+ result = noRedirectClient;
+ if (result == null) {
+ result = copyWithoutRedirects(client);
+ noRedirectClient = result;
+ }
+ }
+ }
+ return result;
+ }
+
+ private static HttpClient copyWithoutRedirects(HttpClient source) {
+ if (source.followRedirects() == HttpClient.Redirect.NEVER) {
+ return source;
+ }
+ HttpClient.Builder builder =
+ HttpClient.newBuilder()
+ .followRedirects(HttpClient.Redirect.NEVER)
+ .version(source.version())
+ .sslContext(source.sslContext())
+ .sslParameters(source.sslParameters());
+ source.connectTimeout().ifPresent(builder::connectTimeout);
+ source.proxy().ifPresent(builder::proxy);
+ source.authenticator().ifPresent(builder::authenticator);
+ source.cookieHandler().ifPresent(builder::cookieHandler);
+ source.executor().ifPresent(builder::executor);
+ return builder.build();
+ }
+
/**
* Rewrites a {@link java.net.http.HttpTimeoutException} the JDK client fails the future with into
* this contract's transport-agnostic {@link HttpTimeoutException}, unwrapping the {@link
diff --git a/src/main/java/com/apify/client/http/HttpTransport.java b/src/main/java/com/apify/client/http/HttpTransport.java
index 074f3b1..d703df2 100644
--- a/src/main/java/com/apify/client/http/HttpTransport.java
+++ b/src/main/java/com/apify/client/http/HttpTransport.java
@@ -41,4 +41,46 @@ public interface HttpTransport {
* stream. Completes exceptionally the same way as {@link #sendAsync(HttpRequest)}.
*/
CompletableFuture> sendStreamingAsync(HttpRequest request);
+
+ /**
+ * As {@link #sendAsync(HttpRequest)}, but a transport that is asked not to follow redirects
+ * returns a redirect response as it is, since the request carries a token that must not follow a
+ * redirect to another host. A request to an Actor's server is sent this way (see {@link
+ * com.apify.client.actor.ActorServerClient#request}).
+ *
+ * The default implementation sends the request as {@link #sendAsync(HttpRequest)} does when
+ * {@code followRedirects} is {@code true}, and otherwise completes the returned future
+ * exceptionally with an {@link UnsupportedOperationException}: a transport that cannot hold back
+ * redirects must not be trusted with such a request. Override it to support them.
+ *
+ * @param followRedirects whether the transport may follow redirects
+ */
+ default CompletableFuture> sendAsync(
+ HttpRequest request, boolean followRedirects) {
+ return followRedirects
+ ? sendAsync(request)
+ : CompletableFuture.failedFuture(redirectsCannotBeDisabled());
+ }
+
+ /**
+ * As {@link #sendStreamingAsync(HttpRequest)}, with the redirect handling and the default
+ * implementation of {@link #sendAsync(HttpRequest, boolean)}.
+ *
+ * @param followRedirects whether the transport may follow redirects
+ */
+ default CompletableFuture> sendStreamingAsync(
+ HttpRequest request, boolean followRedirects) {
+ return followRedirects
+ ? sendStreamingAsync(request)
+ : CompletableFuture.failedFuture(redirectsCannotBeDisabled());
+ }
+
+ /** The failure of the default implementations when redirects must not be followed. */
+ private static UnsupportedOperationException redirectsCannotBeDisabled() {
+ return new UnsupportedOperationException(
+ "This HttpTransport does not support disabling redirects, which a request to an Actor's"
+ + " server needs so that the API token never follows a redirect to another host."
+ + " Override sendAsync(HttpRequest, boolean) and sendStreamingAsync(HttpRequest,"
+ + " boolean) to support it.");
+ }
}
diff --git a/src/main/java/com/apify/client/http/ServerActorException.java b/src/main/java/com/apify/client/http/ServerActorException.java
new file mode 100644
index 0000000..eda7f91
--- /dev/null
+++ b/src/main/java/com/apify/client/http/ServerActorException.java
@@ -0,0 +1,20 @@
+package com.apify.client.http;
+
+/**
+ * Thrown when a call to an Actor's server cannot be made, since the Actor is not a Server Actor, or
+ * since the build that serves its server interface does not exist. The API was reached, but it has
+ * nothing to send the call to, which is why this is neither an {@link ApifyApiException} nor an
+ * {@link ApifyTransportException}.
+ *
+ * An Actor that does not exist at all is reported as a {@link NotFoundError}, as for any other
+ * call.
+ */
+public final class ServerActorException extends ApifyClientException {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Creates the exception with a message that names the Actor or the build. */
+ public ServerActorException(String message) {
+ super(message);
+ }
+}
diff --git a/src/main/java/com/apify/client/internal/AsyncPaginatedPublisher.java b/src/main/java/com/apify/client/internal/AsyncPaginatedPublisher.java
index b3fbf54..2fe01ba 100644
--- a/src/main/java/com/apify/client/internal/AsyncPaginatedPublisher.java
+++ b/src/main/java/com/apify/client/internal/AsyncPaginatedPublisher.java
@@ -24,20 +24,22 @@
* list()}:
*
*
- * - {@code totalLimit} caps the total number of items yielded across all pages ({@code
- * null} = yield everything).
+ *
- {@code totalLimit} caps the total number of rows read across all pages ({@code
+ * null} = read everything). A row is an item unless a page's {@link PaginationList#getCount()
+ * count} differs from the items it returns; all returned items are yielded.
*
- {@code chunkSize} is the per-request page size ({@code null} = let the API choose). Each
- * page requests {@code min(remaining-under-cap, chunkSize)} items so the cap is never
+ * page requests {@code min(remaining-under-cap, chunkSize)} rows so the cap is never
* overshot.
*
*
- * Each page advances the offset by the number of items actually returned, and iteration stops
- * only when the cap is reached or the API returns an empty page. It deliberately does not
- * stop on a short page or a reported {@code total}: the API clamps an over-large requested page
- * size to its own maximum (so a "short" page is common mid-collection), and some endpoints report a
- * {@code total} of {@code 0} or a value that lags right after a write (e.g. dataset items) — either
- * signal would truncate iteration. Terminating on an empty page costs one extra request at the end
- * but yields the complete result in every case.
+ *
Each page advances the offset by the rows it covered, as {@link PaginationList#getCount()}
+ * defines them, capped at the page limit that was requested. Iteration stops only when the cap is
+ * reached or a page covers no rows. It deliberately does not stop on a short page or a
+ * reported {@code total}: the API clamps an over-large requested page size to its own maximum (so a
+ * "short" page is common mid-collection), and some endpoints report a {@code total} of {@code 0} or
+ * a value that lags right after a write (e.g. dataset items) — either signal would truncate
+ * iteration. Terminating on an empty page costs one extra request at the end but yields the
+ * complete result in every case.
*
*
Supports a single subscriber (as permitted by the Reactive Streams specification); subscribing
* twice completes the second subscriber's subscription with an error instead of sharing state
@@ -109,9 +111,10 @@ private final class Session implements Flow.Subscription {
private List buffer = List.of();
private int pos;
private long offset;
- private long yielded;
+ private long rowsRead;
private boolean exhausted;
private boolean fetchInFlight;
+ private Long requestedLimit;
private volatile CompletableFuture inFlight;
Session(Flow.Subscriber subscriber) {
@@ -207,8 +210,9 @@ private void drainLoop() {
private void fetchNextPage() {
fetchInFlight = true;
- Long capRemaining = totalLimit != null ? totalLimit - yielded : null;
+ Long capRemaining = totalLimit != null ? totalLimit - rowsRead : null;
Long pageLimit = minForLimit(capRemaining, chunkSize);
+ requestedLimit = pageLimit;
CompletableFuture> fetch = fetcher.fetch(offset, pageLimit);
inFlight = fetch;
fetch.whenComplete(
@@ -231,39 +235,19 @@ private void applyPage(PaginationList page) {
buffer = page.getItems();
pos = 0;
int count = buffer.size();
- // Advance the offset (and decide exhaustion) by the number of rows the API scanned, not the
- // number of items it returned: an endpoint with item-level filtering (dataset items' clean /
- // skipEmpty / skipHidden) can scan up to the requested page size while returning fewer rows,
- // or none at all. Using the returned count here would re-scan (and re-yield) already-seen
- // rows, or stop iteration early on a page that filtered out everything but still had more
- // data ahead.
- //
- // getScannedCount() is null on every endpoint that does not report the header at all
- // (everything but dataset items). Where it is reported, a *nonzero* value is always trusted:
- // the returned count can legitimately fall on either side of it (a filter drops rows, so
- // returned < scanned; unwind splits one scanned row's array field into several output items,
- // so returned > scanned), and the scanned count is what offset must advance by in both cases.
- // A reported *zero* is trusted only when the page also returned nothing: scanning zero rows
- // can never produce items, so "0 scanned, N>0 returned" is not a real answer the API can give
- // - it means the header is not actually populated for this request yet (observed in practice
- // as a flat 0 on every page of an unfiltered, non-unwound live-API request that did return
- // real items), and falling back to the returned count there keeps this exactly as safe as
- // before the header existed.
- Long reported = page.getScannedCount();
- long scanned = reported != null && (reported > 0 || count == 0) ? reported : count;
- offset += scanned;
- yielded += count;
- // Defensively trim the last page to the cap in case the server returned more than requested,
- // so the subscriber never sees more than `totalLimit` items.
- if (totalLimit != null && yielded > totalLimit) {
- buffer = buffer.subList(0, count - (int) (yielded - totalLimit));
- yielded = totalLimit;
+ // The rows the page covered, which are not the items it returned: see
+ // PaginationList#getCount. Capped at the rows requested, so an unwound page does not advance
+ // past rows the next request would never read.
+ long scanned = Math.max(page.getCount(), count);
+ if (requestedLimit != null) {
+ scanned = Math.min(scanned, requestedLimit);
}
- // Stop on the cap or a page that scanned nothing (the source is exhausted); never on a short
+ offset += scanned;
+ rowsRead += scanned;
+ // Stop on the cap or a page that covered nothing (the source is exhausted); never on a short
// *returned* page (the API clamps large page sizes, and filtering can legitimately return
- // fewer items than were scanned) or the reported total (unreliable on some endpoints — see
- // class doc).
- if (scanned == 0 || (totalLimit != null && yielded >= totalLimit)) {
+ // fewer items than the rows covered) or the reported total (unreliable on some endpoints).
+ if (scanned == 0 || (totalLimit != null && rowsRead >= totalLimit)) {
exhausted = true;
}
}
diff --git a/src/main/java/com/apify/client/internal/HttpClientCore.java b/src/main/java/com/apify/client/internal/HttpClientCore.java
index af6b17a..983540a 100644
--- a/src/main/java/com/apify/client/internal/HttpClientCore.java
+++ b/src/main/java/com/apify/client/internal/HttpClientCore.java
@@ -28,6 +28,7 @@
import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -297,7 +298,7 @@ public CompletableFuture call(
return call(method, url, body, contentType, extraHeaders, timeout, false);
}
- /** The canonical overload every other {@code call} convenience overload delegates to. */
+ /** As the canonical overload, for a call to the Apify API. */
public CompletableFuture call(
String method,
String url,
@@ -306,22 +307,39 @@ public CompletableFuture call(
Map extraHeaders,
Timeout timeout,
boolean doNotRetryTimeouts) {
+ return call(
+ method,
+ url,
+ body,
+ contentType,
+ extraHeaders,
+ timeout,
+ doNotRetryTimeouts,
+ RequestPolicy.API);
+ }
+
+ /**
+ * The canonical overload every other {@code call} convenience overload delegates to. {@code
+ * policy} says how this request differs from a call to the Apify API.
+ */
+ public CompletableFuture call(
+ String method,
+ String url,
+ byte[] body,
+ String contentType,
+ Map extraHeaders,
+ Timeout timeout,
+ boolean doNotRetryTimeouts,
+ RequestPolicy policy) {
Duration baseTimeout = resolveTimeout(timeout);
// Compress the body once, up front, so every retry reuses the same encoded payload.
- byte[] requestBody = body;
- Map headers = extraHeaders;
- if (shouldCompress(requestBody, contentType, extraHeaders)) {
- try {
- requestBody = compressor.compress(requestBody);
- } catch (Exception | LinkageError e) {
- // No fallback to another coding: the configured compressor is the only one that runs. A
- // LinkageError is what a native codec that fails to load raises.
- return failedFuture(new ApifyTransportException(e));
- }
- headers = new LinkedHashMap<>(extraHeaders == null ? Map.of() : extraHeaders);
- headers.put(HttpHeaders.CONTENT_ENCODING, compressor.contentEncoding());
+ PreparedBody prepared;
+ try {
+ prepared = prepareBody(body, contentType, extraHeaders, policy);
+ } catch (ApifyTransportException e) {
+ return failedFuture(e);
}
String path = extractPath(url);
@@ -329,17 +347,48 @@ public CompletableFuture call(
return attemptWithRetry(
method,
url,
- requestBody,
+ prepared.body(),
contentType,
- headers,
+ prepared.headers(),
baseTimeout,
doNotRetryTimeouts,
+ policy,
path,
1,
maxAttempts,
retry.minDelayBetweenRetries);
}
+ /** A request body, as it is sent, with the headers that go with it. */
+ private record PreparedBody(byte[] body, Map headers) {}
+
+ /**
+ * Compresses {@code body} when the policy and the body call for it, and adds the matching {@code
+ * Content-Encoding} header.
+ *
+ * @throws ApifyTransportException if the compressor fails
+ */
+ private PreparedBody prepareBody(
+ byte[] body, String contentType, Map extraHeaders, RequestPolicy policy) {
+ HttpCompressor requestCompressor =
+ policy.compressor() != null ? policy.compressor() : compressor;
+ if (!policy.compress() || !shouldCompress(body, contentType, extraHeaders)) {
+ return new PreparedBody(body, extraHeaders);
+ }
+ byte[] compressed;
+ try {
+ compressed = requestCompressor.compress(body);
+ } catch (Exception | LinkageError e) {
+ // No fallback to another coding: the configured compressor is the only one that runs. A
+ // LinkageError is what a native codec that fails to load raises.
+ throw new ApifyTransportException(e);
+ }
+ Map headers =
+ new LinkedHashMap<>(extraHeaders == null ? Map.of() : extraHeaders);
+ headers.put(HttpHeaders.CONTENT_ENCODING, requestCompressor.contentEncoding());
+ return new PreparedBody(compressed, headers);
+ }
+
/**
* Runs one attempt and, on a retryable failure that has not exhausted {@code maxAttempts},
* schedules the next one after a backoff delay - chained via {@link
@@ -354,13 +403,24 @@ private CompletableFuture attemptWithRetry(
Map headers,
Duration baseTimeout,
boolean doNotRetryTimeouts,
+ RequestPolicy policy,
String path,
int attempt,
int maxAttempts,
Duration delay) {
Duration timeout = attemptTimeout(baseTimeout, attempt);
return attempt(
- method, url, body, contentType, headers, timeout, doNotRetryTimeouts, attempt, path)
+ method,
+ url,
+ body,
+ contentType,
+ headers,
+ timeout,
+ doNotRetryTimeouts,
+ policy,
+ attempt,
+ attempt == maxAttempts,
+ path)
.thenCompose(
outcome -> {
if (outcome.success()) {
@@ -383,6 +443,7 @@ private CompletableFuture attemptWithRetry(
headers,
baseTimeout,
doNotRetryTimeouts,
+ policy,
path,
attempt + 1,
maxAttempts,
@@ -403,19 +464,26 @@ private CompletableFuture attempt(
Map extraHeaders,
Duration timeout,
boolean doNotRetryTimeouts,
+ RequestPolicy policy,
int attemptNum,
+ boolean lastAttempt,
String path) {
- return doAttempt(method, url, body, contentType, extraHeaders, timeout)
+ return doAttempt(method, url, body, contentType, extraHeaders, timeout, policy)
.handle(
(resp, err) -> {
if (err == null) {
if (resp.statusCode() < MAX_SUCCESS_STATUS) {
return new AttemptOutcome(resp, true, false, null);
}
+ boolean retryable = isStatusRetryable(resp.statusCode(), policy);
+ // A request that does not throw hands the error response back once no retry
+ // follows.
+ if (!policy.throwOnErrorStatus() && (!retryable || lastAttempt)) {
+ return new AttemptOutcome(resp, true, false, null);
+ }
RuntimeException apiError =
buildApiError(resp.statusCode(), resp.body(), attemptNum, method, path);
- return new AttemptOutcome(
- null, false, isStatusRetryable(resp.statusCode()), apiError);
+ return new AttemptOutcome(null, false, retryable, apiError);
}
Throwable cause = unwrapCompletion(err);
if (cause instanceof CancellationException cancellation) {
@@ -427,8 +495,10 @@ private CompletableFuture attempt(
? apifyTransportException
: new ApifyTransportException(cause);
// Network/timeout failures are retryable, unless the caller opted out of retrying
- // timeouts.
- boolean retryable = !(doNotRetryTimeouts && transportError.isTimeout());
+ // timeouts or the request retries no transport errors at all.
+ boolean retryable =
+ policy.retryTransportErrors()
+ && !(doNotRetryTimeouts && transportError.isTimeout());
return new AttemptOutcome(null, false, retryable, transportError);
});
}
@@ -443,6 +513,21 @@ HttpRequest buildRequest(
String contentType,
Map extraHeaders,
Duration timeout) {
+ return buildRequest(method, url, body, contentType, extraHeaders, timeout, RequestPolicy.API);
+ }
+
+ /**
+ * As {@link #buildRequest(String, String, byte[], String, Map, Duration)}, with the token in the
+ * header the {@code policy} names.
+ */
+ private HttpRequest buildRequest(
+ String method,
+ String url,
+ byte[] body,
+ String contentType,
+ Map extraHeaders,
+ Duration timeout,
+ RequestPolicy policy) {
HttpRequest.BodyPublisher publisher =
body != null
? HttpRequest.BodyPublishers.ofByteArray(body)
@@ -451,31 +536,51 @@ HttpRequest buildRequest(
if (timeout != null) {
b.timeout(timeout);
}
- b.header(HttpHeaders.USER_AGENT, userAgent);
+ // Header names are case-insensitive, so a default header wins over the built-in one of the same
+ // name whatever its case, and a header of the request itself wins over both.
+ Map headers = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ headers.put(HttpHeaders.USER_AGENT, userAgent);
if (token != null && !token.isEmpty()) {
- b.header(HttpHeaders.AUTHORIZATION, AUTH_BEARER_PREFIX + token);
+ headers.put(HttpHeaders.AUTHORIZATION, AUTH_BEARER_PREFIX + token);
}
if (contentType != null && !contentType.isEmpty()) {
- b.header(HttpHeaders.CONTENT_TYPE, contentType);
+ headers.put(HttpHeaders.CONTENT_TYPE, contentType);
}
- // setHeader replaces what is already set, so a default header wins over the built-in ones and
- // a header of the request itself wins over both.
- defaultHeaders.forEach(b::setHeader);
+ headers.putAll(defaultHeaders);
+ moveAuthorization(headers, policy.authorizationHeader());
if (extraHeaders != null) {
- extraHeaders.forEach(b::setHeader);
+ headers.putAll(extraHeaders);
}
+ headers.forEach(b::setHeader);
return b.build();
}
+ /**
+ * Moves the {@code Authorization} header of {@code headers}, the client's token or the default
+ * header that replaces it, to the header named {@code name}. Leaves the headers as they are when
+ * {@code name} is {@code Authorization} or there is no such header.
+ */
+ private static void moveAuthorization(Map headers, String name) {
+ if (HttpHeaders.AUTHORIZATION.equalsIgnoreCase(name)) {
+ return;
+ }
+ String authorization = headers.remove(HttpHeaders.AUTHORIZATION);
+ if (authorization != null) {
+ headers.put(name, authorization);
+ }
+ }
+
private CompletableFuture doAttempt(
String method,
String url,
byte[] body,
String contentType,
Map extraHeaders,
- Duration timeout) {
- HttpRequest request = buildRequest(method, url, body, contentType, extraHeaders, timeout);
- return CancellableFuture.wrap(transport.sendAsync(request))
+ Duration timeout,
+ RequestPolicy policy) {
+ HttpRequest request =
+ buildRequest(method, url, body, contentType, extraHeaders, timeout, policy);
+ return CancellableFuture.wrap(transport.sendAsync(request, policy.followRedirects()))
.thenApply(resp -> new ApiResponse(resp.statusCode(), resp.headers(), resp.body()));
}
@@ -613,7 +718,10 @@ public static boolean isCompressibleContentType(String contentType) {
return true;
}
- private static boolean isStatusRetryable(int status) {
+ private static boolean isStatusRetryable(int status, RequestPolicy policy) {
+ if (policy.retryableStatus() != null) {
+ return policy.retryableStatus().test(status);
+ }
return status == RATE_LIMIT_EXCEEDED || status >= MIN_SERVER_ERROR;
}
@@ -632,8 +740,8 @@ private static Duration minDuration(Duration a, Duration b) {
return a.compareTo(b) < 0 ? a : b;
}
- private static CompletableFuture failedFuture(RuntimeException error) {
- CompletableFuture future = new CompletableFuture<>();
+ private static CompletableFuture failedFuture(RuntimeException error) {
+ CompletableFuture future = new CompletableFuture<>();
future.completeExceptionally(error);
return future;
}
@@ -737,8 +845,99 @@ public static String extractPath(String url) {
*/
public CompletableFuture> streamAsync(String url) {
HttpRequest request = buildRequest("GET", url, null, null, null, resolveTimeout(Timeout.LONG));
+ return openStream(request, true);
+ }
+
+ /**
+ * Opens a live streaming response to a request, as {@link #call(String, String, byte[], String,
+ * Map, Timeout, boolean, RequestPolicy)} would send it, but without turning an error status into
+ * an exception: every response is handed back with its body unread, for the caller to inspect and
+ * close. A status the policy retries is retried with the client's retry settings, each discarded
+ * response closed, and the last one is handed back once the retries run out. A transport error is
+ * never retried. The future completes exceptionally with an {@link ApifyTransportException} on a
+ * transport failure.
+ */
+ public CompletableFuture> streamRequest(
+ String method,
+ String url,
+ byte[] body,
+ String contentType,
+ Map extraHeaders,
+ Timeout timeout,
+ RequestPolicy policy) {
+ PreparedBody prepared;
+ try {
+ prepared = prepareBody(body, contentType, extraHeaders, policy);
+ } catch (ApifyTransportException e) {
+ return failedFuture(e);
+ }
+ return streamAttempt(
+ method,
+ url,
+ prepared,
+ contentType,
+ resolveTimeout(timeout),
+ policy,
+ 1,
+ retry.maxRetries + 1,
+ retry.minDelayBetweenRetries);
+ }
+
+ /**
+ * Opens one attempt of {@link #streamRequest} and, when the response has a status the policy
+ * retries and attempts are left, closes it and schedules the next attempt after a backoff delay.
+ */
+ private CompletableFuture> streamAttempt(
+ String method,
+ String url,
+ PreparedBody prepared,
+ String contentType,
+ Duration baseTimeout,
+ RequestPolicy policy,
+ int attempt,
+ int maxAttempts,
+ Duration delay) {
+ HttpRequest request =
+ buildRequest(
+ method,
+ url,
+ prepared.body(),
+ contentType,
+ prepared.headers(),
+ attemptTimeout(baseTimeout, attempt),
+ policy);
+ return openStream(request, policy.followRedirects())
+ .thenCompose(
+ response -> {
+ boolean retryable =
+ isStatusRetryable(response.statusCode(), policy) && attempt < maxAttempts;
+ if (!retryable) {
+ return CompletableFuture.completedFuture(response);
+ }
+ closeQuietly(response.body());
+ Duration nextDelay =
+ minDuration(delay.multipliedBy(BACKOFF_FACTOR), retry.maxDelayBetweenRetries);
+ return Async.delay(randomizedDelay(delay))
+ .thenCompose(
+ unused ->
+ streamAttempt(
+ method,
+ url,
+ prepared,
+ contentType,
+ baseTimeout,
+ policy,
+ attempt + 1,
+ maxAttempts,
+ nextDelay));
+ });
+ }
+
+ private CompletableFuture> openStream(
+ HttpRequest request, boolean followRedirects) {
CancellableFuture> result = new CancellableFuture<>();
- CompletableFuture> exchange = transport.sendStreamingAsync(request);
+ CompletableFuture> exchange =
+ transport.sendStreamingAsync(request, followRedirects);
// Cancelling the returned future aborts the exchange.
result.link(exchange);
exchange.whenComplete(
diff --git a/src/main/java/com/apify/client/internal/HttpHeaders.java b/src/main/java/com/apify/client/internal/HttpHeaders.java
index 552d4aa..d478fb3 100644
--- a/src/main/java/com/apify/client/internal/HttpHeaders.java
+++ b/src/main/java/com/apify/client/internal/HttpHeaders.java
@@ -13,6 +13,13 @@ public final class HttpHeaders {
/** The bearer-token authentication header name. */
public static final String AUTHORIZATION = "Authorization";
+ /**
+ * The header that carries the client's token on a request to an Actor's server. The platform
+ * consumes it before the request reaches the Actor's code, which keeps {@link #AUTHORIZATION}
+ * free for the Actor's own authentication.
+ */
+ public static final String X_APIFY_AUTHORIZATION = "X-Apify-Authorization";
+
/** The request/response body media-type header name. */
public static final String CONTENT_TYPE = "Content-Type";
diff --git a/src/main/java/com/apify/client/internal/RequestPolicy.java b/src/main/java/com/apify/client/internal/RequestPolicy.java
new file mode 100644
index 0000000..51a819c
--- /dev/null
+++ b/src/main/java/com/apify/client/internal/RequestPolicy.java
@@ -0,0 +1,74 @@
+package com.apify.client.internal;
+
+import com.apify.client.http.HttpCompressor;
+import java.util.Set;
+import java.util.function.IntPredicate;
+
+/**
+ * How {@link HttpClientCore} treats a single request, where it differs from a call to the Apify
+ * API: which header carries the token, whether an error status throws, which statuses and transport
+ * errors are retried, whether redirects are followed and how the body is compressed. Internal to
+ * the client.
+ *
+ * @param authorizationHeader the name of the header that carries the client's token
+ * @param throwOnErrorStatus whether an error status that is not retried any more is thrown as an
+ * {@link com.apify.client.http.ApifyApiException}; when {@code false} it is returned as the
+ * response instead
+ * @param retryableStatus tells which error statuses are retried, or {@code null} for rate limits
+ * (429) and server errors (5xx)
+ * @param retryTransportErrors whether a transport error, a timeout included, is retried
+ * @param followRedirects whether the transport may follow redirects
+ * @param compress whether the request body is compressed at all
+ * @param compressor the compressor for this request, or {@code null} for the client's own; only
+ * used when {@code compress} is {@code true}
+ */
+public record RequestPolicy(
+ String authorizationHeader,
+ boolean throwOnErrorStatus,
+ IntPredicate retryableStatus,
+ boolean retryTransportErrors,
+ boolean followRedirects,
+ boolean compress,
+ HttpCompressor compressor) {
+
+ /**
+ * Statuses a request to an Actor's server is retried on. The platform answers with them when it
+ * cannot pass a request to a server run, so a retry usually does not run the Actor's code again.
+ * An Actor that returns one itself gets the request again.
+ */
+ private static final Set SERVER_RETRY_STATUS_CODES = Set.of(429, 503);
+
+ /** A call to the Apify API. */
+ public static final RequestPolicy API =
+ new RequestPolicy(
+ /* authorizationHeader= */ HttpHeaders.AUTHORIZATION,
+ /* throwOnErrorStatus= */ true,
+ /* retryableStatus= */ null,
+ /* retryTransportErrors= */ true,
+ /* followRedirects= */ true,
+ /* compress= */ true,
+ /* compressor= */ null);
+
+ /**
+ * A request to an Actor's server. The token travels in the {@code X-Apify-Authorization} header,
+ * which the platform removes before the request reaches the Actor, so the {@code Authorization}
+ * header stays free for the Actor's own authentication. Every response is returned, error
+ * statuses included. Only the statuses in {@link #SERVER_RETRY_STATUS_CODES} are retried, and no
+ * transport error is, since the request may have reached the Actor before it failed. Redirects
+ * are not followed, since that would send the token to a host the Actor's code picked. The body
+ * is sent uncompressed unless a {@code compressor} is given, since the Actor's server may not
+ * decode a compressed body.
+ *
+ * @param compressor compresses the request body, or {@code null} to send it uncompressed
+ */
+ public static RequestPolicy server(HttpCompressor compressor) {
+ return new RequestPolicy(
+ /* authorizationHeader= */ HttpHeaders.X_APIFY_AUTHORIZATION,
+ /* throwOnErrorStatus= */ false,
+ /* retryableStatus= */ SERVER_RETRY_STATUS_CODES::contains,
+ /* retryTransportErrors= */ false,
+ /* followRedirects= */ false,
+ /* compress= */ compressor != null,
+ /* compressor= */ compressor);
+ }
+}
diff --git a/src/main/java/com/apify/client/internal/ResourceContext.java b/src/main/java/com/apify/client/internal/ResourceContext.java
index d1014be..08c0a12 100644
--- a/src/main/java/com/apify/client/internal/ResourceContext.java
+++ b/src/main/java/com/apify/client/internal/ResourceContext.java
@@ -323,8 +323,8 @@ public CompletableFuture> listResource(
* Builds a lazy, backpressure-aware {@link Flow.Publisher} over an offset/limit-paginated
* endpoint. {@code applyFilters} adds the per-endpoint filter params (everything except {@code
* offset}/{@code limit}, which the publisher drives per page). {@code totalLimit} caps the total
- * items yielded; {@code chunkSize} is the page size (both {@code null} meaning "unbounded" /
- * "server default").
+ * rows read (see {@link AsyncPaginatedPublisher}); {@code chunkSize} is the page size (both
+ * {@code null} meaning "unbounded" / "server default").
*/
public Flow.Publisher iterateResource(
String subPath,
diff --git a/src/main/java/com/apify/client/internal/Signatures.java b/src/main/java/com/apify/client/internal/Signatures.java
index 3d4d25d..986fb00 100644
--- a/src/main/java/com/apify/client/internal/Signatures.java
+++ b/src/main/java/com/apify/client/internal/Signatures.java
@@ -26,6 +26,15 @@ public final class Signatures {
private static final BigInteger BASE62 = BigInteger.valueOf(BASE62_ALPHABET.length());
+ private static final long MILLIS_PER_SECOND = 1000L;
+
+ /**
+ * Longest validity of a signed URL in seconds ({@code 4611686018427387}, about 146 million
+ * years). It is half of what fits in milliseconds in a {@code long}, so the current time plus the
+ * validity cannot overflow.
+ */
+ public static final long MAX_EXPIRES_IN_SECS = Long.MAX_VALUE / MILLIS_PER_SECOND / 2;
+
private Signatures() {}
/**
@@ -68,11 +77,17 @@ static String toBase62(BigInteger value) {
* #createHmacSignature}, then returns the base64url (no padding) encoding of {@code
* "{version}.{expiresAtMillis}.{hmac}"}.
*
+ * The expiry counts from this call, so call it after the storage lookup; otherwise the lookup
+ * time comes off the validity.
+ *
* @param expiresInSecs optional expiry in seconds ({@code null} for a non-expiring URL)
+ * @throws IllegalArgumentException if {@code expiresInSecs} fails {@link
+ * #requireValidExpiresInSecs}
*/
public static String signStorageContent(String secretKey, String resourceId, Long expiresInSecs) {
+ requireValidExpiresInSecs(expiresInSecs);
long expiresAtMillis =
- expiresInSecs != null ? System.currentTimeMillis() + expiresInSecs * 1000L : 0L;
+ expiresInSecs != null ? System.currentTimeMillis() + expiresInSecs * MILLIS_PER_SECOND : 0L;
String version = STORAGE_CONTENT_SIGNATURE_VERSION;
String message = version + "." + expiresAtMillis + "." + resourceId;
String hmac = createHmacSignature(secretKey, message);
@@ -82,6 +97,29 @@ public static String signStorageContent(String secretKey, String resourceId, Lon
.encodeToString(envelope.getBytes(StandardCharsets.UTF_8));
}
+ /**
+ * Checks the validity of a signed URL. {@code null} is a URL that does not expire. A value must
+ * be positive, because {@code 0} or a negative value would sign a URL that has already expired,
+ * and at most {@link #MAX_EXPIRES_IN_SECS}.
+ *
+ * @throws IllegalArgumentException if the value is out of range
+ */
+ public static void requireValidExpiresInSecs(Long expiresInSecs) {
+ if (expiresInSecs == null) {
+ return;
+ }
+ if (expiresInSecs <= 0) {
+ throw new IllegalArgumentException(
+ "expiresInSecs must be positive, got "
+ + expiresInSecs
+ + "; use null for a URL that does not expire");
+ }
+ if (expiresInSecs > MAX_EXPIRES_IN_SECS) {
+ throw new IllegalArgumentException(
+ "expiresInSecs must be at most " + MAX_EXPIRES_IN_SECS + ", got " + expiresInSecs);
+ }
+ }
+
private static byte[] hmacSha256(String secretKey, String message) {
try {
Mac mac = Mac.getInstance("HmacSHA256");
diff --git a/src/main/java/com/apify/client/keyvalue/KeyValueStoreClient.java b/src/main/java/com/apify/client/keyvalue/KeyValueStoreClient.java
index b65c1b9..054024f 100644
--- a/src/main/java/com/apify/client/keyvalue/KeyValueStoreClient.java
+++ b/src/main/java/com/apify/client/keyvalue/KeyValueStoreClient.java
@@ -382,7 +382,11 @@ public CompletableFuture getRecordPublicUrl(String key) {
/**
* Builds a public URL for listing this store's keys. As with {@link #getRecordPublicUrl}, a
* signature is appended for private stores. {@code expiresInSecs} optionally bounds the validity
- * of a signed URL ({@code null} for non-expiring).
+ * of a signed URL ({@code null} for non-expiring); it must be positive and at most
+ * 4611686018427387.
+ *
+ * @throws IllegalArgumentException before any request is sent, if {@code expiresInSecs} is not
+ * {@code null} and is not positive or exceeds that maximum
*/
public CompletableFuture createKeysPublicUrl(Long expiresInSecs) {
return createKeysPublicUrl(new ListKeysOptions(), expiresInSecs);
@@ -393,10 +397,14 @@ public CompletableFuture createKeysPublicUrl(Long expiresInSecs) {
* ({@code limit}, {@code prefix}, {@code collection}, {@code exclusiveStartKey}) into the URL. As
* with {@link #getRecordPublicUrl}, a signature is appended for private stores unless the caller
* already supplied one. {@code expiresInSecs} optionally bounds the validity of a signed URL
- * ({@code null} for non-expiring).
+ * ({@code null} for non-expiring); it must be positive and at most 4611686018427387.
+ *
+ * @throws IllegalArgumentException before any request is sent, if {@code expiresInSecs} is not
+ * {@code null} and is not positive or exceeds that maximum
*/
public CompletableFuture createKeysPublicUrl(
ListKeysOptions options, Long expiresInSecs) {
+ Signatures.requireValidExpiresInSecs(expiresInSecs);
QueryParams params = new QueryParams();
options.apply(params);
if (options.signatureValue() != null) {
diff --git a/src/main/java/com/apify/client/requestqueue/RequestQueueClient.java b/src/main/java/com/apify/client/requestqueue/RequestQueueClient.java
index b87200b..5c70c7d 100644
--- a/src/main/java/com/apify/client/requestqueue/RequestQueueClient.java
+++ b/src/main/java/com/apify/client/requestqueue/RequestQueueClient.java
@@ -778,9 +778,9 @@ private void applyPage(RequestsList page, Long capRemaining) {
buffer = page.getItems();
pos = 0;
nextCursor = page.getNextCursor();
- // Defensively trim to the cap in case the server returned more than requested, matching
- // AsyncPaginatedPublisher's same guard, so a subscriber never sees more than `totalLimit`
- // items even if the server ignores (or overshoots) the requested per-page `limit`.
+ // Defensively trim to the cap in case the server returned more than requested, so a
+ // subscriber never sees more than `totalLimit` items even if the server ignores (or
+ // overshoots) the requested per-page `limit`.
if (capRemaining != null && buffer.size() > capRemaining) {
buffer = buffer.subList(0, capRemaining.intValue());
}
diff --git a/src/main/java/com/apify/client/run/RunDatasetItems.java b/src/main/java/com/apify/client/run/RunDatasetItems.java
index aae86c4..a31438d 100644
--- a/src/main/java/com/apify/client/run/RunDatasetItems.java
+++ b/src/main/java/com/apify/client/run/RunDatasetItems.java
@@ -67,8 +67,8 @@ private enum Phase {
}
/**
- * Reads the next page. Completes with a page whose scanned count is non-zero while there may be
- * more rows, and with an empty page that scanned nothing once iteration is over.
+ * Reads the next page. Completes with a page whose count is non-zero while there may be more
+ * rows, and with an empty page of count zero once iteration is over.
*/
CompletableFuture> next() {
switch (phase) {
@@ -131,17 +131,16 @@ private CompletableFuture> tail() {
return page(position, limit)
.thenCompose(
page -> {
- Long scanned = page.getScannedCount();
- boolean empty = page.getItems().isEmpty() && (scanned == null || scanned == 0);
- if (!empty) {
+ // Only a page that covered no rows marks the end, as filters can shorten a full one.
+ if (page.getCount() > 0) {
return CompletableFuture.completedFuture(advance(page, limit));
}
if (!access.dropsRows(options)) {
return finish();
}
- // A page that clean, skipEmpty or unwind emptied past a lagging itemCount reports no
- // scanned rows either, so a read of the next row without those filters (they would
- // empty it again) tells whether any rows are left.
+ // A page that clean, skipEmpty or unwind emptied past a lagging itemCount covers no
+ // rows either, so a read of the next row without those filters (they would empty it
+ // again) tells whether any rows are left.
return dataset
.listItems(access.plainPage(position, 1L), JsonNode.class)
.thenCompose(
@@ -165,20 +164,20 @@ private CompletableFuture> page(long offset, long limit) {
return dataset.listItems(access.page(options, offset, limit), itemClass);
}
- /** Moves past the {@code limit} rows a page covered and marks it as having scanned them. */
+ /** Moves past the {@code limit} rows a page covered and marks it as having covered them. */
private PaginationList advance(PaginationList page, long limit) {
position += limit;
if (end != null && position >= end && phase == Phase.TAIL) {
phase = Phase.DONE;
}
- page.setScannedCount(Math.max(limit, 1));
+ page.setCount(Math.max(limit, 1));
return page;
}
private PaginationList endOfData() {
PaginationList page = new PaginationList<>();
page.setItems(List.of());
- page.setScannedCount(0L);
+ page.setCount(0);
return page;
}
}
diff --git a/src/test/java/com/apify/client/ActorServerClientTest.java b/src/test/java/com/apify/client/ActorServerClientTest.java
new file mode 100644
index 0000000..2bb7ebb
--- /dev/null
+++ b/src/test/java/com/apify/client/ActorServerClientTest.java
@@ -0,0 +1,985 @@
+package com.apify.client;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.apify.client.actor.ActorClient;
+import com.apify.client.actor.ActorServerClient;
+import com.apify.client.actor.ActorServerRequestOptions;
+import com.apify.client.actor.ActorServerStreamResponse;
+import com.apify.client.http.ApiResponse;
+import com.apify.client.http.ApifyTransportException;
+import com.apify.client.http.DefaultHttpTransport;
+import com.apify.client.http.GzipHttpCompressor;
+import com.apify.client.http.HttpTransport;
+import com.apify.client.http.NotFoundError;
+import com.apify.client.http.ServerActorException;
+import com.apify.client.http.Timeout;
+import com.apify.client.internal.HttpClientCore;
+import com.apify.client.internal.Json;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import tools.jackson.databind.JsonNode;
+
+/**
+ * Tests of {@link ActorServerClient} against a local HTTP server that plays both the Apify API and
+ * the Actor's server, so the headers, redirects and dropped connections are real network behavior.
+ * Token-free.
+ */
+class ActorServerClientTest {
+
+ private static final String ACTOR_ID = "test-actor-id";
+ private static final String ACTOR_PATH = "/v2/actors/" + ACTOR_ID;
+ private static final String SERVER_PREFIX = "/server";
+ private static final JsonNode OPENAPI_SPEC =
+ Json.toTree(Map.of("openapi", "3.0.0", "paths", Map.of()));
+ private static final String TOKEN = "test-token";
+
+ private TestHttpServer server;
+
+ /** The Actor the API returns; null answers 404, as for an Actor that does not exist. */
+ private Map actor;
+
+ /** Build numbers of the Actor's builds by build id, newest first, as the builds list has them. */
+ private Map builds;
+
+ /** How many builds the builds list returns per page. */
+ private int buildsPerPage;
+
+ /** The {@code actorDefinition} every build the API returns has; null for none. */
+ private Map actorDefinition;
+
+ /** Statuses the server answers {@code /flaky} with, one per request, before it answers 200. */
+ private final List flakyStatuses = Collections.synchronizedList(new ArrayList<>());
+
+ @BeforeEach
+ void startServer() throws IOException {
+ server = new TestHttpServer(this::answer);
+ actor = serverActor(Map.of());
+ builds = new LinkedHashMap<>();
+ buildsPerPage = Integer.MAX_VALUE;
+ actorDefinition = Map.of("webServerSchema", Json.toTree(OPENAPI_SPEC));
+ }
+
+ @AfterEach
+ void stopServer() throws IOException {
+ server.close();
+ }
+
+ private Map serverActor(Map fields) {
+ Map result = new LinkedHashMap<>();
+ result.put("id", ACTOR_ID);
+ result.put("name", "test-actor");
+ result.put("actorStandby", Map.of("isEnabled", true));
+ result.put(
+ "taggedBuilds",
+ Map.of(
+ "latest", Map.of("buildId", "latest-build-id", "buildNumber", "0.0.2"),
+ "beta", Map.of("buildId", "beta-build-id", "buildNumber", "0.0.3")));
+ result.put("standbyUrl", server.url() + SERVER_PREFIX);
+ result.putAll(fields);
+ return result;
+ }
+
+ private TestHttpServer.Reply answer(TestHttpServer.Received request) {
+ String path = request.path();
+ if (path.equals(ACTOR_PATH)) {
+ return actor == null
+ ? TestHttpServer.Reply.json(
+ 404,
+ "{\"error\":{\"type\":\"record-not-found\",\"message\":\"Actor was not found\"}}")
+ : data(actor);
+ }
+ if (path.equals(ACTOR_PATH + "/builds")) {
+ return buildList(request);
+ }
+ if (path.startsWith("/v2/actor-builds/")) {
+ Map build = new LinkedHashMap<>();
+ build.put("id", path.substring(path.lastIndexOf('/') + 1));
+ build.put("actorDefinition", actorDefinition);
+ return data(build);
+ }
+ if (path.equals(SERVER_PREFIX + "/flaky") && !flakyStatuses.isEmpty()) {
+ return TestHttpServer.Reply.json(flakyStatuses.remove(0), "{\"error\":\"flaky\"}");
+ }
+ if (path.equals(SERVER_PREFIX + "/hang-up")) {
+ return TestHttpServer.Reply.drop();
+ }
+ if (path.equals(SERVER_PREFIX + "/missing")) {
+ return TestHttpServer.Reply.of(404, "text/plain", "Not here");
+ }
+ return TestHttpServer.Reply.json(200, "{\"ok\":true}");
+ }
+
+ private TestHttpServer.Reply buildList(TestHttpServer.Received request) {
+ long offset = 0;
+ for (String pair : request.query().split("&")) {
+ if (pair.startsWith("offset=")) {
+ offset = Long.parseLong(pair.substring("offset=".length()));
+ }
+ }
+ List> all = new ArrayList<>();
+ builds.forEach((id, number) -> all.add(Map.of("id", id, "buildNumber", number)));
+ int from = (int) Math.min(offset, all.size());
+ List> page =
+ all.subList(from, (int) Math.min(all.size(), (long) from + buildsPerPage));
+ return data(
+ Map.of(
+ "total",
+ all.size(),
+ "offset",
+ offset,
+ "limit",
+ 1000,
+ "count",
+ page.size(),
+ "items",
+ page));
+ }
+
+ private static TestHttpServer.Reply data(Object payload) {
+ return TestHttpServer.Reply.json(
+ 200, new String(Json.toBytes(Map.of("data", payload)), StandardCharsets.UTF_8));
+ }
+
+ private ApifyClient client(int maxRetries) {
+ return ApifyClient.builder()
+ .token(TOKEN)
+ .baseUrl(server.url())
+ .maxRetries(maxRetries)
+ .minDelayBetweenRetries(Duration.ofMillis(1))
+ .build();
+ }
+
+ private ApifyClient client() {
+ return client(0);
+ }
+
+ private List serverRequests() {
+ return server.received().stream()
+ .filter(request -> request.path().startsWith(SERVER_PREFIX + "/"))
+ .toList();
+ }
+
+ private long actorGets() {
+ return server.received().stream().filter(request -> request.path().equals(ACTOR_PATH)).count();
+ }
+
+ private static Throwable rootOf(Throwable error) {
+ Throwable current = error;
+ while (current instanceof CompletionException && current.getCause() != null) {
+ current = current.getCause();
+ }
+ return current;
+ }
+
+ private static T failureOf(Class type, CompletableFuture future) {
+ Throwable error = assertThrows(CompletionException.class, future::join);
+ return assertInstanceOf(type, rootOf(error));
+ }
+
+ // ---- request -----------------------------------------------------------------------------
+
+ @Test
+ void requestSendsTheMethodPathQueryBodyAndTokenToTheServerUrl() {
+ ApiResponse response =
+ client()
+ .actor(ACTOR_ID)
+ .server()
+ .request(
+ "/search",
+ new ActorServerRequestOptions()
+ .method("POST")
+ .param("q", "apify")
+ .body(Map.of("limit", 5)))
+ .join();
+
+ assertEquals(200, response.statusCode());
+ assertEquals(Json.toTree(Map.of("ok", true)), response.bodyAs(JsonNode.class));
+ assertEquals(1, serverRequests().size());
+ TestHttpServer.Received request = serverRequests().get(0);
+ assertEquals("POST", request.method());
+ assertEquals(SERVER_PREFIX + "/search", request.path());
+ assertEquals("q=apify", request.query());
+ assertEquals(Json.toTree(Map.of("limit", 5)), Json.parse(request.body(), JsonNode.class));
+ assertEquals("application/json", request.header("Content-Type"));
+ assertEquals("Bearer " + TOKEN, request.header("X-Apify-Authorization"));
+ assertNull(request.header("Authorization"));
+ }
+
+ @Test
+ void requestSendsAnAuthorizationHeaderOfTheCallerAsItIs() {
+ client()
+ .actor(ACTOR_ID)
+ .server()
+ .request("/", new ActorServerRequestOptions().header("Authorization", "Bearer actor-token"))
+ .join();
+
+ TestHttpServer.Received request = serverRequests().get(0);
+ assertEquals("Bearer actor-token", request.header("Authorization"));
+ assertEquals("Bearer " + TOKEN, request.header("X-Apify-Authorization"));
+ }
+
+ @Test
+ void requestReturnsARedirectOfTheServerAsItIsSoTheTokenNeverReachesItsTarget()
+ throws IOException {
+ for (HttpTransport transport :
+ List.of(
+ new DefaultHttpTransport(),
+ new DefaultHttpTransport(
+ HttpClient.newBuilder().followRedirects(HttpClient.Redirect.ALWAYS).build()))) {
+ AtomicBoolean redirected = new AtomicBoolean();
+ try (TestHttpServer other =
+ new TestHttpServer(
+ request -> {
+ redirected.set(true);
+ return TestHttpServer.Reply.json(200, "{}");
+ });
+ TestHttpServer redirecting =
+ new TestHttpServer(request -> TestHttpServer.Reply.redirect(other.url() + "/"))) {
+ actor = serverActor(Map.of("standbyUrl", redirecting.url()));
+ ApifyClient client =
+ ApifyClient.builder()
+ .token(TOKEN)
+ .baseUrl(server.url())
+ .httpTransport(transport)
+ .maxRetries(0)
+ .build();
+
+ ApiResponse response = client.actor(ACTOR_ID).server().request("/").join();
+
+ assertEquals(302, response.statusCode());
+ assertEquals(other.url() + "/", response.headers().firstValue("Location").orElseThrow());
+ assertFalse(redirected.get(), "the redirect target must never be contacted");
+ }
+ }
+ }
+
+ @Test
+ void requestReusesTheServerUrlFetchedByTheFirstCall() {
+ ActorClient actorClient = client().actor(ACTOR_ID);
+
+ actorClient.server().request().join();
+ actorClient.server().request().join();
+
+ assertEquals(1, actorGets());
+ assertEquals(
+ List.of(SERVER_PREFIX + "/", SERVER_PREFIX + "/"),
+ serverRequests().stream().map(TestHttpServer.Received::path).toList());
+ }
+
+ @Test
+ void getUrlReturnsTheStandbyUrlOfTheActorAndFetchesItOnlyOnce() {
+ ActorClient actorClient = client().actor(ACTOR_ID);
+
+ assertEquals(server.url() + SERVER_PREFIX, actorClient.server().getUrl().join());
+ assertEquals(server.url() + SERVER_PREFIX, actorClient.server().getUrl().join());
+
+ assertEquals(1, actorGets());
+ }
+
+ @Test
+ void getUrlFetchesTheActorOnceForConcurrentFirstCalls() {
+ ActorServerClient serverClient = client().actor(ACTOR_ID).server();
+
+ CompletableFuture url = serverClient.getUrl();
+ CompletableFuture response = serverClient.request();
+ CompletableFuture.allOf(url, response).join();
+
+ assertEquals(server.url() + SERVER_PREFIX, url.join());
+ assertEquals(1, actorGets());
+ }
+
+ @Test
+ void getUrlFetchesTheActorAgainAfterAFailedLookup() {
+ ActorServerClient serverClient = client().actor(ACTOR_ID).server();
+ Map existing = actor;
+ actor = null;
+
+ failureOf(NotFoundError.class, serverClient.getUrl());
+ actor = existing;
+
+ assertEquals(server.url() + SERVER_PREFIX, serverClient.getUrl().join());
+ assertEquals(2, actorGets());
+ }
+
+ @Test
+ void concurrentFirstCallsAllSeeTheErrorOfAFailedLookup() throws InterruptedException {
+ actor = null;
+ ActorServerClient serverClient = client().actor(ACTOR_ID).server();
+ int callers = 8;
+ List failures = Collections.synchronizedList(new ArrayList<>());
+ List threads = new ArrayList<>();
+ for (int i = 0; i < callers; i++) {
+ Thread thread =
+ new Thread(
+ () -> {
+ try {
+ serverClient.getUrl().join();
+ } catch (Throwable error) {
+ failures.add(rootOf(error));
+ }
+ });
+ threads.add(thread);
+ }
+ threads.forEach(Thread::start);
+ for (Thread thread : threads) {
+ thread.join();
+ }
+
+ assertEquals(callers, failures.size());
+ failures.forEach(failure -> assertInstanceOf(NotFoundError.class, failure));
+ }
+
+ @Test
+ void cancellingTheFutureOfGetUrlLeavesTheSharedLookupAlone() {
+ ActorServerClient serverClient = client().actor(ACTOR_ID).server();
+
+ serverClient.getUrl().cancel(true);
+
+ assertEquals(server.url() + SERVER_PREFIX, serverClient.getUrl().join());
+ }
+
+ @Test
+ void requestReusesTheServerUrlFetchedByGetOpenApiSpec() {
+ ActorServerClient serverClient = client().actor(ACTOR_ID).server();
+
+ serverClient.getOpenApiSpec().join();
+ serverClient.request().join();
+
+ assertEquals(1, actorGets());
+ }
+
+ @Test
+ void callsFailForAnActorThatIsNotAServerActor() {
+ actor = serverActor(new LinkedHashMap<>(Collections.singletonMap("standbyUrl", null)));
+ ActorServerClient serverClient = client().actor(ACTOR_ID).server();
+
+ for (CompletableFuture call :
+ List.of(serverClient.getUrl(), serverClient.getOpenApiSpec(), serverClient.request())) {
+ ServerActorException error = failureOf(ServerActorException.class, call);
+ assertTrue(error.getMessage().contains("is not a Server Actor"), error.getMessage());
+ }
+ assertEquals(0, serverRequests().size());
+ }
+
+ @Test
+ void requestFailsWithNotFoundForAnActorThatDoesNotExist() {
+ actor = null;
+
+ failureOf(NotFoundError.class, client().actor(ACTOR_ID).server().request());
+ }
+
+ @Test
+ void requestSendsAPathThatLooksLikeAnotherHostToTheServerAsAPathOnIt() {
+ for (String path :
+ List.of("//evil.example/steal", "@evil.example/steal", "https://evil.example/steal")) {
+ server.received().clear();
+
+ ApiResponse response = client().actor(ACTOR_ID).server().request(path).join();
+
+ assertEquals(200, response.statusCode(), path);
+ assertEquals(1, serverRequests().size(), path);
+ }
+ }
+
+ @Test
+ void requestDropsAFragmentInThePathSoTheParamsStillReachTheServer() {
+ client()
+ .actor(ACTOR_ID)
+ .server()
+ .request("/search#results", new ActorServerRequestOptions().param("q", "apify"))
+ .join();
+
+ TestHttpServer.Received request = serverRequests().get(0);
+ assertEquals(SERVER_PREFIX + "/search", request.path());
+ assertEquals("q=apify", request.query());
+ }
+
+ @Test
+ void requestConvertsTheQueryParametersAsForApiCalls() {
+ client()
+ .actor(ACTOR_ID)
+ .server()
+ .request(
+ "/",
+ new ActorServerRequestOptions()
+ .param("flag", true)
+ .param("off", false)
+ .param("tags", List.of("a", "b"))
+ .param("skipped", null)
+ .param("limit", 5))
+ .join();
+
+ assertEquals("flag=1&off=0&tags=a%2Cb&limit=5", serverRequests().get(0).query());
+ }
+
+ // ---- body ---------------------------------------------------------------------------------
+
+ private static Map largePayload() {
+ return Map.of("items", Collections.nCopies(100, "x".repeat(100)));
+ }
+
+ @Test
+ void requestSendsABodyLargeEnoughToCompressUncompressed() {
+ Map payload = largePayload();
+
+ client()
+ .actor(ACTOR_ID)
+ .server()
+ .request("/items", new ActorServerRequestOptions().method("POST").body(payload))
+ .join();
+
+ TestHttpServer.Received request = serverRequests().get(0);
+ assertNull(request.header("Content-Encoding"));
+ assertEquals(Json.toTree(payload), Json.parse(request.body(), JsonNode.class));
+ }
+
+ @Test
+ void requestCompressesTheBodyWhenACompressorIsSet() throws IOException {
+ Map payload = largePayload();
+
+ client()
+ .actor(ACTOR_ID)
+ .server()
+ .request(
+ "/items",
+ new ActorServerRequestOptions()
+ .method("POST")
+ .body(payload)
+ .compression(new GzipHttpCompressor(1)))
+ .join();
+
+ TestHttpServer.Received request = serverRequests().get(0);
+ assertEquals("gzip", request.header("Content-Encoding"));
+ try (GZIPInputStream in = new GZIPInputStream(new ByteArrayInputStream(request.body()))) {
+ assertEquals(Json.toTree(payload), Json.parse(in.readAllBytes(), JsonNode.class));
+ }
+ }
+
+ @Test
+ void requestSendsAContentEncodingHeaderOfTheCallerWithTheBodyUntouched() throws IOException {
+ byte[] raw = "x".repeat(2000).getBytes(StandardCharsets.UTF_8);
+ java.io.ByteArrayOutputStream compressed = new java.io.ByteArrayOutputStream();
+ try (GZIPOutputStream out = new GZIPOutputStream(compressed)) {
+ out.write(raw);
+ }
+
+ client()
+ .actor(ACTOR_ID)
+ .server()
+ .request(
+ "/items",
+ new ActorServerRequestOptions()
+ .method("POST")
+ .header("content-encoding", "gzip")
+ .body(compressed.toByteArray())
+ .compression(new GzipHttpCompressor(1)))
+ .join();
+
+ TestHttpServer.Received request = serverRequests().get(0);
+ assertEquals("gzip", request.header("Content-Encoding"));
+ assertTrue(java.util.Arrays.equals(compressed.toByteArray(), request.body()));
+ }
+
+ @Test
+ void requestSendsAStringBodyAsItIsWithTheContentTypeOfTheCaller() {
+ client()
+ .actor(ACTOR_ID)
+ .server()
+ .request(
+ "/echo",
+ new ActorServerRequestOptions()
+ .method("PUT")
+ .header("Content-Type", "text/plain")
+ .body("plain text"))
+ .join();
+
+ TestHttpServer.Received request = serverRequests().get(0);
+ assertEquals("PUT", request.method());
+ assertEquals("text/plain", request.header("Content-Type"));
+ assertEquals("plain text", request.bodyText());
+ }
+
+ @Test
+ void optionsRejectAnUnknownMethodAndATimeoutThatIsNull() {
+ ActorServerRequestOptions options = new ActorServerRequestOptions();
+
+ assertThrows(IllegalArgumentException.class, () -> options.method("get"));
+ assertThrows(IllegalArgumentException.class, () -> options.method("CONNECT"));
+ assertThrows(IllegalArgumentException.class, () -> options.method(null));
+ assertThrows(IllegalArgumentException.class, () -> options.timeout(null));
+ assertEquals(0, server.received().size());
+ }
+
+ // ---- responses and retries ----------------------------------------------------------------
+
+ @Test
+ void requestReturnsAnErrorResponseOfTheServerAsItIs() {
+ ApiResponse response = client().actor(ACTOR_ID).server().request("/missing").join();
+
+ assertEquals(404, response.statusCode());
+ assertEquals("Not here", response.bodyAsString());
+ }
+
+ @Test
+ void requestRetriesARateLimitAndServiceUnavailable() {
+ for (int status : List.of(429, 503)) {
+ server.received().clear();
+ flakyStatuses.clear();
+ flakyStatuses.addAll(List.of(status, status));
+
+ ApiResponse response = client(2).actor(ACTOR_ID).server().request("/flaky").join();
+
+ assertEquals(200, response.statusCode(), "status " + status);
+ assertEquals(3, serverRequests().size(), "status " + status);
+ }
+ }
+
+ @Test
+ void requestReturnsTheLastRetriedResponseOnceTheRetriesRunOut() {
+ flakyStatuses.addAll(List.of(503, 503, 503));
+
+ ApiResponse response = client(1).actor(ACTOR_ID).server().request("/flaky").join();
+
+ assertEquals(503, response.statusCode());
+ assertEquals(2, serverRequests().size());
+ }
+
+ @Test
+ void requestReturnsAServerErrorOtherThan503WithoutRetryingIt() {
+ flakyStatuses.add(500);
+
+ ApiResponse response =
+ client(2)
+ .actor(ACTOR_ID)
+ .server()
+ .request("/flaky", new ActorServerRequestOptions().method("POST").body(Map.of("a", 1)))
+ .join();
+
+ assertEquals(500, response.statusCode());
+ assertEquals(1, serverRequests().size());
+ }
+
+ @Test
+ void requestDoesNotRetryATransportErrorSinceTheActorMayHaveAlreadyHandledTheRequest() {
+ CompletableFuture call =
+ client(2)
+ .actor(ACTOR_ID)
+ .server()
+ .request(
+ "/hang-up", new ActorServerRequestOptions().method("POST").body(Map.of("a", 1)));
+
+ failureOf(ApifyTransportException.class, call);
+ assertEquals(1, serverRequests().size());
+ }
+
+ @Test
+ void apiCallsStillThrowOnAnErrorStatusAndRetryIt() {
+ actor = null;
+
+ failureOf(NotFoundError.class, client(2).actor(ACTOR_ID).server().getUrl());
+ assertEquals(1, actorGets(), "a 404 is not retried");
+ }
+
+ // ---- streaming ----------------------------------------------------------------------------
+
+ @Test
+ void requestStreamReturnsTheBodyUnreadOfAnyResponse() throws IOException {
+ ActorServerClient serverClient = client().actor(ACTOR_ID).server();
+
+ try (ActorServerStreamResponse response = serverClient.requestStream("/", null).join()) {
+ assertEquals(200, response.statusCode());
+ assertEquals(
+ Json.toTree(Map.of("ok", true)),
+ Json.parse(response.body().readAllBytes(), JsonNode.class));
+ }
+ try (ActorServerStreamResponse response =
+ serverClient.requestStream("/missing", new ActorServerRequestOptions()).join()) {
+ assertEquals(404, response.statusCode());
+ assertEquals("Not here", new String(response.body().readAllBytes(), StandardCharsets.UTF_8));
+ }
+ TestHttpServer.Received request = serverRequests().get(0);
+ assertEquals("Bearer " + TOKEN, request.header("X-Apify-Authorization"));
+ assertNull(request.header("Authorization"));
+ }
+
+ @Test
+ void requestStreamRetriesARateLimitAndServiceUnavailable() throws IOException {
+ flakyStatuses.addAll(List.of(429, 503));
+
+ try (ActorServerStreamResponse response =
+ client(2).actor(ACTOR_ID).server().requestStream("/flaky", null).join()) {
+ assertEquals(200, response.statusCode());
+ assertEquals(
+ Json.toTree(Map.of("ok", true)),
+ Json.parse(response.body().readAllBytes(), JsonNode.class));
+ }
+ assertEquals(3, serverRequests().size());
+ }
+
+ @Test
+ void requestStreamReturnsTheLastRetriedResponseOnceTheRetriesRunOut() throws IOException {
+ flakyStatuses.addAll(List.of(503, 503, 503));
+
+ try (ActorServerStreamResponse response =
+ client(1).actor(ACTOR_ID).server().requestStream("/flaky", null).join()) {
+ assertEquals(503, response.statusCode());
+ assertEquals(
+ "{\"error\":\"flaky\"}",
+ new String(response.body().readAllBytes(), StandardCharsets.UTF_8));
+ }
+ assertEquals(2, serverRequests().size());
+ }
+
+ @Test
+ void requestStreamDoesNotFollowARedirect() throws IOException {
+ try (TestHttpServer redirecting =
+ new TestHttpServer(request -> TestHttpServer.Reply.redirect("http://127.0.0.1:1/"))) {
+ actor = serverActor(Map.of("standbyUrl", redirecting.url()));
+
+ try (ActorServerStreamResponse response =
+ client().actor(ACTOR_ID).server().requestStream("/", null).join()) {
+ assertEquals(302, response.statusCode());
+ }
+ }
+ }
+
+ // ---- transports that cannot hold back redirects -------------------------------------------
+
+ @Test
+ void aTransportThatCannotDisableRedirectsIsNotTrustedWithARequestToAServer() {
+ MockTransport backend =
+ MockTransport.ofRouter(
+ request ->
+ request.uri().getPath().equals("/v2/actors/a")
+ ? MockTransport.ok(
+ 200,
+ "{\"data\":{\"id\":\"a\",\"standbyUrl\":\"https://a--b.apify.actor\"}}")
+ : MockTransport.ok(200, "{}"));
+ ApifyClient client =
+ ApifyClient.builder().token(TOKEN).httpTransport(backend).maxRetries(0).build();
+
+ CompletableFuture call = client.actor("a").server().request("/x");
+
+ Throwable cause = rootOf(assertThrows(CompletionException.class, call::join));
+ assertInstanceOf(ApifyTransportException.class, cause);
+ assertInstanceOf(UnsupportedOperationException.class, cause.getCause());
+ assertEquals(1, backend.calls, "only the Actor lookup was sent, never the server request");
+ }
+
+ // ---- URL cache and cancellation ---------------------------------------------------------
+
+ /** A transport that delegates to a mock and supports requests that do not follow redirects. */
+ private record RedirectAwareTransport(MockTransport mock) implements HttpTransport {
+ @Override
+ public CompletableFuture> sendAsync(HttpRequest request) {
+ return mock.sendAsync(request);
+ }
+
+ @Override
+ public CompletableFuture> sendStreamingAsync(HttpRequest request) {
+ return mock.sendStreamingAsync(request);
+ }
+
+ @Override
+ public CompletableFuture> sendAsync(
+ HttpRequest request, boolean followRedirects) {
+ return mock.sendAsync(request);
+ }
+
+ @Override
+ public CompletableFuture> sendStreamingAsync(
+ HttpRequest request, boolean followRedirects) {
+ return mock.sendStreamingAsync(request);
+ }
+ }
+
+ /** Cancelling a request after the URL was looked up by {@code warmUp} cancels the HTTP call. */
+ private void assertCancellingARequestReachesTheHttpCall(Consumer warmUp) {
+ MockTransport backend =
+ MockTransport.ofRouter(
+ request -> {
+ String path = request.uri().getPath();
+ if (path.equals("/v2/actors/a")) {
+ return MockTransport.ok(
+ 200,
+ "{\"data\":{\"id\":\"a\",\"standbyUrl\":\"https://a--b.apify.actor\","
+ + "\"taggedBuilds\":{\"latest\":{\"buildId\":\"b1\"}}}}");
+ }
+ return MockTransport.ok(200, "{\"data\":{\"id\":\"b1\"}}");
+ },
+ request -> request.uri().getHost().equals("a--b.apify.actor"));
+ ApifyClient client =
+ ApifyClient.builder()
+ .token(TOKEN)
+ .httpTransport(new RedirectAwareTransport(backend))
+ .maxRetries(0)
+ .build();
+ ActorServerClient serverClient = client.actor("a").server();
+ warmUp.accept(serverClient);
+
+ CompletableFuture call = serverClient.request("/x");
+ assertEquals(1, backend.hung.size(), "the server request is in flight");
+ call.cancel(true);
+
+ assertTrue(backend.allHungCancelled(), "cancelling the request must cancel the HTTP call");
+ }
+
+ @Test
+ void cancellingARequestReachesTheHttpCallAfterGetUrl() {
+ assertCancellingARequestReachesTheHttpCall(serverClient -> serverClient.getUrl().join());
+ }
+
+ @Test
+ void cancellingARequestReachesTheHttpCallAfterGetOpenApiSpec() {
+ assertCancellingARequestReachesTheHttpCall(
+ serverClient -> serverClient.getOpenApiSpec().join());
+ }
+
+ @Test
+ void aNullTimeoutIsRejectedAndLeavesTheClientUsable() {
+ ActorServerClient serverClient = client().actor(ACTOR_ID).server();
+
+ assertThrows(IllegalArgumentException.class, () -> serverClient.getUrl(null));
+ assertThrows(IllegalArgumentException.class, () -> serverClient.getOpenApiSpec(null));
+
+ assertEquals(server.url() + SERVER_PREFIX, serverClient.getUrl().join());
+ assertEquals(200, serverClient.request().join().statusCode());
+ }
+
+ /** A transport whose first call throws instead of returning a future, then serves for real. */
+ private static final class ThrowingOnceTransport implements HttpTransport {
+ private final HttpTransport delegate = new DefaultHttpTransport();
+ private final AtomicBoolean thrown = new AtomicBoolean();
+
+ @Override
+ public CompletableFuture> sendAsync(HttpRequest request) {
+ if (thrown.compareAndSet(false, true)) {
+ throw new IllegalStateException("the transport failed before it returned a future");
+ }
+ return delegate.sendAsync(request);
+ }
+
+ @Override
+ public CompletableFuture> sendStreamingAsync(HttpRequest request) {
+ return delegate.sendStreamingAsync(request);
+ }
+
+ @Override
+ public CompletableFuture> sendAsync(
+ HttpRequest request, boolean followRedirects) {
+ return followRedirects ? sendAsync(request) : delegate.sendAsync(request, false);
+ }
+
+ @Override
+ public CompletableFuture> sendStreamingAsync(
+ HttpRequest request, boolean followRedirects) {
+ return delegate.sendStreamingAsync(request, followRedirects);
+ }
+ }
+
+ @Test
+ void aLookupThatFailsWhileStartingIsNotKept() throws Exception {
+ ApifyClient client =
+ ApifyClient.builder()
+ .token(TOKEN)
+ .baseUrl(server.url())
+ .httpTransport(new ThrowingOnceTransport())
+ .maxRetries(0)
+ .build();
+ ActorServerClient serverClient = client.actor(ACTOR_ID).server();
+
+ failureOf(IllegalStateException.class, serverClient.getUrl());
+
+ assertEquals(server.url() + SERVER_PREFIX, serverClient.getUrl().get(10, TimeUnit.SECONDS));
+ }
+
+ // ---- OpenAPI specification ----------------------------------------------------------------
+
+ @Test
+ void getOpenApiSpecReturnsTheWebServerSchemaOfTheTaggedBuild() {
+ Map> buildTags = new LinkedHashMap<>();
+ buildTags.put("latest", Optional.empty());
+ buildTags.put("beta", Optional.of("beta"));
+ buildTags.put("no build set uses latest", null);
+ for (Map.Entry> tag : buildTags.entrySet()) {
+ Map standby = new LinkedHashMap<>();
+ standby.put("isEnabled", true);
+ if (tag.getValue() != null && tag.getValue().isPresent()) {
+ standby.put("build", tag.getValue().get());
+ }
+ actor = serverActor(Map.of("actorStandby", standby));
+ server.received().clear();
+
+ Optional spec = client().actor(ACTOR_ID).server().getOpenApiSpec().join();
+
+ assertEquals(Optional.of(OPENAPI_SPEC), spec, tag.getKey());
+ String expectedBuild = "beta".equals(tag.getKey()) ? "beta-build-id" : "latest-build-id";
+ assertEquals(
+ "/v2/actor-builds/" + expectedBuild,
+ server.received().get(server.received().size() - 1).path(),
+ tag.getKey());
+ }
+ }
+
+ @Test
+ void getOpenApiSpecLooksABuildNumberUpInTheBuildsOfTheActor() {
+ actor = serverActor(Map.of("actorStandby", Map.of("isEnabled", true, "build", "0.0.1")));
+ builds.put("latest-build-id", "0.0.2");
+ builds.put("old-build-id", "0.0.1");
+
+ Optional spec = client().actor(ACTOR_ID).server().getOpenApiSpec().join();
+
+ assertEquals(Optional.of(OPENAPI_SPEC), spec);
+ assertEquals(
+ "/v2/actor-builds/old-build-id",
+ server.received().get(server.received().size() - 1).path());
+ }
+
+ @Test
+ void getUrlAndGetOpenApiSpecTakeTheTimeoutOfTheCaller() {
+ MockTransport backend =
+ MockTransport.ofRouter(
+ request -> {
+ String path = request.uri().getPath();
+ if (path.equals("/v2/actors/a")) {
+ return MockTransport.ok(
+ 200,
+ "{\"data\":{\"id\":\"a\",\"standbyUrl\":\"https://a--b.apify.actor\","
+ + "\"actorStandby\":{\"build\":\"0.0.1\"}}}");
+ }
+ if (path.equals("/v2/actors/a/builds")) {
+ return MockTransport.ok(
+ 200,
+ "{\"data\":{\"total\":1,\"offset\":0,\"limit\":1,\"count\":1,"
+ + "\"items\":[{\"id\":\"b1\",\"buildNumber\":\"0.0.1\"}]}}");
+ }
+ return MockTransport.ok(200, "{\"data\":{\"id\":\"b1\"}}");
+ });
+ ApifyClient client =
+ ApifyClient.builder().token(TOKEN).httpTransport(backend).maxRetries(0).build();
+ Timeout timeout = Timeout.ofSeconds(7);
+
+ client.actor("a").server().getOpenApiSpec(timeout).join();
+ assertEquals(3, backend.calls, "the Actor, the builds list and the build");
+ assertEquals(
+ List.of(Optional.of(Duration.ofSeconds(7))), backend.timeouts.stream().distinct().toList());
+
+ backend.timeouts.clear();
+ client.actor("a").server().getUrl(timeout).join();
+ assertEquals(List.of(Optional.of(Duration.ofSeconds(7))), backend.timeouts);
+ }
+
+ @Test
+ void getOpenApiSpecReturnsEmptyForABuildWithoutAWebServerSchema() {
+ actorDefinition = Map.of("name", "test-actor");
+ assertEquals(Optional.empty(), client().actor(ACTOR_ID).server().getOpenApiSpec().join());
+
+ actorDefinition = null;
+ assertEquals(Optional.empty(), client().actor(ACTOR_ID).server().getOpenApiSpec().join());
+ }
+
+ @Test
+ void getOpenApiSpecFailsForABuildThatIsNeitherATagNorAListedBuildNumber() {
+ actor = serverActor(Map.of("actorStandby", Map.of("isEnabled", true, "build", "9.9.9")));
+ builds.put("latest-build-id", "0.0.2");
+
+ ServerActorException error =
+ failureOf(ServerActorException.class, client().actor(ACTOR_ID).server().getOpenApiSpec());
+
+ assertEquals("Build 9.9.9 of Actor test-actor-id does not exist.", error.getMessage());
+ }
+
+ @Test
+ void getOpenApiSpecStopsTheBuildNumberLookupAtAnOlderBuildOfTheSameVersion() {
+ actor = serverActor(Map.of("actorStandby", Map.of("isEnabled", true, "build", "0.0.4")));
+ builds.put("build-5", "0.0.5");
+ builds.put("build-3", "0.0.3");
+ builds.put("build-2", "0.0.2");
+ builds.put("build-1", "0.0.1");
+ buildsPerPage = 1;
+
+ failureOf(ServerActorException.class, client().actor(ACTOR_ID).server().getOpenApiSpec());
+
+ assertEquals(2, buildListRequests(), "build-5, then build-3 which is older than 0.0.4");
+ }
+
+ @Test
+ void getOpenApiSpecFailsForAnUnknownTagWithoutListingTheBuilds() {
+ actor = serverActor(Map.of("actorStandby", Map.of("isEnabled", true, "build", "nightly")));
+
+ ServerActorException error =
+ failureOf(ServerActorException.class, client().actor(ACTOR_ID).server().getOpenApiSpec());
+
+ assertEquals("Build nightly of Actor test-actor-id does not exist.", error.getMessage());
+ assertEquals(0, buildListRequests());
+ }
+
+ private long buildListRequests() {
+ return server.received().stream()
+ .filter(request -> request.path().equals(ACTOR_PATH + "/builds"))
+ .count();
+ }
+
+ // ---- ActorClient --------------------------------------------------------------------------
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void standbyReturnsTheSameServerClientAsServer() {
+ ActorClient actorClient = client().actor(ACTOR_ID);
+
+ ActorServerClient standby = actorClient.standby();
+
+ assertNotNull(standby);
+ assertSame(standby, actorClient.server());
+ assertEquals(200, standby.request("/search").join().statusCode());
+ }
+
+ @Test
+ void actorModelExposesTheStandbyUrl() {
+ assertEquals(
+ server.url() + SERVER_PREFIX,
+ client().actor(ACTOR_ID).get().join().orElseThrow().getStandbyUrl());
+ }
+
+ // ---- internal HTTP core -------------------------------------------------------------------
+
+ @Test
+ void theDefaultApiCallKeepsTheTokenInTheAuthorizationHeader() {
+ client().actor(ACTOR_ID).get().join();
+
+ TestHttpServer.Received request = server.received().get(0);
+ assertEquals("Bearer " + TOKEN, request.header("Authorization"));
+ assertNull(request.header("X-Apify-Authorization"));
+ assertEquals(HttpClientCore.MAX_SUCCESS_STATUS, 300);
+ }
+}
diff --git a/src/test/java/com/apify/client/AsyncPaginatedPublisherTest.java b/src/test/java/com/apify/client/AsyncPaginatedPublisherTest.java
index 58a46f8..50057c4 100644
--- a/src/test/java/com/apify/client/AsyncPaginatedPublisherTest.java
+++ b/src/test/java/com/apify/client/AsyncPaginatedPublisherTest.java
@@ -20,6 +20,9 @@
*/
class AsyncPaginatedPublisherTest {
+ /** The fetches recorded by the tests that script their own page fetcher. */
+ private final List fetches = new ArrayList<>();
+
/**
* A fake paginated endpoint over integers {@code [0, available)}. Like the real API, it clamps a
* requested page size (or an unset one) to {@code maxPageSize} ({@code 0} = no clamp).
@@ -120,19 +123,23 @@ void startOffsetIsHonored() {
}
@Test
- void trimsOverReturnedPageToCap() {
- // A misbehaving server that returns more items than requested must not make the publisher
- // overshoot the caller's total cap.
+ void capCountsTheRowsRequestedNotTheItemsAPageReturns() {
+ // A page can return more items than the rows it covers (dataset items with unwind), and all of
+ // its items are yielded. The cap counts rows, capped at the page limit that was requested, so
+ // the next page starts right after the rows asked for and asks only for what is left.
AsyncPaginatedPublisher.PageFetcher overReturning =
(offset, limit) -> {
+ fetches.add(new long[] {offset, limit});
PaginationList page = new PaginationList<>();
page.setItems(List.of(1, 2, 3, 4, 5)); // ignores the requested limit
- page.setTotal(5);
page.setCount(5);
return CompletableFuture.completedFuture(page);
};
List got = drain(new AsyncPaginatedPublisher<>(3L, 2L, null, overReturning));
- assertEquals(List.of(1, 2, 3), got, "the last page is trimmed to the cap");
+ assertEquals(10, got.size(), "every item of the pages read is yielded");
+ assertEquals(2, fetches.size());
+ assertEquals(List.of(0L, 2L), List.of(fetches.get(0)[0], fetches.get(1)[0]));
+ assertEquals(List.of(2L, 1L), List.of(fetches.get(0)[1], fetches.get(1)[1]));
}
@Test
diff --git a/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java b/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java
index a87dfa3..983e5dc 100644
--- a/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java
+++ b/src/test/java/com/apify/client/ClientBehaviourRegressionTest.java
@@ -33,14 +33,17 @@
import com.apify.client.store.StoreListOptions;
import com.apify.client.webhook.NestedWebhookCollectionClient;
import com.apify.client.webhook.Webhook;
+import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Base64;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Flow;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.Test;
@@ -285,6 +288,28 @@ void createKeysPublicUrlForwardsFilterOptions() {
assertFalse(url.contains("signature="), "a public store should get no signature: " + url);
}
+ @Test
+ void createKeysPublicUrlRejectsAnInvalidExpiryBeforeAnyRequest() {
+ MockTransport backend = MockTransport.ofConstant(200, "{\"data\":{\"id\":\"store1\"}}");
+ var store = client(backend).keyValueStore("store1");
+ for (long expiresInSecs : new long[] {0L, -1L, Long.MAX_VALUE}) {
+ assertThrows(IllegalArgumentException.class, () -> store.createKeysPublicUrl(expiresInSecs));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> store.createKeysPublicUrl(new ListKeysOptions(), expiresInSecs));
+ }
+ assertEquals(0, backend.calls, "an invalid expiresInSecs must fail before any request");
+ }
+
+ @Test
+ void createKeysPublicUrlSignsAPrivateStoreWithAPositiveExpiry() {
+ MockTransport backend =
+ MockTransport.ofConstant(
+ 200, "{\"data\":{\"id\":\"store1\",\"urlSigningSecretKey\":\"secret\"}}");
+ String url = client(backend).keyValueStore("store1").createKeysPublicUrl(3600L).join();
+ assertTrue(url.contains("signature="), url);
+ }
+
@Test
void downloadItemsForwardsItemSelectionParams() {
MockTransport backend = MockTransport.ofConstant(200, "col1,col2\n");
@@ -362,7 +387,7 @@ void batchAddRequestsThrowsWhenSingleRequestExceedsPayloadLimit() {
void paginateRequestsTrimsPageToTotalLimitWhenServerOvershoots() {
// If the server ignores (or overshoots) the requested per-page `limit` and returns more items
// than the caller's totalLimit cap allows, the publisher must still yield exactly totalLimit
- // items, mirroring AsyncPaginatedPublisher's own defensive trim.
+ // items.
MockTransport backend =
MockTransport.ofConstant(
200,
@@ -476,6 +501,34 @@ void storeIterateWalksMultiplePages() {
assertEquals(3, backend.calls, "two data pages plus the terminating empty page");
}
+ @Test
+ void storeIterateSkipsEntriesTheApiDroppedAfterCountingThem() {
+ // The Store list can drop entries after counting them: a page that counted 2 rows returns 1
+ // item. The next page must start after both rows, or it would read the dropped one again, and
+ // a page that counted rows but returned none must not end the iteration.
+ MockTransport backend =
+ new MockTransport(
+ List.of(
+ MockTransport.ok(
+ 200,
+ "{\"data\":{\"items\":[{}],\"total\":5,\"offset\":0,\"limit\":2,\"count\":2}}"),
+ MockTransport.ok(
+ 200,
+ "{\"data\":{\"items\":[],\"total\":5,\"offset\":2,\"limit\":2,\"count\":2}}"),
+ MockTransport.ok(
+ 200,
+ "{\"data\":{\"items\":[{}],\"total\":5,\"offset\":4,\"limit\":2,\"count\":1}}"),
+ MockTransport.ok(
+ 200,
+ "{\"data\":{\"items\":[],\"total\":5,\"offset\":5,\"limit\":2,\"count\":0}}")));
+ List items =
+ Publishers.collect(client(backend).store().iterate(new StoreListOptions(), 2L)).join();
+ assertEquals(2, items.size());
+ assertEquals(4, backend.calls, "a page that counted rows continues even without items");
+ assertTrue(backend.urls.get(1).contains("offset=2"), backend.urls.get(1));
+ assertTrue(backend.urls.get(2).contains("offset=4"), backend.urls.get(2));
+ }
+
@Test
void collectionIterateSingleArgDelegatesToServerDefaultPageSize() {
// The arg-less-chunkSize convenience overload must page correctly (delegates with null chunk).
@@ -917,6 +970,62 @@ void createItemsPublicUrlWithFormatSetsQueryParam() {
assertTrue(url.contains("format=csv"), url);
}
+ @Test
+ void createItemsPublicUrlRejectsAnInvalidExpiryBeforeAnyRequest() {
+ MockTransport backend = MockTransport.ofConstant(200, "{\"data\":{\"id\":\"d1\"}}");
+ var dataset = client(backend).dataset("d1");
+ for (long expiresInSecs : new long[] {0L, -1L, Long.MAX_VALUE}) {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> dataset.createItemsPublicUrl(new DatasetListItemsOptions(), expiresInSecs));
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ dataset.createItemsPublicUrl(
+ new DatasetListItemsOptions(), expiresInSecs, DownloadItemsFormat.CSV));
+ }
+ assertEquals(0, backend.calls, "an invalid expiresInSecs must fail before any request");
+ }
+
+ @Test
+ void createItemsPublicUrlSignsAPrivateDatasetWithAPositiveExpiry() {
+ MockTransport backend =
+ MockTransport.ofConstant(
+ 200, "{\"data\":{\"id\":\"d1\",\"urlSigningSecretKey\":\"secret\"}}");
+ String url =
+ client(backend)
+ .dataset("d1")
+ .createItemsPublicUrl(new DatasetListItemsOptions(), 3600L)
+ .join();
+ assertTrue(url.contains("signature="), url);
+ }
+
+ @Test
+ void createItemsPublicUrlCountsTheValidityFromAfterTheLookup() {
+ AtomicLong lookupDoneAt = new AtomicLong();
+ MockTransport backend =
+ MockTransport.ofRouter(
+ request -> {
+ try {
+ Thread.sleep(300);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ lookupDoneAt.set(System.currentTimeMillis());
+ return new MockTransport.Scripted(
+ 200, "{\"data\":{\"id\":\"d1\",\"urlSigningSecretKey\":\"secret\"}}", null);
+ });
+ String url =
+ client(backend)
+ .dataset("d1")
+ .createItemsPublicUrl(new DatasetListItemsOptions(), 60L)
+ .join();
+ String signature = url.substring(url.indexOf("signature=") + "signature=".length());
+ String decoded = new String(Base64.getUrlDecoder().decode(signature), StandardCharsets.UTF_8);
+ long expiresAt = Long.parseLong(decoded.split("\\.")[1]);
+ assertTrue(expiresAt >= lookupDoneAt.get() + 60_000L, decoded + " signed before the lookup");
+ }
+
@Test
void createItemsPublicUrlWithoutFormatOmitsQueryParam() {
MockTransport backend = MockTransport.ofConstant(200, "{\"data\":{\"id\":\"d1\"}}");
diff --git a/src/test/java/com/apify/client/DatasetItemsIteratorTest.java b/src/test/java/com/apify/client/DatasetItemsIteratorTest.java
index 393f10a..2813fce 100644
--- a/src/test/java/com/apify/client/DatasetItemsIteratorTest.java
+++ b/src/test/java/com/apify/client/DatasetItemsIteratorTest.java
@@ -76,31 +76,71 @@ static final class Row {
}
@Test
- void totalCapTrimsDatasetItems() {
- // The cap wins even though the server would return more; only the first page is requested.
+ void limitCapsTheRowsReadNotTheItemsReturned() {
+ // The page returns more items than the 2 rows asked for, as an unwound page does: all of them
+ // are yielded, and no further page is requested since the 2 rows were read.
MockTransport backend = MockTransport.ofConstant(200, "[{\"n\":1},{\"n\":2},{\"n\":3}]");
List seen =
collect(
client(backend)
.dataset("d1")
.iterateItems(new DatasetListItemsOptions().limit(2L), 5L));
- List ns = seen.stream().map(n -> n.get("n").asInt()).toList();
- assertEquals(List.of(1, 2), ns, "limit caps the total items yielded");
+ assertEquals(3, seen.size());
assertEquals(1, backend.calls);
}
+ @Test
+ void limitCountsTheRowsAFilterDropped() {
+ // With limit 4 and chunk 2, the two pages cover 4 rows although a filter left one item: the
+ // iteration must stop there instead of reading on until it has yielded 4 items.
+ MockTransport backend =
+ new MockTransport(
+ List.of(
+ MockTransport.ok(200, "[{\"n\":1}]", Map.of("X-Apify-Pagination-Count", "2")),
+ MockTransport.ok(200, "[]", Map.of("X-Apify-Pagination-Count", "2")),
+ MockTransport.ok(200, "[{\"n\":9}]", Map.of("X-Apify-Pagination-Count", "2"))));
+ List seen =
+ collect(
+ client(backend)
+ .dataset("d1")
+ .iterateItems(new DatasetListItemsOptions().clean(true).limit(4L), 2L));
+ assertEquals(List.of(1), seen.stream().map(n -> n.get("n").asInt()).toList());
+ assertEquals(2, backend.calls, "rows 0-3 were read, so there is no third page");
+ assertTrue(backend.urls.get(1).contains("offset=2"), backend.urls.get(1));
+ assertTrue(backend.urls.get(1).contains("limit=2"), backend.urls.get(1));
+ }
+
/** A typed item for {@link #decodesIntoRequestedType()}. */
public record Item(int n) {}
@Test
- void listItemsExposesScannedCountFromHeader() {
+ void listItemsCountIsTheScannedRowsWhenFiltersDroppedItems() {
MockTransport backend =
new MockTransport(
List.of(MockTransport.ok(200, "[{\"n\":1}]", Map.of("X-Apify-Pagination-Count", "5"))));
PaginationList page =
client(backend).dataset("d1").listItems(new DatasetListItemsOptions()).join();
- assertEquals(1, page.getCount(), "count stays the number of items actually returned");
- assertEquals(5L, page.getScannedCount(), "scannedCount comes from X-Apify-Pagination-Count");
+ assertEquals(5, page.getCount(), "count is the rows scanned, which filters leave above items");
+ assertEquals(1, page.getItems().size());
+ assertEquals(5L, page.getScannedCount(), "the raw header value stays available");
+ }
+
+ @Test
+ void listItemsCountIsTheLargerOfTheScannedRowsAndTheItemsReturned() {
+ // The header is derived from the dataset's item count, which lags a fresh push, and an unwound
+ // page returns more items than the rows it scanned.
+ assertEquals(3L, listedCount("[{\"n\":1}]", "3"), "filtered page");
+ assertEquals(2L, listedCount("[{\"n\":1},{\"n\":2}]", "0"), "lagging header");
+ assertEquals(2L, listedCount("[{\"n\":1},{\"n\":2}]", "1"), "unwound page");
+ assertEquals(1L, listedCount("[{\"n\":1}]", null), "missing header");
+ assertEquals(1L, listedCount("[]", "1"), "an emptied page still covered a row");
+ }
+
+ private static long listedCount(String body, String scannedHeader) {
+ Map headers =
+ scannedHeader == null ? Map.of() : Map.of("X-Apify-Pagination-Count", scannedHeader);
+ MockTransport backend = new MockTransport(List.of(MockTransport.ok(200, body, headers)));
+ return client(backend).dataset("d1").listItems(new DatasetListItemsOptions()).join().getCount();
}
@Test
@@ -179,6 +219,26 @@ void advancesByScannedCountEvenWhenUnwindReturnsMoreThanScanned() {
assertFalse(backend.urls.get(1).contains("offset=4"), backend.urls.get(1));
}
+ @Test
+ void capsTheStepAtTheRequestedLimitWhenTheHeaderLags() {
+ // An unwound page returns more items than the rows it scanned and the lagging header reads 0,
+ // so the count is the items returned. The next page must still start right after the 2 rows
+ // the request asked for, not after the 4 items returned, or the rows between are never read.
+ MockTransport backend =
+ new MockTransport(
+ List.of(
+ MockTransport.ok(
+ 200,
+ "[{\"n\":1},{\"n\":2},{\"n\":3},{\"n\":4}]",
+ Map.of("X-Apify-Pagination-Count", "0")),
+ MockTransport.ok(200, "[]", Map.of("X-Apify-Pagination-Count", "0"))));
+ collect(
+ client(backend)
+ .dataset("d1")
+ .iterateItems(new DatasetListItemsOptions().unwind(List.of("arr")), 2L));
+ assertTrue(backend.urls.get(1).contains("offset=2"), backend.urls.get(1));
+ }
+
@Test
void fallsBackToReturnedCountWhenHeaderMissing() {
// Endpoints/mocks that do not send X-Apify-Pagination-Count (every collection but dataset
diff --git a/src/test/java/com/apify/client/MockTransport.java b/src/test/java/com/apify/client/MockTransport.java
index c9a9cec..12e9be0 100644
--- a/src/test/java/com/apify/client/MockTransport.java
+++ b/src/test/java/com/apify/client/MockTransport.java
@@ -120,6 +120,17 @@ static MockTransport ofRouter(Function router) {
return transport;
}
+ /**
+ * A transport that answers every request with {@code router}'s response for it, except the
+ * requests {@code hangWhen} accepts, which it never answers.
+ */
+ static MockTransport ofRouter(
+ Function router, Predicate hangWhen) {
+ MockTransport transport = ofRouter(router);
+ transport.hangWhen = hangWhen;
+ return transport;
+ }
+
/** A transport that never answers the requests {@code hangWhen} accepts. */
static MockTransport ofHanging(Predicate hangWhen) {
MockTransport transport = ofConstant(200, "{\"data\":{\"id\":\"r1\",\"status\":\"RUNNING\"}}");
diff --git a/src/test/java/com/apify/client/TestHttpServer.java b/src/test/java/com/apify/client/TestHttpServer.java
new file mode 100644
index 0000000..e89ca24
--- /dev/null
+++ b/src/test/java/com/apify/client/TestHttpServer.java
@@ -0,0 +1,169 @@
+package com.apify.client;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Function;
+
+/**
+ * A minimal HTTP/1.1 server on a local port, for tests that need a real network exchange (header
+ * names, redirects, dropped connections) that {@link MockTransport} cannot show. It answers every
+ * request through a handler and closes the connection after each response. It is built on {@link
+ * ServerSocket} because the {@code com.sun.net.httpserver} module is not readable from the tests.
+ */
+final class TestHttpServer implements AutoCloseable {
+
+ /** A request the server received. */
+ record Received(
+ String method, String path, String query, Map headers, byte[] body) {
+
+ /** The value of a header, whatever the case of its name, or null. */
+ String header(String name) {
+ return headers.get(name.toLowerCase(Locale.ROOT));
+ }
+
+ String bodyText() {
+ return new String(body, StandardCharsets.UTF_8);
+ }
+ }
+
+ /** A response to send; {@code dropConnection} closes the connection without answering. */
+ record Reply(int status, Map headers, byte[] body, boolean dropConnection) {
+
+ static Reply of(int status, String contentType, String body) {
+ return new Reply(
+ status,
+ Map.of("Content-Type", contentType),
+ body.getBytes(StandardCharsets.UTF_8),
+ false);
+ }
+
+ static Reply json(int status, String body) {
+ return of(status, "application/json", body);
+ }
+
+ static Reply redirect(String location) {
+ return new Reply(302, Map.of("Location", location), new byte[0], false);
+ }
+
+ static Reply drop() {
+ return new Reply(0, Map.of(), new byte[0], true);
+ }
+ }
+
+ private final ServerSocket socket;
+ private final Function handler;
+ private final List received = new CopyOnWriteArrayList<>();
+ private final Thread acceptor;
+
+ TestHttpServer(Function handler) throws IOException {
+ this.socket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
+ this.handler = handler;
+ this.acceptor = new Thread(this::acceptLoop, "test-http-server");
+ this.acceptor.setDaemon(true);
+ this.acceptor.start();
+ }
+
+ /** The origin of the server, for example {@code http://127.0.0.1:41234}. */
+ String url() {
+ return "http://127.0.0.1:" + socket.getLocalPort();
+ }
+
+ /** The requests received so far, oldest first. */
+ List received() {
+ return received;
+ }
+
+ private void acceptLoop() {
+ while (!socket.isClosed()) {
+ try {
+ Socket connection = socket.accept();
+ Thread worker = new Thread(() -> serve(connection), "test-http-server-connection");
+ worker.setDaemon(true);
+ worker.start();
+ } catch (IOException e) {
+ return; // the socket was closed
+ }
+ }
+ }
+
+ private void serve(Socket connection) {
+ try (connection) {
+ InputStream in = connection.getInputStream();
+ Received request = read(in);
+ if (request == null) {
+ return;
+ }
+ received.add(request);
+ Reply reply = handler.apply(request);
+ if (reply.dropConnection()) {
+ return;
+ }
+ write(connection.getOutputStream(), reply);
+ } catch (IOException e) {
+ // The client went away; there is nobody to answer.
+ }
+ }
+
+ private static Received read(InputStream in) throws IOException {
+ String requestLine = readLine(in);
+ if (requestLine == null || requestLine.isEmpty()) {
+ return null;
+ }
+ String[] parts = requestLine.split(" ");
+ Map headers = new TreeMap<>();
+ for (String line = readLine(in); line != null && !line.isEmpty(); line = readLine(in)) {
+ int colon = line.indexOf(':');
+ headers.put(
+ line.substring(0, colon).toLowerCase(Locale.ROOT), line.substring(colon + 1).trim());
+ }
+ int length = Integer.parseInt(headers.getOrDefault("content-length", "0"));
+ byte[] body = in.readNBytes(length);
+ String target = parts[1];
+ int question = target.indexOf('?');
+ String path = question < 0 ? target : target.substring(0, question);
+ String query = question < 0 ? "" : target.substring(question + 1);
+ return new Received(parts[0], path, query, headers, body);
+ }
+
+ private static String readLine(InputStream in) throws IOException {
+ ByteArrayOutputStream line = new ByteArrayOutputStream();
+ int b;
+ while ((b = in.read()) != -1) {
+ if (b == '\n') {
+ break;
+ }
+ if (b != '\r') {
+ line.write(b);
+ }
+ }
+ return b == -1 && line.size() == 0 ? null : line.toString(StandardCharsets.UTF_8);
+ }
+
+ private static void write(OutputStream out, Reply reply) throws IOException {
+ StringBuilder head = new StringBuilder("HTTP/1.1 " + reply.status() + " Reply\r\n");
+ reply
+ .headers()
+ .forEach((name, value) -> head.append(name).append(": ").append(value).append("\r\n"));
+ head.append("Content-Length: ").append(reply.body().length).append("\r\n");
+ head.append("Connection: close\r\n\r\n");
+ out.write(head.toString().getBytes(StandardCharsets.UTF_8));
+ out.write(reply.body());
+ out.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ socket.close();
+ }
+}
diff --git a/src/test/java/com/apify/client/integration/ActorServerIntegrationTest.java b/src/test/java/com/apify/client/integration/ActorServerIntegrationTest.java
new file mode 100644
index 0000000..21b7735
--- /dev/null
+++ b/src/test/java/com/apify/client/integration/ActorServerIntegrationTest.java
@@ -0,0 +1,184 @@
+package com.apify.client.integration;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import com.apify.client.ApifyClient;
+import com.apify.client.actor.Actor;
+import com.apify.client.actor.ActorBuildOptions;
+import com.apify.client.actor.ActorServerClient;
+import com.apify.client.actor.ActorServerRequestOptions;
+import com.apify.client.build.Build;
+import com.apify.client.http.ApiResponse;
+import com.apify.client.http.ServerActorException;
+import com.apify.client.internal.Json;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletionException;
+import org.junit.jupiter.api.Test;
+import tools.jackson.databind.JsonNode;
+
+/**
+ * Creates a Server Actor that runs an echo web server, builds it, and sends requests to it through
+ * {@link ActorServerClient}.
+ */
+class ActorServerIntegrationTest extends IntegrationBase {
+
+ private static final Map OPENAPI_SPEC =
+ Map.of(
+ "openapi",
+ "3.0.0",
+ "info",
+ Map.of("title", "Echo server", "version", "1.0.0"),
+ "paths",
+ Map.of(
+ "/echo",
+ Map.of(
+ "post",
+ Map.of(
+ "responses", Map.of("200", Map.of("description", "The echoed request."))))));
+
+ /**
+ * A web server that answers every request, including the platform's readiness probe, with the
+ * request it received, under the status in the {@code status} query parameter, or 200.
+ */
+ private static final String ECHO_SERVER_SOURCE =
+ """
+ const http = require('http');
+
+ http.createServer((req, res) => {
+ let body = '';
+ req.on('data', (chunk) => { body += chunk; });
+ req.on('end', () => {
+ const status = Number(new URL(req.url, 'http://localhost').searchParams.get('status')) || 200;
+ res.writeHead(status, { 'Content-Type': 'application/json' });
+ res.end(JSON.stringify({ method: req.method, url: req.url, headers: req.headers, body }));
+ });
+ }).listen(process.env.ACTOR_WEB_SERVER_PORT);
+ """;
+
+ private static final String DOCKERFILE =
+ "FROM node:22-alpine\nCOPY main.js ./\nCMD [\"node\", \"main.js\"]\n";
+
+ /** How long the build of the echo Actor may take, in seconds. */
+ private static final long BUILD_WAIT_SECS = 300L;
+
+ private static Map sourceFile(String name, String content) {
+ return Map.of("name", name, "format", "TEXT", "content", content);
+ }
+
+ private static Actor createEchoServerActor(ApifyClient client, String name) {
+ Map actorJson =
+ Map.of(
+ "actorSpecification",
+ 1,
+ "name",
+ name,
+ "version",
+ "0.0",
+ "usesStandbyMode",
+ true,
+ "webServerSchema",
+ OPENAPI_SPEC);
+ return client
+ .actors()
+ .create(
+ Map.of(
+ "name",
+ name,
+ "versions",
+ List.of(
+ Map.of(
+ "versionNumber",
+ "0.0",
+ "sourceType",
+ "SOURCE_FILES",
+ "buildTag",
+ "latest",
+ "sourceFiles",
+ List.of(
+ sourceFile("Dockerfile", DOCKERFILE),
+ sourceFile("main.js", ECHO_SERVER_SOURCE),
+ sourceFile(
+ ".actor/actor.json",
+ new String(Json.toBytes(actorJson), StandardCharsets.UTF_8))))),
+ "actorStandby",
+ Map.of("isEnabled", true, "memoryMbytes", 256, "idleTimeoutSecs", 60)))
+ .join();
+ }
+
+ @Test
+ void theServerClientReturnsTheUrlAndSpecificationAndSendsRequestsToTheServer() {
+ ApifyClient client = requireClient();
+ Actor created = createEchoServerActor(client, uniqueName("server"));
+ try {
+ Build build = client.actor(created.getId()).build("0.0", new ActorBuildOptions()).join();
+ Build finished = client.build(build.getId()).waitForFinish(BUILD_WAIT_SECS).join();
+ assertEquals("SUCCEEDED", finished.getStatus());
+
+ Actor actor = client.actor(created.getId()).get().join().orElseThrow();
+ ActorServerClient server = client.actor(created.getId()).server();
+
+ assertEquals(actor.getStandbyUrl(), server.getUrl().join());
+ Optional spec = server.getOpenApiSpec().join();
+ assertEquals(Optional.of(Json.toTree(OPENAPI_SPEC)), spec);
+
+ ApiResponse response =
+ server
+ .request(
+ "/echo",
+ new ActorServerRequestOptions()
+ .method("POST")
+ .param("q", "apify")
+ .body(Map.of("limit", 5)))
+ .join();
+
+ assertEquals(200, response.statusCode());
+ JsonNode echoed = response.bodyAs(JsonNode.class);
+ assertEquals("POST", echoed.get("method").asText());
+ assertEquals("/echo?q=apify", echoed.get("url").asText());
+ assertEquals(
+ Json.toTree(Map.of("limit", 5)),
+ Json.parse(echoed.get("body").asText().getBytes(StandardCharsets.UTF_8), JsonNode.class));
+ // The platform consumes the client's token, so it never reaches the Actor's code.
+ assertFalse(echoed.get("headers").has("x-apify-authorization"));
+ assertFalse(echoed.get("headers").has("authorization"));
+
+ ApiResponse withOwnAuthorization =
+ server
+ .request(
+ "/echo",
+ new ActorServerRequestOptions().header("Authorization", "Bearer actor-token"))
+ .join();
+ assertEquals(
+ "Bearer actor-token",
+ withOwnAuthorization.bodyAs(JsonNode.class).get("headers").get("authorization").asText());
+
+ ApiResponse errorResponse =
+ server.request("/echo", new ActorServerRequestOptions().param("status", 404)).join();
+ assertEquals(404, errorResponse.statusCode());
+ assertEquals("/echo?status=404", errorResponse.bodyAs(JsonNode.class).get("url").asText());
+ } finally {
+ client.actor(created.getId()).delete().join();
+ }
+ }
+
+ @Test
+ void anActorThatIsNotAServerActorIsRefused() {
+ ApifyClient client = requireClient();
+ Actor actor = client.actor("apify/hello-world").get().join().orElseThrow();
+ assertNull(actor.getStandbyUrl(), "hello-world does not run in Standby mode");
+
+ CompletionException error =
+ assertThrows(
+ CompletionException.class,
+ () -> client.actor("apify/hello-world").server().getUrl().join());
+
+ assertInstanceOf(ServerActorException.class, error.getCause());
+ }
+}
diff --git a/src/test/java/com/apify/client/integration/IterationIntegrationTest.java b/src/test/java/com/apify/client/integration/IterationIntegrationTest.java
index 2270130..ecf8ad9 100644
--- a/src/test/java/com/apify/client/integration/IterationIntegrationTest.java
+++ b/src/test/java/com/apify/client/integration/IterationIntegrationTest.java
@@ -144,7 +144,8 @@ void iterateDatasetItems() {
List seen = all.stream().map(n -> n.get("n").asInt()).toList();
assertEquals(List.of(1, 2, 3, 4, 5), seen, "iterateItems should yield all items in order");
- // The total cap trims the tail: limit=3 over 5 items with page size 2 yields exactly 3.
+ // limit=3 over 5 items with page size 2 reads 2 rows, then the 1 row that is left of the
+ // limit.
List cappedAll =
Publishers.collect(dataset.iterateItems(new DatasetListItemsOptions().limit(3L), 2L))
.join();
diff --git a/src/test/java/com/apify/client/internal/SignatureTest.java b/src/test/java/com/apify/client/internal/SignatureTest.java
index e351441..ace9d4a 100644
--- a/src/test/java/com/apify/client/internal/SignatureTest.java
+++ b/src/test/java/com/apify/client/internal/SignatureTest.java
@@ -1,11 +1,16 @@
package com.apify.client.internal;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
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.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
/** Known-answer tests pinning the URL-signing scheme against upstream {@code @apify/utilities}. */
class SignatureTest {
@@ -32,4 +37,40 @@ void storageContentSignatureNonExpiring() {
String expected = "0.0." + Signatures.createHmacSignature("secret", "0.0.RESID");
assertEquals(expected, decoded);
}
+
+ @Test
+ void storageContentSignatureExpiresInTheValidityAfterSigning() {
+ long before = System.currentTimeMillis();
+ String sig = Signatures.signStorageContent("secret", "RESID", 60L);
+ long after = System.currentTimeMillis();
+ String decoded = new String(Base64.getUrlDecoder().decode(sig), StandardCharsets.UTF_8);
+ long expiresAt = Long.parseLong(decoded.split("\\.")[1]);
+ assertTrue(expiresAt >= before + 60_000L && expiresAt <= after + 60_000L, decoded);
+ }
+
+ @ParameterizedTest
+ @ValueSource(
+ longs = {0L, -1L, Long.MIN_VALUE, Signatures.MAX_EXPIRES_IN_SECS + 1, Long.MAX_VALUE})
+ void signingRejectsAnExpiryThatIsNotPositiveOrTooLarge(long expiresInSecs) {
+ IllegalArgumentException e =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> Signatures.signStorageContent("secret", "RESID", expiresInSecs));
+ assertTrue(e.getMessage().contains("expiresInSecs"), e.getMessage());
+ }
+
+ @ParameterizedTest
+ @ValueSource(longs = {1L, 3600L, Signatures.MAX_EXPIRES_IN_SECS})
+ void signingAcceptsAPositiveExpiryUpToTheMaximum(long expiresInSecs) {
+ assertDoesNotThrow(() -> Signatures.signStorageContent("secret", "RESID", expiresInSecs));
+ }
+
+ @Test
+ void theMaximumExpiryDoesNotOverflow() {
+ assertEquals(4_611_686_018_427_387L, Signatures.MAX_EXPIRES_IN_SECS);
+ String sig = Signatures.signStorageContent("secret", "RESID", Signatures.MAX_EXPIRES_IN_SECS);
+ String decoded = new String(Base64.getUrlDecoder().decode(sig), StandardCharsets.UTF_8);
+ long expiresAt = Long.parseLong(decoded.split("\\.")[1]);
+ assertTrue(expiresAt > System.currentTimeMillis(), decoded);
+ }
}