getStatistics() {
- return ctx.getResourceRequired("statistics", new QueryParams(), JsonNode.class);
+ return ctx.getResourceRequired("statistics", new QueryParams(), JsonNode.class, Timeout.SHORT);
}
/**
diff --git a/src/main/java/com/apify/client/dataset/DatasetListItemsOptions.java b/src/main/java/com/apify/client/dataset/DatasetListItemsOptions.java
index 6960086..0e6399a 100644
--- a/src/main/java/com/apify/client/dataset/DatasetListItemsOptions.java
+++ b/src/main/java/com/apify/client/dataset/DatasetListItemsOptions.java
@@ -1,5 +1,6 @@
package com.apify.client.dataset;
+import com.apify.client.internal.DatasetItemsOptionsAccess;
import com.apify.client.internal.QueryParams;
import java.util.List;
@@ -121,6 +122,72 @@ public DatasetListItemsOptions signature(String signature) {
return this;
}
+ static {
+ DatasetItemsOptionsAccess.register(
+ new DatasetItemsOptionsAccess() {
+ @Override
+ public long offset(DatasetListItemsOptions options) {
+ return options.offset != null && options.offset > 0 ? options.offset : 0;
+ }
+
+ @Override
+ public Long limit(DatasetListItemsOptions options) {
+ return options.limit;
+ }
+
+ @Override
+ public boolean dropsRows(DatasetListItemsOptions options) {
+ return Boolean.TRUE.equals(options.clean)
+ || Boolean.TRUE.equals(options.skipEmpty)
+ || (options.unwind != null && !options.unwind.isEmpty());
+ }
+
+ @Override
+ public DatasetListItemsOptions snapshot(DatasetListItemsOptions options) {
+ DatasetListItemsOptions c =
+ options != null ? options.copy() : new DatasetListItemsOptions();
+ c.desc = null;
+ return c;
+ }
+
+ @Override
+ public DatasetListItemsOptions page(
+ DatasetListItemsOptions options, long offset, long limit) {
+ DatasetListItemsOptions c = options.copy();
+ c.desc = null;
+ c.offset = offset;
+ c.limit = limit;
+ return c;
+ }
+
+ @Override
+ public DatasetListItemsOptions plainPage(long offset, long limit) {
+ return new DatasetListItemsOptions().offset(offset).limit(limit);
+ }
+ });
+ }
+
+ /** Returns an independent copy of these options. */
+ private DatasetListItemsOptions copy() {
+ DatasetListItemsOptions c = new DatasetListItemsOptions();
+ c.offset = offset;
+ c.limit = limit;
+ c.desc = desc;
+ c.fields = fields;
+ c.outputFields = outputFields;
+ c.omit = omit;
+ c.skipEmpty = skipEmpty;
+ c.skipHidden = skipHidden;
+ c.clean = clean;
+ c.unwind = unwind;
+ c.flatten = flatten;
+ c.view = view;
+ c.simplified = simplified;
+ c.skipFailedPages = skipFailedPages;
+ c.signature = signature;
+ return c;
+ }
+
Boolean descValue() {
return desc;
}
diff --git a/src/main/java/com/apify/client/http/BrotliHttpCompressor.java b/src/main/java/com/apify/client/http/BrotliHttpCompressor.java
new file mode 100644
index 0000000..95942f5
--- /dev/null
+++ b/src/main/java/com/apify/client/http/BrotliHttpCompressor.java
@@ -0,0 +1,73 @@
+package com.apify.client.http;
+
+import com.aayushatharva.brotli4j.Brotli4jLoader;
+import com.aayushatharva.brotli4j.encoder.Encoder;
+import java.io.IOException;
+
+/**
+ * An {@link HttpCompressor} that brotli-encodes request bodies ({@code Content-Encoding: br}).
+ *
+ * It needs a platform-specific native library bundled with {@code brotli4j}; {@link
+ * #isAvailable()} tells whether one loaded. The client uses this compressor by default where it is
+ * available and gzip elsewhere.
+ */
+public final class BrotliHttpCompressor implements HttpCompressor {
+
+ private static final int MAX_QUALITY = 11;
+
+ private static final int DEFAULT_QUALITY = 6;
+
+ private final int quality;
+
+ /**
+ * Whether the brotli native library loaded for the running platform. Brotli needs a native codec
+ * that is not bundled for every OS and CPU; the check never throws.
+ */
+ public static boolean isAvailable() {
+ try {
+ Brotli4jLoader.ensureAvailability();
+ return true;
+ } catch (Throwable t) {
+ // Catch Throwable: native loading raises UnsatisfiedLinkError/NoClassDefFoundError, not only
+ // Exception, and "unavailable" is the answer in every such case.
+ return false;
+ }
+ }
+
+ /**
+ * A compressor with the default quality, 6.
+ *
+ * @throws IllegalStateException if the brotli native library is not available on this platform
+ */
+ public BrotliHttpCompressor() {
+ this(DEFAULT_QUALITY);
+ }
+
+ /**
+ * A compressor with the given quality.
+ *
+ * @param quality the compression quality from 0 (fastest) to 11 (smallest output)
+ * @throws IllegalArgumentException if {@code quality} is outside 0-11
+ * @throws IllegalStateException if the brotli native library is not available on this platform
+ */
+ public BrotliHttpCompressor(int quality) {
+ if (quality < 0 || quality > MAX_QUALITY) {
+ throw new IllegalArgumentException("brotli quality must be between 0 and 11, was " + quality);
+ }
+ if (!isAvailable()) {
+ throw new IllegalStateException(
+ "the brotli native library is not available on this platform; use GzipHttpCompressor");
+ }
+ this.quality = quality;
+ }
+
+ @Override
+ public String contentEncoding() {
+ return "br";
+ }
+
+ @Override
+ public byte[] compress(byte[] data) throws IOException {
+ return Encoder.compress(data, new Encoder.Parameters().setQuality(quality));
+ }
+}
diff --git a/src/main/java/com/apify/client/http/DefaultHttpTransport.java b/src/main/java/com/apify/client/http/DefaultHttpTransport.java
index 325da87..027ef37 100644
--- a/src/main/java/com/apify/client/http/DefaultHttpTransport.java
+++ b/src/main/java/com/apify/client/http/DefaultHttpTransport.java
@@ -69,6 +69,13 @@ public CompletableFuture> sendStreamingAsync(HttpReque
*/
private static CompletableFuture unwrapTimeout(CompletableFuture future) {
CompletableFuture result = new CompletableFuture<>();
+ // Cancelling the returned future aborts the exchange the JDK client has in flight.
+ result.whenComplete(
+ (value, error) -> {
+ if (result.isCancelled()) {
+ future.cancel(true);
+ }
+ });
future.whenComplete(
(value, error) -> {
if (error == null) {
diff --git a/src/main/java/com/apify/client/http/GzipHttpCompressor.java b/src/main/java/com/apify/client/http/GzipHttpCompressor.java
new file mode 100644
index 0000000..8c9d1c6
--- /dev/null
+++ b/src/main/java/com/apify/client/http/GzipHttpCompressor.java
@@ -0,0 +1,53 @@
+package com.apify.client.http;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.zip.GZIPOutputStream;
+
+/** An {@link HttpCompressor} that gzip-encodes request bodies ({@code Content-Encoding: gzip}). */
+public final class GzipHttpCompressor implements HttpCompressor {
+
+ /** The highest compression level, and the default. */
+ private static final int MAX_LEVEL = 9;
+
+ private static final int MIN_LEVEL = 1;
+
+ private final int level;
+
+ /** A compressor with the default level, 9. */
+ public GzipHttpCompressor() {
+ this(MAX_LEVEL);
+ }
+
+ /**
+ * A compressor with the given level.
+ *
+ * @param level the compression level from 1 (fastest) to 9 (smallest output)
+ * @throws IllegalArgumentException if {@code level} is outside 1-9
+ */
+ public GzipHttpCompressor(int level) {
+ if (level < MIN_LEVEL || level > MAX_LEVEL) {
+ throw new IllegalArgumentException("gzip level must be between 1 and 9, was " + level);
+ }
+ this.level = level;
+ }
+
+ @Override
+ public String contentEncoding() {
+ return "gzip";
+ }
+
+ @Override
+ public byte[] compress(byte[] data) throws IOException {
+ ByteArrayOutputStream out = new ByteArrayOutputStream(data.length);
+ try (GZIPOutputStream gzip =
+ new GZIPOutputStream(out) {
+ {
+ def.setLevel(level);
+ }
+ }) {
+ gzip.write(data);
+ }
+ return out.toByteArray();
+ }
+}
diff --git a/src/main/java/com/apify/client/http/HttpCompressor.java b/src/main/java/com/apify/client/http/HttpCompressor.java
new file mode 100644
index 0000000..6b4a3fe
--- /dev/null
+++ b/src/main/java/com/apify/client/http/HttpCompressor.java
@@ -0,0 +1,28 @@
+package com.apify.client.http;
+
+import java.io.IOException;
+
+/**
+ * Compresses request bodies before they are sent. The client compresses a body only when it is at
+ * least 1 KiB, has a compressible content type and carries no {@code Content-Encoding} header of
+ * its own; the configured compressor then encodes it and its {@link #contentEncoding()} is sent as
+ * the {@code Content-Encoding} header. Streams are never compressed.
+ *
+ * Use {@link BrotliHttpCompressor} (the default, where its native library is available) or
+ * {@link GzipHttpCompressor}, or implement this interface for another coding. Configure it with
+ * {@link com.apify.client.ApifyClientBuilder#compression(HttpCompressor)}. Implementations must be
+ * safe for concurrent use.
+ */
+public interface HttpCompressor {
+
+ /** The {@code Content-Encoding} token that describes the output of {@link #compress}. */
+ String contentEncoding();
+
+ /**
+ * Compresses {@code data}. A failure fails the request: the client does not fall back to another
+ * coding.
+ *
+ * @throws IOException if the data cannot be compressed
+ */
+ byte[] compress(byte[] data) throws IOException;
+}
diff --git a/src/main/java/com/apify/client/http/RetryConfig.java b/src/main/java/com/apify/client/http/RetryConfig.java
index 4472593..01a8ce6 100644
--- a/src/main/java/com/apify/client/http/RetryConfig.java
+++ b/src/main/java/com/apify/client/http/RetryConfig.java
@@ -4,6 +4,16 @@
/** Retry/timeout policy for the orchestrating HTTP client. Internal to the client. */
public final class RetryConfig {
+
+ /** Default duration of the {@link Timeout#SHORT} tier. */
+ public static final Duration DEFAULT_TIMEOUT_SHORT = Duration.ofSeconds(5);
+
+ /** Default duration of the {@link Timeout#MEDIUM} tier. */
+ public static final Duration DEFAULT_TIMEOUT_MEDIUM = Duration.ofSeconds(30);
+
+ /** Default duration of the {@link Timeout#LONG} tier. */
+ public static final Duration DEFAULT_TIMEOUT_LONG = Duration.ofSeconds(360);
+
/**
* Maximum number of retries (the request is attempted up to {@code maxRetries + 1} times). Public
* (not just within this package) so the internal HTTP client, which lives in a separate,
@@ -17,21 +27,52 @@ public final class RetryConfig {
/** Upper bound on the (exponentially growing) inter-retry delay. */
public final Duration maxDelayBetweenRetries;
+ /** Duration of the {@link Timeout#SHORT} tier. */
+ public final Duration timeoutShort;
+
+ /** Duration of the {@link Timeout#MEDIUM} tier. */
+ public final Duration timeoutMedium;
+
+ /** Duration of the {@link Timeout#LONG} tier. */
+ public final Duration timeoutLong;
+
/**
- * Maximum per-attempt request (socket) timeout, and the default base timeout for a call. Each
- * attempt's timeout may grow from a smaller per-call base but is capped at this value. This
- * bounds a single attempt, not the cumulative time across all retries.
+ * Cap on the timeout of a single request attempt. The first attempt gets the requested timeout,
+ * each retry doubles it, and none exceeds this value. This bounds a single attempt, not the
+ * cumulative time across all retries.
*/
public final Duration timeout;
+ /** A policy with the default tier durations and the given cap on a single attempt. */
public RetryConfig(
int maxRetries,
Duration minDelayBetweenRetries,
Duration maxDelayBetweenRetries,
Duration timeout) {
+ this(
+ maxRetries,
+ minDelayBetweenRetries,
+ maxDelayBetweenRetries,
+ DEFAULT_TIMEOUT_SHORT,
+ DEFAULT_TIMEOUT_MEDIUM,
+ DEFAULT_TIMEOUT_LONG,
+ timeout);
+ }
+
+ public RetryConfig(
+ int maxRetries,
+ Duration minDelayBetweenRetries,
+ Duration maxDelayBetweenRetries,
+ Duration timeoutShort,
+ Duration timeoutMedium,
+ Duration timeoutLong,
+ Duration timeoutMax) {
this.maxRetries = maxRetries;
this.minDelayBetweenRetries = minDelayBetweenRetries;
this.maxDelayBetweenRetries = maxDelayBetweenRetries;
- this.timeout = timeout;
+ this.timeoutShort = timeoutShort;
+ this.timeoutMedium = timeoutMedium;
+ this.timeoutLong = timeoutLong;
+ this.timeout = timeoutMax;
}
}
diff --git a/src/main/java/com/apify/client/http/Timeout.java b/src/main/java/com/apify/client/http/Timeout.java
new file mode 100644
index 0000000..4ad80b0
--- /dev/null
+++ b/src/main/java/com/apify/client/http/Timeout.java
@@ -0,0 +1,110 @@
+package com.apify.client.http;
+
+import java.time.Duration;
+import java.util.Optional;
+
+/**
+ * The timeout of a single API request attempt: one of the preconfigured {@linkplain Tier tiers}, an
+ * exact duration, or {@link #NO_TIMEOUT}.
+ *
+ *
Every client method picks the tier that fits the expected duration of its request: {@link
+ * #SHORT} for simple metadata reads and writes, {@link #MEDIUM} for listing, batch and trigger
+ * operations, {@link #LONG} for downloads, uploads and streaming. The duration of each tier is set
+ * on the {@link com.apify.client.ApifyClientBuilder}. Pass a {@code Timeout} to {@link
+ * com.apify.client.ApifyClient#withTimeout(Timeout)} to replace the tier of every request a client
+ * sends.
+ *
+ *
A tier or an explicit duration above the builder's {@code timeoutMax} is capped at it, and the
+ * client logs a warning when that happens. The timeout applies to each attempt separately and
+ * doubles with every retry, up to the same cap. With {@link #NO_TIMEOUT} the client never aborts
+ * the request, so a connection that stalls is not retried either.
+ */
+public final class Timeout {
+
+ /** The preconfigured timeout tiers. */
+ public enum Tier {
+ /** Simple metadata reads and writes (default 5 seconds). */
+ SHORT,
+ /** Listing, batch and trigger operations (default 30 seconds). */
+ MEDIUM,
+ /** Downloads, uploads and streaming (default 360 seconds). */
+ LONG
+ }
+
+ /** The {@link Tier#SHORT} tier. */
+ public static final Timeout SHORT = new Timeout(Tier.SHORT, null, false);
+
+ /** The {@link Tier#MEDIUM} tier. */
+ public static final Timeout MEDIUM = new Timeout(Tier.MEDIUM, null, false);
+
+ /** The {@link Tier#LONG} tier. */
+ public static final Timeout LONG = new Timeout(Tier.LONG, null, false);
+
+ /** Lets the request run for as long as it takes. */
+ public static final Timeout NO_TIMEOUT = new Timeout(null, null, true);
+
+ private final Tier tier;
+ private final Duration duration;
+ private final boolean noTimeout;
+
+ private Timeout(Tier tier, Duration duration, boolean noTimeout) {
+ this.tier = tier;
+ this.duration = duration;
+ this.noTimeout = noTimeout;
+ }
+
+ /**
+ * An exact per-attempt timeout.
+ *
+ * @throws IllegalArgumentException if {@code duration} is null, zero or negative; use {@link
+ * #NO_TIMEOUT} to run without a timeout
+ */
+ public static Timeout of(Duration duration) {
+ if (duration == null || duration.isZero() || duration.isNegative()) {
+ throw new IllegalArgumentException(
+ "timeout must be strictly positive (use Timeout.NO_TIMEOUT to run without one)");
+ }
+ return new Timeout(null, duration, false);
+ }
+
+ /** An exact per-attempt timeout in seconds; see {@link #of(Duration)}. */
+ public static Timeout ofSeconds(long seconds) {
+ return of(Duration.ofSeconds(seconds));
+ }
+
+ /** The tier this timeout names, or empty for an exact duration and for {@link #NO_TIMEOUT}. */
+ public Optional tier() {
+ return Optional.ofNullable(tier);
+ }
+
+ /** The exact duration, or empty for a tier and for {@link #NO_TIMEOUT}. */
+ public Optional duration() {
+ return Optional.ofNullable(duration);
+ }
+
+ /** Whether this is {@link #NO_TIMEOUT}. */
+ public boolean isNoTimeout() {
+ return noTimeout;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof Timeout other
+ && tier == other.tier
+ && noTimeout == other.noTimeout
+ && java.util.Objects.equals(duration, other.duration);
+ }
+
+ @Override
+ public int hashCode() {
+ return java.util.Objects.hash(tier, duration, noTimeout);
+ }
+
+ @Override
+ public String toString() {
+ if (noTimeout) {
+ return "Timeout.NO_TIMEOUT";
+ }
+ return tier != null ? "Timeout." + tier : "Timeout.of(" + duration + ")";
+ }
+}
diff --git a/src/main/java/com/apify/client/internal/Async.java b/src/main/java/com/apify/client/internal/Async.java
index e773c8e..4c0aea8 100644
--- a/src/main/java/com/apify/client/internal/Async.java
+++ b/src/main/java/com/apify/client/internal/Async.java
@@ -4,6 +4,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
@@ -32,11 +33,22 @@ public final class Async {
private Async() {}
- /** Returns a future that completes after {@code delay} has elapsed (never negative/blocking). */
+ /**
+ * Returns a future that completes after {@code delay} has elapsed (never negative/blocking).
+ * Cancelling it cancels the timer.
+ */
public static CompletableFuture delay(Duration delay) {
- CompletableFuture future = new CompletableFuture<>();
+ CancellableFuture future = new CancellableFuture<>();
long millis = Math.max(0, delay.toMillis());
- SCHEDULER.schedule(() -> future.complete(null), millis, TimeUnit.MILLISECONDS);
+ ScheduledFuture timer =
+ SCHEDULER.schedule(() -> future.complete(null), millis, TimeUnit.MILLISECONDS);
+ // A cancelled delay releases its timer instead of firing into a future nobody waits for.
+ future.whenComplete(
+ (unused, error) -> {
+ if (future.isCancelled()) {
+ timer.cancel(false);
+ }
+ });
return future;
}
}
diff --git a/src/main/java/com/apify/client/internal/AsyncPaginatedPublisher.java b/src/main/java/com/apify/client/internal/AsyncPaginatedPublisher.java
index 155cfbf..b3fbf54 100644
--- a/src/main/java/com/apify/client/internal/AsyncPaginatedPublisher.java
+++ b/src/main/java/com/apify/client/internal/AsyncPaginatedPublisher.java
@@ -112,6 +112,7 @@ private final class Session implements Flow.Subscription {
private long yielded;
private boolean exhausted;
private boolean fetchInFlight;
+ private volatile CompletableFuture inFlight;
Session(Flow.Subscriber subscriber) {
this.subscriber = subscriber;
@@ -141,6 +142,11 @@ public void request(long n) {
@Override
public void cancel() {
cancelled.set(true);
+ // Abort the page being fetched, with its request and retries.
+ CompletableFuture fetch = inFlight;
+ if (fetch != null) {
+ fetch.cancel(true);
+ }
}
/** Enters the drain loop unless another drain (of this same session) is already running. */
@@ -203,22 +209,22 @@ private void fetchNextPage() {
fetchInFlight = true;
Long capRemaining = totalLimit != null ? totalLimit - yielded : null;
Long pageLimit = minForLimit(capRemaining, chunkSize);
- fetcher
- .fetch(offset, pageLimit)
- .whenComplete(
- (page, error) -> {
- fetchInFlight = false;
- if (cancelled.get()) {
- draining.set(false);
- return;
- }
- if (error != null) {
- terminate(() -> subscriber.onError(HttpClientCore.unwrapCompletion(error)));
- return;
- }
- applyPage(page);
- drainLoop();
- });
+ CompletableFuture> fetch = fetcher.fetch(offset, pageLimit);
+ inFlight = fetch;
+ fetch.whenComplete(
+ (page, error) -> {
+ fetchInFlight = false;
+ if (cancelled.get()) {
+ draining.set(false);
+ return;
+ }
+ if (error != null) {
+ terminate(() -> subscriber.onError(HttpClientCore.unwrapCompletion(error)));
+ return;
+ }
+ applyPage(page);
+ drainLoop();
+ });
}
private void applyPage(PaginationList page) {
diff --git a/src/main/java/com/apify/client/internal/CancellableFuture.java b/src/main/java/com/apify/client/internal/CancellableFuture.java
new file mode 100644
index 0000000..ea3938e
--- /dev/null
+++ b/src/main/java/com/apify/client/internal/CancellableFuture.java
@@ -0,0 +1,162 @@
+package com.apify.client.internal;
+
+import java.lang.ref.WeakReference;
+import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.function.Function;
+
+/**
+ * A {@link CompletableFuture} whose cancellation reaches the work behind it. Cancelling a plain
+ * {@code CompletableFuture} completes only that future: the request it waits for, the retry or poll
+ * chain that produced it and the timers that drive it keep running. Every future the client returns
+ * is a {@code CancellableFuture}, and so is every stage derived from it with {@code thenApply},
+ * {@code thenCompose}, {@code handle}, ...: cancelling the future a caller holds cancels each
+ * upstream stage (and the future {@code thenCompose} is waiting on) down to the HTTP request in
+ * flight, and stops the retries, polls and pages that would follow. Internal to the client.
+ */
+public final class CancellableFuture extends CompletableFuture {
+
+ /**
+ * The futures this one depends on, cancelled together with it. Held weakly: a future that is
+ * still pending is kept alive by whatever will complete it (the request in flight, the timer, the
+ * stage it feeds), and holding a completed one strongly would keep its result (a response body,
+ * say) reachable from every later stage for as long as the call lives, which for a long {@code
+ * waitForFinish} poll loop is unbounded.
+ */
+ private final ConcurrentLinkedQueue>> upstream =
+ new ConcurrentLinkedQueue<>();
+
+ public CancellableFuture() {}
+
+ /**
+ * A future that completes as {@code source} does and, when cancelled, cancels {@code source}.
+ * Wraps the future a transport returns, so cancelling a call aborts the request in flight.
+ */
+ public static CancellableFuture wrap(CompletionStage source) {
+ CancellableFuture result = new CancellableFuture<>();
+ CompletableFuture sourceFuture = source.toCompletableFuture();
+ result.link(sourceFuture);
+ sourceFuture.whenComplete(
+ (value, error) -> {
+ if (error == null) {
+ result.complete(value);
+ } else {
+ result.completeExceptionally(error);
+ }
+ });
+ return result;
+ }
+
+ /** An already completed future. */
+ public static CancellableFuture completed(T value) {
+ CancellableFuture future = new CancellableFuture<>();
+ future.complete(value);
+ return future;
+ }
+
+ /** An already failed future. */
+ public static CancellableFuture failed(Throwable error) {
+ CancellableFuture future = new CancellableFuture<>();
+ future.completeExceptionally(error);
+ return future;
+ }
+
+ /** A future that completes once all of {@code futures} have, and cancels them when cancelled. */
+ public static CancellableFuture allOf(List> futures) {
+ CancellableFuture result = new CancellableFuture<>();
+ futures.forEach(result::link);
+ CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
+ .whenComplete(
+ (unused, error) -> {
+ if (error == null) {
+ result.complete(null);
+ } else {
+ result.completeExceptionally(error);
+ }
+ });
+ return result;
+ }
+
+ /** Makes {@code other} be cancelled whenever this future is (now, if it already is). */
+ public void link(CompletableFuture other) {
+ upstream.add(new WeakReference<>(other));
+ if (isCancelled()) {
+ other.cancel(true);
+ }
+ }
+
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ boolean cancelled = super.cancel(mayInterruptIfRunning);
+ if (cancelled) {
+ for (WeakReference> reference : upstream) {
+ CompletableFuture source = reference.get();
+ if (source != null) {
+ source.cancel(mayInterruptIfRunning);
+ }
+ }
+ }
+ return cancelled;
+ }
+
+ /** Every stage derived from this one cancels this one when it is cancelled. */
+ @Override
+ public CompletableFuture newIncompleteFuture() {
+ CancellableFuture dependent = new CancellableFuture<>();
+ dependent.upstream.add(new WeakReference<>(this));
+ return dependent;
+ }
+
+ /**
+ * As {@link CompletableFuture#thenCompose}, additionally linking the future {@code fn} returns,
+ * so cancelling the composed stage cancels the work it was waiting on, and skipping {@code fn}
+ * altogether once the composed stage is cancelled (so a cancelled call sends no further request).
+ */
+ @Override
+ public CompletableFuture thenCompose(
+ Function> fn) {
+ Composition composition = new Composition<>();
+ CompletableFuture composed =
+ super.thenCompose(
+ value -> {
+ CancellableFuture dependent = composition.dependent;
+ if (dependent != null && dependent.isCancelled()) {
+ return failed(new CancellationException());
+ }
+ CompletionStage inner = fn.apply(value);
+ CompletableFuture innerFuture = inner.toCompletableFuture();
+ synchronized (composition) {
+ dependent = composition.dependent;
+ if (dependent == null) {
+ composition.inner = innerFuture;
+ }
+ }
+ if (dependent != null) {
+ dependent.link(innerFuture);
+ }
+ return inner;
+ });
+ CancellableFuture dependent = (CancellableFuture) composed;
+ CompletableFuture early;
+ synchronized (composition) {
+ composition.dependent = dependent;
+ early = composition.inner;
+ composition.inner = null;
+ }
+ if (early != null) {
+ dependent.link(early);
+ }
+ return composed;
+ }
+
+ /**
+ * Hands the composed stage and the future {@code fn} returned to each other, whichever is first.
+ */
+ private static final class Composition {
+ volatile CancellableFuture dependent;
+ CompletableFuture inner;
+ }
+}
diff --git a/src/main/java/com/apify/client/internal/DatasetItemsOptionsAccess.java b/src/main/java/com/apify/client/internal/DatasetItemsOptionsAccess.java
new file mode 100644
index 0000000..6139505
--- /dev/null
+++ b/src/main/java/com/apify/client/internal/DatasetItemsOptionsAccess.java
@@ -0,0 +1,56 @@
+package com.apify.client.internal;
+
+import com.apify.client.dataset.DatasetListItemsOptions;
+
+/**
+ * Cross-package access to the item options of {@link DatasetListItemsOptions}, whose accessors are
+ * package-private because only the client itself needs them. {@code DatasetListItemsOptions}
+ * registers its implementation when it is loaded, so code in another package (the run client's
+ * dataset iteration) can read and derive options without those accessors being public API. Internal
+ * to the client.
+ */
+public interface DatasetItemsOptionsAccess {
+
+ /** The registered implementation; the options class is loaded before it is looked up. */
+ static DatasetItemsOptionsAccess get() {
+ // Loading the options class runs its static initializer, which registers the implementation.
+ new DatasetListItemsOptions();
+ return Holder.instance;
+ }
+
+ /** Registers the implementation; called once by {@link DatasetListItemsOptions}. */
+ static void register(DatasetItemsOptionsAccess access) {
+ Holder.instance = access;
+ }
+
+ /** The first row {@code options} asks for, or 0. */
+ long offset(DatasetListItemsOptions options);
+
+ /** The maximum number of rows {@code options} asks for, or {@code null}. */
+ Long limit(DatasetListItemsOptions options);
+
+ /**
+ * Whether {@code clean}, {@code skipEmpty} or {@code unwind} is set: the server-side filters that
+ * can leave a page of existing rows without a single returned item.
+ */
+ boolean dropsRows(DatasetListItemsOptions options);
+
+ /**
+ * An independent copy of {@code options} (the defaults for {@code null}) for reading rows in the
+ * order they were pushed, so a later change of the caller's options cannot reach the iteration.
+ */
+ DatasetListItemsOptions snapshot(DatasetListItemsOptions options);
+
+ /** A copy of {@code options} for the given page of rows, in the order they were pushed. */
+ DatasetListItemsOptions page(DatasetListItemsOptions options, long offset, long limit);
+
+ /** Options for the given page of rows with none of the item filters of the caller. */
+ DatasetListItemsOptions plainPage(long offset, long limit);
+
+ /** Holds the registered implementation. */
+ final class Holder {
+ private static volatile DatasetItemsOptionsAccess instance;
+
+ private Holder() {}
+ }
+}
diff --git a/src/main/java/com/apify/client/internal/HttpClientCore.java b/src/main/java/com/apify/client/internal/HttpClientCore.java
index 4eaea99..2b3554c 100644
--- a/src/main/java/com/apify/client/internal/HttpClientCore.java
+++ b/src/main/java/com/apify/client/internal/HttpClientCore.java
@@ -1,20 +1,21 @@
package com.apify.client.internal;
-import com.aayushatharva.brotli4j.Brotli4jLoader;
-import com.aayushatharva.brotli4j.encoder.Encoder;
import com.apify.client.http.ApiResponse;
import com.apify.client.http.ApifyApiException;
import com.apify.client.http.ApifyTransportException;
+import com.apify.client.http.BrotliHttpCompressor;
import com.apify.client.http.ConflictError;
import com.apify.client.http.ForbiddenError;
+import com.apify.client.http.GzipHttpCompressor;
+import com.apify.client.http.HttpCompressor;
import com.apify.client.http.HttpTransport;
import com.apify.client.http.InvalidRequestError;
import com.apify.client.http.NotFoundError;
import com.apify.client.http.RateLimitError;
import com.apify.client.http.RetryConfig;
import com.apify.client.http.ServerError;
+import com.apify.client.http.Timeout;
import com.apify.client.http.UnauthorizedError;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
@@ -27,11 +28,14 @@
import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadLocalRandom;
-import java.util.zip.GZIPOutputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* The orchestrating HTTP client shared by every resource client. It owns the transport, the
@@ -64,84 +68,200 @@ public final class HttpClientCore {
*/
private static final int MIN_COMPRESS_BYTES = 1024;
- /** {@code Content-Encoding} value for brotli-compressed bodies (matches the reference client). */
- private static final String ENCODING_BROTLI = "br";
-
- /** {@code Content-Encoding} value for gzip-compressed bodies (the fallback coding). */
- private static final String ENCODING_GZIP = "gzip";
-
/** Prefix required before the token value in the {@code Authorization} header. */
private static final String AUTH_BEARER_PREFIX = "Bearer ";
+ /** The API holds a {@code waitForFinish} request for at most this many seconds. */
+ private static final long MAX_HOLD_SECS = 60;
+
/**
- * Whether a brotli native codec loaded for the running platform. Resolved once at class load:
- * brotli4j needs a platform-specific native library, so on platforms without one (or if loading
- * fails) the client falls back to gzip. Mirrors the reference JS client, which prefers brotli and
- * falls back to gzip.
+ * Seconds of a caller-chosen request timeout kept free of a {@code waitForFinish} hold, so the
+ * API answers before the client's own timeout fires.
*/
- private static final boolean BROTLI_AVAILABLE = detectBrotli();
+ private static final long HOLD_MARGIN_SECS = 5;
- private static boolean detectBrotli() {
- try {
- Brotli4jLoader.ensureAvailability();
- return true;
- } catch (Throwable t) {
- // Native codec unavailable (no bundled binary for this OS/arch, or a link error). Catch
- // Throwable because native loading can raise UnsatisfiedLinkError/NoClassDefFoundError, not
- // just Exception. The client stays fully functional using gzip.
- return false;
- }
- }
+ private static final double MILLIS_PER_SECOND = 1000.0;
- /**
- * Reports whether the brotli path is active on this platform. Public so tests outside this
- * package (which lives in the non-exported {@code com.apify.client.internal} package) can assert
- * on it.
- */
- public static boolean brotliAvailable() {
- return BROTLI_AVAILABLE;
- }
+ private static final Logger LOG = LoggerFactory.getLogger(HttpClientCore.class);
private final HttpTransport transport;
private final String token;
private final String userAgent;
private final RetryConfig retry;
+ private final HttpCompressor compressor;
+ private final Map defaultHeaders;
+
+ /** When set, replaces the timeout every request would otherwise get. */
+ private final Timeout timeoutOverride;
+
+ /** Requested timeouts already warned about, shared by every copy made by {@link #withTimeout}. */
+ private final Set warnedCappedTimeouts;
+ /**
+ * @param compressor compresses large request bodies; {@code null} picks brotli where its native
+ * library is available and gzip elsewhere
+ * @param defaultHeaders headers sent with every request; they replace the built-in {@code
+ * User-Agent} and {@code Authorization} headers of the same name and are replaced by a header
+ * the request sets itself
+ */
public HttpClientCore(
- HttpTransport transport, String token, String userAgent, RetryConfig retry) {
+ HttpTransport transport,
+ String token,
+ String userAgent,
+ RetryConfig retry,
+ HttpCompressor compressor,
+ Map defaultHeaders) {
+ this(
+ transport,
+ token,
+ userAgent,
+ retry,
+ compressor != null ? compressor : defaultCompressor(),
+ defaultHeaders == null ? Map.of() : Map.copyOf(defaultHeaders),
+ null,
+ ConcurrentHashMap.newKeySet());
+ }
+
+ private static HttpCompressor defaultCompressor() {
+ return BrotliHttpCompressor.isAvailable()
+ ? new BrotliHttpCompressor()
+ : new GzipHttpCompressor();
+ }
+
+ private HttpClientCore(
+ HttpTransport transport,
+ String token,
+ String userAgent,
+ RetryConfig retry,
+ HttpCompressor compressor,
+ Map defaultHeaders,
+ Timeout timeoutOverride,
+ Set warnedCappedTimeouts) {
this.transport = transport;
this.token = token;
this.userAgent = userAgent;
this.retry = retry;
+ this.compressor = compressor;
+ this.defaultHeaders = defaultHeaders;
+ this.timeoutOverride = timeoutOverride;
+ this.warnedCappedTimeouts = warnedCappedTimeouts;
}
public String userAgent() {
return userAgent;
}
- /** The configured maximum per-attempt request timeout, in whole seconds. */
- public long requestTimeoutSeconds() {
- return retry.timeout.getSeconds();
+ /**
+ * A copy of this client that sends every request with {@code timeout} instead of the tier the
+ * calling method would pick. The transport, token and retry policy are shared.
+ */
+ public HttpClientCore withTimeout(Timeout timeout) {
+ if (timeout == null) {
+ throw new IllegalArgumentException("timeout must not be null");
+ }
+ return new HttpClientCore(
+ transport,
+ token,
+ userAgent,
+ retry,
+ compressor,
+ defaultHeaders,
+ timeout,
+ warnedCappedTimeouts);
}
/**
- * The default per-attempt request timeout used as the base for standard API calls. It is the
- * client's configured timeout (single source of truth), so a caller-configured timeout applies to
- * the first attempt too. Individual calls may pass a smaller base (e.g. key-value-store uploads),
- * which then grows back up toward this value on retries.
+ * This client without the override set by {@link #withTimeout}: every request gets the timeout of
+ * its own method again.
*/
- public Duration baseRequestTimeout() {
- return retry.timeout;
+ public HttpClientCore withoutTimeoutOverride() {
+ if (timeoutOverride == null) {
+ return this;
+ }
+ return new HttpClientCore(
+ transport, token, userAgent, retry, compressor, defaultHeaders, null, warnedCappedTimeouts);
}
+ /**
+ * Resolves a timeout to the duration of its first attempt: a tier to its configured duration, an
+ * exact timeout to itself, and {@link Timeout#NO_TIMEOUT} to {@code null} (no timeout). A value
+ * above the configured maximum is capped at it, which logs a warning once per value, since the
+ * request does not get the whole time it asked for. The override set by {@link #withTimeout}
+ * takes precedence over {@code requested}.
+ */
+ public Duration resolveTimeout(Timeout requested) {
+ Timeout effective = timeoutOverride != null ? timeoutOverride : requested;
+ if (effective.isNoTimeout()) {
+ return null;
+ }
+ Duration duration =
+ effective
+ .duration()
+ .orElseGet(
+ () ->
+ switch (effective.tier().orElseThrow()) {
+ case SHORT -> retry.timeoutShort;
+ case MEDIUM -> retry.timeoutMedium;
+ case LONG -> retry.timeoutLong;
+ });
+ if (duration.compareTo(retry.timeout) > 0) {
+ if (warnedCappedTimeouts.add(duration)) {
+ LOG.warn(
+ "The requested timeout of {}s exceeds timeoutMax ({}s) and is capped at it. Raise"
+ + " timeoutMax on the client builder to allow longer request timeouts.",
+ duration.toMillis() / MILLIS_PER_SECOND,
+ retry.timeout.toMillis() / MILLIS_PER_SECOND);
+ }
+ return retry.timeout;
+ }
+ return duration;
+ }
+
+ /**
+ * The {@code waitForFinish} value to send and the timeout to send it with, for a request that
+ * asks the API to hold its response for up to {@code requestedHold} seconds.
+ *
+ * The hold is at most {@link #MAX_HOLD_SECS} and is cut down to what the request timeout can
+ * cover, so the client never aborts a request the API is still holding (which, for a start or a
+ * build, would retry a call that creates a resource). Without an override the timeout is the
+ * tier's plus the hold, and the hold is limited by what {@code timeoutMax} leaves above the tier.
+ * An override set by {@link #withTimeout} is kept as given, and the hold is limited to it less a
+ * safety margin.
+ */
+ public WaitHold waitHold(Timeout tier, Long requestedHold) {
+ if (requestedHold == null) {
+ return new WaitHold(null, tier);
+ }
+ long hold = Math.min(Math.max(requestedHold, 0), MAX_HOLD_SECS);
+ Duration base = resolveTimeout(tier);
+ if (base == null) {
+ return new WaitHold(hold, tier);
+ }
+ if (timeoutOverride != null) {
+ long coverable = Math.max(0, base.toSeconds() - HOLD_MARGIN_SECS);
+ return new WaitHold(Math.min(hold, coverable), tier);
+ }
+ hold = Math.min(hold, Math.max(0, retry.timeout.toSeconds() - base.toSeconds()));
+ return new WaitHold(hold, hold > 0 ? Timeout.of(base.plusSeconds(hold)) : tier);
+ }
+
+ /**
+ * The outcome of {@link #waitHold}.
+ *
+ * @param waitForFinishSecs the {@code waitForFinish} query value to send, or {@code null} for
+ * none
+ * @param timeout the timeout to send the request with
+ */
+ public record WaitHold(Long waitForFinishSecs, Timeout timeout) {}
+
/** Sends a request with auth, User-Agent and the retry policy applied. */
public CompletableFuture call(
- String method, String url, byte[] body, String contentType, Duration baseTimeout) {
- return call(method, url, body, contentType, null, baseTimeout, false);
+ String method, String url, byte[] body, String contentType, Timeout timeout) {
+ return call(method, url, body, contentType, null, timeout, false);
}
/**
- * Like {@link #call(String, String, byte[], String, Duration)} but, when {@code
+ * Like {@link #call(String, String, byte[], String, Timeout)} but, when {@code
* doNotRetryTimeouts} is {@code true}, a transport timeout is treated as a terminal failure
* rather than being retried (other transport/network errors and retryable statuses are still
* retried).
@@ -151,13 +271,13 @@ public CompletableFuture call(
String url,
byte[] body,
String contentType,
- Duration baseTimeout,
+ Timeout timeout,
boolean doNotRetryTimeouts) {
- return call(method, url, body, contentType, null, baseTimeout, doNotRetryTimeouts);
+ return call(method, url, body, contentType, null, timeout, doNotRetryTimeouts);
}
/**
- * Like {@link #call(String, String, byte[], String, Duration)} but additionally sets the given
+ * Like {@link #call(String, String, byte[], String, Timeout)} but additionally sets the given
* extra headers on every attempt.
*/
public CompletableFuture call(
@@ -166,8 +286,8 @@ public CompletableFuture call(
byte[] body,
String contentType,
Map extraHeaders,
- Duration baseTimeout) {
- return call(method, url, body, contentType, extraHeaders, baseTimeout, false);
+ Timeout timeout) {
+ return call(method, url, body, contentType, extraHeaders, timeout, false);
}
/** The canonical overload every other {@code call} convenience overload delegates to. */
@@ -177,17 +297,24 @@ public CompletableFuture call(
byte[] body,
String contentType,
Map extraHeaders,
- Duration baseTimeout,
+ Timeout timeout,
boolean doNotRetryTimeouts) {
+ 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)) {
- Compressed compressed = compress(requestBody, BROTLI_AVAILABLE);
- requestBody = compressed.body();
+ 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, compressed.encoding());
+ headers.put(HttpHeaders.CONTENT_ENCODING, compressor.contentEncoding());
}
String path = extractPath(url);
@@ -284,6 +411,10 @@ private CompletableFuture attempt(
null, false, isStatusRetryable(resp.statusCode()), apiError);
}
Throwable cause = unwrapCompletion(err);
+ if (cause instanceof CancellationException cancellation) {
+ // The caller cancelled the call: end it, never retry.
+ return new AttemptOutcome(null, false, false, cancellation);
+ }
ApifyTransportException transportError =
cause instanceof ApifyTransportException apifyTransportException
? apifyTransportException
@@ -309,8 +440,10 @@ HttpRequest buildRequest(
body != null
? HttpRequest.BodyPublishers.ofByteArray(body)
: HttpRequest.BodyPublishers.noBody();
- HttpRequest.Builder b =
- HttpRequest.newBuilder(URI.create(url)).method(method, publisher).timeout(timeout);
+ HttpRequest.Builder b = HttpRequest.newBuilder(URI.create(url)).method(method, publisher);
+ if (timeout != null) {
+ b.timeout(timeout);
+ }
b.header(HttpHeaders.USER_AGENT, userAgent);
if (token != null && !token.isEmpty()) {
b.header(HttpHeaders.AUTHORIZATION, AUTH_BEARER_PREFIX + token);
@@ -318,8 +451,11 @@ HttpRequest buildRequest(
if (contentType != null && !contentType.isEmpty()) {
b.header(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);
if (extraHeaders != null) {
- extraHeaders.forEach(b::header);
+ extraHeaders.forEach(b::setHeader);
}
return b.build();
}
@@ -332,8 +468,7 @@ private CompletableFuture doAttempt(
Map extraHeaders,
Duration timeout) {
HttpRequest request = buildRequest(method, url, body, contentType, extraHeaders, timeout);
- return transport
- .sendAsync(request)
+ return CancellableFuture.wrap(transport.sendAsync(request))
.thenApply(resp -> new ApiResponse(resp.statusCode(), resp.headers(), resp.body()));
}
@@ -346,6 +481,9 @@ private CompletableFuture doAttempt(
* across all retries.
*/
private Duration attemptTimeout(Duration base, int attempt) {
+ if (base == null) {
+ return null; // Timeout.NO_TIMEOUT
+ }
Duration scaled = base;
for (int i = 1; i < attempt; i++) {
scaled = scaled.multipliedBy(2);
@@ -436,8 +574,8 @@ private static boolean shouldCompress(
* ending in {@code +json}/{@code +xml} (e.g. {@code image/svg+xml}). A missing content type is
* assumed compressible. Matches the reference JS client's {@code isCompressibleContentType}.
*
- * Public (like {@link #compress} and {@link #brotliAvailable}) so {@code CompressionTest},
- * outside this non-exported package, can exercise the classification directly.
+ *
Public so {@code CompressionTest}, outside this non-exported package, can exercise the
+ * classification directly.
*/
public static boolean isCompressibleContentType(String contentType) {
if (contentType == null || contentType.isEmpty()) {
@@ -468,70 +606,6 @@ public static boolean isCompressibleContentType(String contentType) {
return true;
}
- /**
- * A compressed request body together with the {@code Content-Encoding} token that describes it.
- *
- * @param body the compressed request body
- * @param encoding the {@code Content-Encoding} token describing {@code body}
- */
- public record Compressed(byte[] body, String encoding) {}
-
- /**
- * Compresses a request body, preferring brotli and falling back to gzip, matching the reference
- * JS client. When {@code preferBrotli} is {@code true} the body is brotli-encoded ({@code
- * Content-Encoding: br}); otherwise it is gzip-encoded ({@code Content-Encoding: gzip}). Callers
- * pass {@link #BROTLI_AVAILABLE}; making the coding an explicit parameter keeps this a pure
- * function of its inputs rather than of hidden static state. Public so {@code CompressionTest}
- * (outside this non-exported package) can exercise it directly.
- *
- *
If the brotli path itself fails despite the native codec having loaded, this falls back to
- * gzip rather than failing the whole request - matching the reference JS client, which since apify-client-js#990 keys the
- * fallback on compression actually failing rather than only on an upfront availability check.
- * Catches {@code Throwable}, not just {@code RuntimeException}: a native codec that loaded
- * successfully at startup can still fail a specific call with an {@code Error} (e.g. a partial or
- * mismatched native library surfacing {@link UnsatisfiedLinkError} only once a method is actually
- * invoked), the same reason {@link #detectBrotli()} itself catches {@code Throwable} rather than
- * {@code Exception}.
- */
- public static Compressed compress(byte[] data, boolean preferBrotli) {
- if (preferBrotli) {
- try {
- return new Compressed(brotli(data), ENCODING_BROTLI);
- } catch (Throwable t) {
- return new Compressed(gzip(data), ENCODING_GZIP);
- }
- }
- return new Compressed(gzip(data), ENCODING_GZIP);
- }
-
- /** Brotli-compresses a request body using the loaded native codec. */
- private static byte[] brotli(byte[] data) {
- try {
- return Encoder.compress(data);
- } catch (IOException e) {
- // Encoding an in-memory byte[] performs no real I/O, so this is unreachable in practice.
- throw new ApifyTransportException(e);
- }
- }
-
- /**
- * Gzip-compresses a request body. Used as the fallback coding when no brotli native codec is
- * available for the running platform; the JDK always ships a gzip codec ({@link
- * GZIPOutputStream}).
- */
- private static byte[] gzip(byte[] data) {
- ByteArrayOutputStream out = new ByteArrayOutputStream(data.length);
- try (GZIPOutputStream gz = new GZIPOutputStream(out)) {
- gz.write(data);
- } catch (IOException e) {
- // Compressing an in-memory byte[] cannot perform real I/O, so this is unreachable in
- // practice.
- throw new ApifyTransportException(e);
- }
- return out.toByteArray();
- }
-
private static boolean isStatusRetryable(int status) {
return status == RATE_LIMIT_EXCEEDED || status >= MIN_SERVER_ERROR;
}
@@ -655,18 +729,22 @@ public static String extractPath(String url) {
* matching {@link HttpTransport}'s contract.
*/
public CompletableFuture> streamAsync(String url) {
- HttpRequest request = buildRequest("GET", url, null, null, null, retry.timeout);
- CompletableFuture> result = new CompletableFuture<>();
- transport
- .sendStreamingAsync(request)
- .whenComplete(
- (resp, err) -> {
- if (err == null) {
- result.complete(resp);
- } else {
- result.completeExceptionally(new ApifyTransportException(unwrapCompletion(err)));
- }
- });
+ HttpRequest request = buildRequest("GET", url, null, null, null, resolveTimeout(Timeout.LONG));
+ CancellableFuture> result = new CancellableFuture<>();
+ CompletableFuture> exchange = transport.sendStreamingAsync(request);
+ // Cancelling the returned future aborts the exchange.
+ result.link(exchange);
+ exchange.whenComplete(
+ (resp, err) -> {
+ Throwable cause = err == null ? null : unwrapCompletion(err);
+ if (cause == null) {
+ result.complete(resp);
+ } else if (cause instanceof CancellationException) {
+ result.cancel(true);
+ } else {
+ result.completeExceptionally(new ApifyTransportException(cause));
+ }
+ });
return result;
}
diff --git a/src/main/java/com/apify/client/internal/Json.java b/src/main/java/com/apify/client/internal/Json.java
index f1d1096..bcf6155 100644
--- a/src/main/java/com/apify/client/internal/Json.java
+++ b/src/main/java/com/apify/client/internal/Json.java
@@ -68,6 +68,31 @@ public static byte[] toBytes(Object value) {
return MAPPER.writeValueAsBytes(value);
}
+ /**
+ * The request body for a caller-supplied payload (an Actor or task input): a {@code byte[]} and a
+ * {@link String} are already serialized and sent as they are (a string is UTF-8 encoded), so an
+ * input read from a file, or the exact bytes of a non-JSON content type, are not wrapped in a
+ * second layer of JSON. Anything else is serialized to JSON. A {@code null} payload sends no
+ * body.
+ */
+ public static byte[] toRequestBody(Object payload) {
+ if (payload == null) {
+ return null;
+ }
+ if (payload instanceof byte[] bytes) {
+ return bytes;
+ }
+ if (payload instanceof String text) {
+ return text.getBytes(java.nio.charset.StandardCharsets.UTF_8);
+ }
+ return toBytes(payload);
+ }
+
+ /** Converts a value to its JSON tree, as {@link #toBytes} would serialize it. */
+ public static JsonNode toTree(Object value) {
+ return MAPPER.valueToTree(value);
+ }
+
/** Parses JSON bytes into the given type. */
public static T parse(byte[] body, JavaType type) {
return MAPPER.readValue(body, type);
diff --git a/src/main/java/com/apify/client/internal/ListPublisher.java b/src/main/java/com/apify/client/internal/ListPublisher.java
index 05f38a3..7aa6188 100644
--- a/src/main/java/com/apify/client/internal/ListPublisher.java
+++ b/src/main/java/com/apify/client/internal/ListPublisher.java
@@ -79,6 +79,7 @@ public void request(long n) {
@Override
public void cancel() {
cancelled.set(true);
+ source.cancel(true);
}
private void drain() {
diff --git a/src/main/java/com/apify/client/internal/NormalizedUrlDeserializer.java b/src/main/java/com/apify/client/internal/NormalizedUrlDeserializer.java
new file mode 100644
index 0000000..a2659a6
--- /dev/null
+++ b/src/main/java/com/apify/client/internal/NormalizedUrlDeserializer.java
@@ -0,0 +1,22 @@
+package com.apify.client.internal;
+
+import tools.jackson.core.JsonParser;
+import tools.jackson.databind.DeserializationContext;
+import tools.jackson.databind.deser.std.StdDeserializer;
+
+/**
+ * Deserializes a {@code format: uri} string field of an API response into its {@linkplain
+ * UrlNormalizer#normalize normalized} form. Internal to the client; models reference it with
+ * {@code @JsonDeserialize}.
+ */
+public final class NormalizedUrlDeserializer extends StdDeserializer {
+
+ public NormalizedUrlDeserializer() {
+ super(String.class);
+ }
+
+ @Override
+ public String deserialize(JsonParser p, DeserializationContext ctxt) {
+ return UrlNormalizer.normalize(p.getValueAsString());
+ }
+}
diff --git a/src/main/java/com/apify/client/internal/QueryParams.java b/src/main/java/com/apify/client/internal/QueryParams.java
index a0ca627..eca5c53 100644
--- a/src/main/java/com/apify/client/internal/QueryParams.java
+++ b/src/main/java/com/apify/client/internal/QueryParams.java
@@ -34,6 +34,12 @@ public QueryParams addLong(String key, Long value) {
return this;
}
+ /** Sets an integer parameter, replacing any earlier value; a {@code null} value removes it. */
+ public QueryParams setLong(String key, Long value) {
+ pairs.removeIf(pair -> pair.key().equals(key));
+ return addLong(key, value);
+ }
+
/** Adds a floating-point parameter if the value is non-null. */
public QueryParams addDouble(String key, Double value) {
if (value != null) {
@@ -64,6 +70,16 @@ public QueryParams addRaw(String key, String value) {
return this;
}
+ /** The value of the first {@code key} parameter parsed as an integer, or {@code null}. */
+ public Long getLong(String key) {
+ for (Pair p : pairs) {
+ if (p.key().equals(key)) {
+ return Long.valueOf(p.value());
+ }
+ }
+ return null;
+ }
+
boolean isEmpty() {
return pairs.isEmpty();
}
diff --git a/src/main/java/com/apify/client/internal/ResourceContext.java b/src/main/java/com/apify/client/internal/ResourceContext.java
index c66c86e..d1014be 100644
--- a/src/main/java/com/apify/client/internal/ResourceContext.java
+++ b/src/main/java/com/apify/client/internal/ResourceContext.java
@@ -4,6 +4,7 @@
import com.apify.client.http.ApiResponse;
import com.apify.client.http.ApifyApiException;
import com.apify.client.http.ApifyTransportException;
+import com.apify.client.http.Timeout;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
@@ -35,13 +36,6 @@ public final class ResourceContext {
/** Server-side waitForFinish chunk size (the API caps server waiting at 60 seconds). */
private static final long WAIT_REQUEST_SECS = 60;
- /**
- * Safety margin subtracted from the configured per-request timeout when choosing the server-side
- * {@code waitForFinish} value, so the server responds before the client's socket timeout fires
- * (otherwise a short client timeout would abort every healthy poll and burn all retries).
- */
- private static final long WAIT_TIMEOUT_MARGIN_SECS = 5;
-
/**
* Finite upper bound used when the caller asks to wait "indefinitely" ({@code waitSecs == null}).
* The API will not accept "Infinity", and an unbounded loop can spin forever on a transient 404;
@@ -140,7 +134,7 @@ public static ResourceContext nestedCollection(
public static ResourceContext single(
HttpClientCore http, String baseUrl, String resourcePath, String id) {
return new ResourceContext(
- http, baseUrl + "/" + resourcePath + "/" + toSafeId(id), baseUrl, true);
+ http, baseUrl + "/" + resourcePath + "/" + encodePathSegment(toSafeId(id)), baseUrl, true);
}
/** A copy of this context with the origin used to build public URLs overridden. */
@@ -190,12 +184,17 @@ public ResourceContext seedParams(QueryParams inherited) {
public CompletableFuture> getResource(
String subPath, QueryParams params, JavaType dataType) {
+ return getResource(subPath, params, dataType, Timeout.SHORT);
+ }
+
+ public CompletableFuture> getResource(
+ String subPath, QueryParams params, JavaType dataType, Timeout timeout) {
// ofNullable, not of: an HTTP 200 with body {"data": null} unwraps to null, which is a valid
// "no resource" answer rather than a programming error — never surface it as a raw NPE.
// Explicit witness: getResourceRequired is the *receiver* of .handle below (not an argument
// to another generic call), so unlike a direct `return`, its type parameter cannot be inferred
// from the surrounding target type and must be pinned here.
- return this.getResourceRequired(subPath, params, dataType)
+ return this.getResourceRequired(subPath, params, dataType, timeout)
.handle(
(value, error) -> {
if (error == null) {
@@ -214,10 +213,20 @@ public CompletableFuture> getResource(
return getResource(subPath, params, Json.type(dataClass));
}
+ public CompletableFuture> getResource(
+ String subPath, QueryParams params, Class dataClass, Timeout timeout) {
+ return getResource(subPath, params, Json.type(dataClass), timeout);
+ }
+
public CompletableFuture getResourceRequired(
String subPath, QueryParams params, JavaType dataType) {
+ return getResourceRequired(subPath, params, dataType, Timeout.SHORT);
+ }
+
+ public CompletableFuture getResourceRequired(
+ String subPath, QueryParams params, JavaType dataType, Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("GET", u, null, "", http.baseRequestTimeout())
+ return http.call("GET", u, null, "", timeout)
.thenApply(resp -> Json.parseData(resp.body(), dataType));
}
@@ -226,6 +235,11 @@ public CompletableFuture getResourceRequired(
return getResourceRequired(subPath, params, Json.type(dataClass));
}
+ public CompletableFuture getResourceRequired(
+ String subPath, QueryParams params, Class dataClass, Timeout timeout) {
+ return getResourceRequired(subPath, params, Json.type(dataClass), timeout);
+ }
+
/**
* Like {@link #getResource}, but a 404 is only swallowed to {@link Optional#empty()} when this
* context {@link #ownId addresses its resource by an explicit id}. Otherwise (a nested path with
@@ -236,10 +250,16 @@ public CompletableFuture getResourceRequired(
*/
public CompletableFuture> getResourceUnlessAmbiguous(
String subPath, QueryParams params, JavaType dataType) {
+ return getResourceUnlessAmbiguous(subPath, params, dataType, Timeout.SHORT);
+ }
+
+ public CompletableFuture> getResourceUnlessAmbiguous(
+ String subPath, QueryParams params, JavaType dataType, Timeout timeout) {
if (ownId) {
- return getResource(subPath, params, dataType);
+ return getResource(subPath, params, dataType, timeout);
}
- return this.getResourceRequired(subPath, params, dataType).thenApply(Optional::ofNullable);
+ return this.getResourceRequired(subPath, params, dataType, timeout)
+ .thenApply(Optional::ofNullable);
}
public CompletableFuture> getResourceUnlessAmbiguous(
@@ -248,15 +268,20 @@ public CompletableFuture> getResourceUnlessAmbiguous(
}
public CompletableFuture updateResource(String subPath, Object body, Class dataClass) {
+ return updateResource(subPath, body, dataClass, Timeout.SHORT);
+ }
+
+ public CompletableFuture updateResource(
+ String subPath, Object body, Class dataClass, Timeout timeout) {
String u = mergedParams(new QueryParams()).applyToUrl(subUrl(subPath));
- return http.call("PUT", u, Json.toBytes(body), CONTENT_TYPE_JSON, http.baseRequestTimeout())
+ return http.call("PUT", u, Json.toBytes(body), CONTENT_TYPE_JSON, timeout)
.thenApply(resp -> Json.parseData(resp.body(), dataClass));
}
/** Performs a DELETE; a not-found is treated as a successful no-op. */
public CompletableFuture deleteResource(String subPath) {
String u = mergedParams(new QueryParams()).applyToUrl(subUrl(subPath));
- return http.call("DELETE", u, null, "", http.baseRequestTimeout())
+ return http.call("DELETE", u, null, "", Timeout.SHORT)
.handle(
(resp, error) -> {
if (error == null) {
@@ -280,13 +305,18 @@ public CompletableFuture deleteResourceUnlessAmbiguous(String subPath) {
return deleteResource(subPath);
}
String u = mergedParams(new QueryParams()).applyToUrl(subUrl(subPath));
- return http.call("DELETE", u, null, "", http.baseRequestTimeout()).thenApply(resp -> null);
+ return http.call("DELETE", u, null, "", Timeout.SHORT).thenApply(resp -> null);
}
public CompletableFuture> listResource(
String subPath, QueryParams params, Class itemClass) {
+ return listResource(subPath, params, itemClass, Timeout.MEDIUM);
+ }
+
+ public CompletableFuture> listResource(
+ String subPath, QueryParams params, Class itemClass, Timeout timeout) {
JavaType listType = Json.parametric(PaginationList.class, Json.type(itemClass));
- return getResourceRequired(subPath, params, listType);
+ return getResourceRequired(subPath, params, listType, timeout);
}
/**
@@ -320,8 +350,13 @@ public Flow.Publisher iterateResource(
public CompletableFuture createResource(
QueryParams params, Object body, Class dataClass) {
+ return createResource(params, body, dataClass, Timeout.SHORT);
+ }
+
+ public CompletableFuture createResource(
+ QueryParams params, Object body, Class dataClass, Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(""));
- return http.call("POST", u, Json.toBytes(body), CONTENT_TYPE_JSON, http.baseRequestTimeout())
+ return http.call("POST", u, Json.toBytes(body), CONTENT_TYPE_JSON, timeout)
.thenApply(resp -> Json.parseData(resp.body(), dataClass));
}
@@ -354,21 +389,30 @@ public CompletableFuture getOrCreateNamed(String name, Object body, Class
}
String u = params.applyToUrl(subUrl(""));
byte[] bodyBytes = body == null ? null : Json.toBytes(body);
- return http.call(
- "POST", u, bodyBytes, body == null ? "" : CONTENT_TYPE_JSON, http.baseRequestTimeout())
+ return http.call("POST", u, bodyBytes, body == null ? "" : CONTENT_TYPE_JSON, Timeout.SHORT)
.thenApply(resp -> Json.parseData(resp.body(), dataClass));
}
/** POST with a raw body (optional) and content type, unwrapping the data envelope. */
public CompletableFuture postWithBody(
- String subPath, QueryParams params, byte[] body, String contentType, Class dataClass) {
- return postWithBody(subPath, params, body, contentType, Json.type(dataClass));
+ String subPath,
+ QueryParams params,
+ byte[] body,
+ String contentType,
+ Class dataClass,
+ Timeout timeout) {
+ return postWithBody(subPath, params, body, contentType, Json.type(dataClass), timeout);
}
public CompletableFuture postWithBody(
- String subPath, QueryParams params, byte[] body, String contentType, JavaType dataType) {
+ String subPath,
+ QueryParams params,
+ byte[] body,
+ String contentType,
+ JavaType dataType,
+ Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("POST", u, body, contentType, http.baseRequestTimeout())
+ return http.call("POST", u, body, contentType, timeout)
.thenApply(resp -> Json.parseData(resp.body(), dataType));
}
@@ -378,17 +422,27 @@ public CompletableFuture postWithBody(
* plain object rather than the standard data envelope.
*/
public CompletableFuture postWithBodyNoEnvelope(
- String subPath, QueryParams params, byte[] body, String contentType, Class dataClass) {
+ String subPath,
+ QueryParams params,
+ byte[] body,
+ String contentType,
+ Class dataClass,
+ Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("POST", u, body, contentType, http.baseRequestTimeout())
+ return http.call("POST", u, body, contentType, timeout)
.thenApply(resp -> Json.parse(resp.body(), dataClass));
}
/** PUT with a raw body (optional) and content type, unwrapping the data envelope. */
public CompletableFuture putWithBody(
- String subPath, QueryParams params, byte[] body, String contentType, Class dataClass) {
+ String subPath,
+ QueryParams params,
+ byte[] body,
+ String contentType,
+ Class dataClass,
+ Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("PUT", u, body, contentType, http.baseRequestTimeout())
+ return http.call("PUT", u, body, contentType, timeout)
.thenApply(resp -> Json.parseData(resp.body(), dataClass));
}
@@ -398,26 +452,32 @@ public CompletableFuture putWithBody(
* response is a plain object rather than the standard data envelope.
*/
public CompletableFuture putWithBodyNoEnvelope(
- String subPath, QueryParams params, byte[] body, String contentType, Class dataClass) {
+ String subPath,
+ QueryParams params,
+ byte[] body,
+ String contentType,
+ Class dataClass,
+ Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("PUT", u, body, contentType, http.baseRequestTimeout())
+ return http.call("PUT", u, body, contentType, timeout)
.thenApply(resp -> Json.parse(resp.body(), dataClass));
}
/** DELETE with a JSON body (used for batch request deletion), unwrapping the data envelope. */
public CompletableFuture deleteWithBody(
- String subPath, QueryParams params, Object body, Class dataClass) {
+ String subPath, QueryParams params, Object body, Class dataClass, Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("DELETE", u, Json.toBytes(body), CONTENT_TYPE_JSON, http.baseRequestTimeout())
+ return http.call("DELETE", u, Json.toBytes(body), CONTENT_TYPE_JSON, timeout)
.thenApply(resp -> Json.parseData(resp.body(), dataClass));
}
/**
* GET returning the raw response (no data envelope). Completes with {@code null} on not-found.
*/
- public CompletableFuture getRaw(String subPath, QueryParams params) {
+ public CompletableFuture getRaw(
+ String subPath, QueryParams params, Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("GET", u, null, "", http.baseRequestTimeout())
+ return http.call("GET", u, null, "", timeout)
.handle(
(resp, error) -> {
if (error == null) {
@@ -434,23 +494,26 @@ public CompletableFuture getRaw(String subPath, QueryParams params)
/**
* As {@link #getRaw}, but a 404 is not swallowed - it propagates as an {@link ApifyApiException}.
*/
- public CompletableFuture getRawRequired(String subPath, QueryParams params) {
+ public CompletableFuture getRawRequired(
+ String subPath, QueryParams params, Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("GET", u, null, "", http.baseRequestTimeout());
+ return http.call("GET", u, null, "", timeout);
}
/**
* Like {@link #getRaw}, but - mirroring {@link #getResourceUnlessAmbiguous} - a 404 is only
* swallowed to {@code null} when this context addresses its resource by an explicit id.
*/
- public CompletableFuture getRawUnlessAmbiguous(String subPath, QueryParams params) {
- return ownId ? getRaw(subPath, params) : getRawRequired(subPath, params);
+ public CompletableFuture getRawUnlessAmbiguous(
+ String subPath, QueryParams params, Timeout timeout) {
+ return ownId ? getRaw(subPath, params, timeout) : getRawRequired(subPath, params, timeout);
}
/** HEAD request; completes with whether the resource exists. */
- public CompletableFuture headExists(String subPath, QueryParams params) {
+ public CompletableFuture headExists(
+ String subPath, QueryParams params, Timeout timeout) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
- return http.call("HEAD", u, null, "", http.baseRequestTimeout())
+ return http.call("HEAD", u, null, "", timeout)
.handle(
(resp, error) -> {
if (error == null) {
@@ -467,7 +530,7 @@ public CompletableFuture headExists(String subPath, QueryParams params)
/** PUT with raw bytes and a content type (used for key-value-store record uploads). */
public CompletableFuture putRaw(
String subPath, QueryParams params, byte[] body, String contentType) {
- return putRaw(subPath, params, body, contentType, http.baseRequestTimeout(), false);
+ return putRaw(subPath, params, body, contentType, Timeout.SHORT, false);
}
/**
@@ -481,16 +544,16 @@ public CompletableFuture putRaw(
QueryParams params,
byte[] body,
String contentType,
- Duration timeout,
+ Timeout timeout,
boolean doNotRetryTimeouts) {
return putRaw(subPath, params, body, contentType, null, timeout, doNotRetryTimeouts);
}
/**
- * As {@link #putRaw(String, QueryParams, byte[], String, Duration, boolean)}, additionally
- * setting {@code extraHeaders} on the request (used to forward a caller-supplied {@code
- * Content-Encoding} on a key-value-store record upload - see {@link HttpHeaders#CONTENT_ENCODING}
- * - which also stops {@link HttpClientCore} from compressing the already-encoded body again).
+ * As {@link #putRaw(String, QueryParams, byte[], String, Timeout, boolean)}, additionally setting
+ * {@code extraHeaders} on the request (used to forward a caller-supplied {@code Content-Encoding}
+ * on a key-value-store record upload - see {@link HttpHeaders#CONTENT_ENCODING} - which also
+ * stops {@link HttpClientCore} from compressing the already-encoded body again).
*/
public CompletableFuture putRaw(
String subPath,
@@ -498,38 +561,13 @@ public CompletableFuture putRaw(
byte[] body,
String contentType,
Map extraHeaders,
- Duration timeout,
+ Timeout timeout,
boolean doNotRetryTimeouts) {
String u = mergedParams(params).applyToUrl(subUrl(subPath));
return http.call("PUT", u, body, contentType, extraHeaders, timeout, doNotRetryTimeouts)
.thenApply(resp -> null);
}
- /**
- * The largest server-side {@code waitForFinish} value that is safe to send: below the configured
- * per-request timeout by a safety margin (or the API's 60s cap when no finite timeout is set), so
- * the server always responds before the client's socket timeout fires.
- */
- private long serverWaitCapSecs() {
- long configuredTimeoutSecs = http.requestTimeoutSeconds();
- return configuredTimeoutSecs > 0
- ? Math.max(0, configuredTimeoutSecs - WAIT_TIMEOUT_MARGIN_SECS)
- : WAIT_REQUEST_SECS;
- }
-
- /**
- * Clamps a caller-supplied server-side {@code waitForFinish} value (seconds) to {@link
- * #serverWaitCapSecs()}, so a synchronous get/wait can never ask the server to hold the
- * connection longer than the client's own per-request timeout. Returns {@code null} for a {@code
- * null} input (no server-side wait requested).
- */
- public Long clampServerWait(Long waitForFinishSecs) {
- if (waitForFinishSecs == null) {
- return null;
- }
- return Math.min(Math.max(0, waitForFinishSecs), serverWaitCapSecs());
- }
-
/**
* Polls a GET endpoint with {@code waitForFinish} until the resource reaches a terminal state or
* the wait budget elapses. {@code waitSecs == null} means "wait indefinitely", implemented as a
@@ -557,9 +595,13 @@ public CompletableFuture waitForFinish(
: MAX_WAIT_FOR_FINISH_SECS;
long budgetMillis = effectiveWaitSecs * 1000L;
long start = System.currentTimeMillis();
- long serverWaitCap = serverWaitCapSecs();
- return pollOnce(
- dataType, isTerminal, resourceName, start, budgetMillis, serverWaitCap, null, null);
+ // The polls have no timeout, and a withTimeout override does not apply to them: it would cut
+ // every hold short (a 5 s override would poll four times a second), and cancelling the returned
+ // future is the way out of a poll that never answers.
+ ResourceContext polling =
+ new ResourceContext(
+ http.withoutTimeoutOverride(), url, baseParams, apiOrigin, publicOrigin, ownId);
+ return polling.pollOnce(dataType, isTerminal, resourceName, start, budgetMillis, null, null);
}
/**
@@ -594,18 +636,16 @@ private CompletableFuture pollOnce(
String resourceName,
long start,
long budgetMillis,
- long serverWaitCap,
T lastSeen,
Long absentSinceMillis) {
long elapsed = System.currentTimeMillis() - start;
long remainingSecs = (budgetMillis - elapsed) / 1000L;
- long requestSecs =
- Math.min(Math.min(Math.max(remainingSecs, 0), WAIT_REQUEST_SECS), serverWaitCap);
+ long requestSecs = Math.min(Math.max(remainingSecs, 0), WAIT_REQUEST_SECS);
QueryParams params = new QueryParams();
params.addLong("waitForFinish", requestSecs);
- return this.getResourceRequired("", params, dataType)
+ return this.getResourceRequired("", params, dataType, Timeout.NO_TIMEOUT)
.handle(
(value, error) -> {
if (error == null) {
@@ -653,7 +693,6 @@ private CompletableFuture pollOnce(
resourceName,
start,
budgetMillis,
- serverWaitCap,
seen,
absentSince));
});
@@ -715,7 +754,15 @@ public static String encodePathSegment(String input) {
throw new IllegalArgumentException(
"a URL path segment must not be empty or a dot segment (\".\" or \"..\"), was: " + input);
}
- return URLEncoder.encode(input, StandardCharsets.UTF_8).replace("+", "%20");
+ // URLEncoder targets form encoding: it writes a space as "+" and escapes "~" and the
+ // characters of encodeURIComponent's unreserved set, none of which a path segment needs.
+ return URLEncoder.encode(input, StandardCharsets.UTF_8)
+ .replace("+", "%20")
+ .replace("%7E", "~")
+ .replace("%21", "!")
+ .replace("%27", "'")
+ .replace("%28", "(")
+ .replace("%29", ")");
}
/** Extracts the origin ({@code scheme://host[:port]}) from a URL, dropping any path. */
diff --git a/src/main/java/com/apify/client/internal/RunStartSupport.java b/src/main/java/com/apify/client/internal/RunStartSupport.java
index c02ce02..8f526e8 100644
--- a/src/main/java/com/apify/client/internal/RunStartSupport.java
+++ b/src/main/java/com/apify/client/internal/RunStartSupport.java
@@ -3,6 +3,7 @@
import com.apify.client.ApifyClient;
import com.apify.client.actor.ActorStartOptions;
import com.apify.client.http.ApifyApiException;
+import com.apify.client.http.Timeout;
import com.apify.client.log.StreamedLog;
import com.apify.client.log.StreamedLogOptions;
import com.apify.client.run.ActorRun;
@@ -10,6 +11,8 @@
import java.time.Duration;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -57,10 +60,10 @@ private RunStartSupport() {}
/**
* Starts a run on {@code ctx}'s {@code runs} sub-resource and completes with the created run as
- * soon as it exists (no waiting). {@code input} is any JSON-serializable value, or {@code null}
- * for no input. {@code applyParams} applies every start-option query parameter (build, memory,
- * timeout, ...) the caller's options type configures; {@code contentType} is the input body's
- * content type.
+ * soon as it exists (no waiting). {@code input} is any JSON-serializable value, a {@code byte[]}
+ * or {@link String} sent as it is, or {@code null} for no input. {@code applyParams} applies
+ * every start-option query parameter (build, memory, timeout, ...) the caller's options type
+ * configures; {@code contentType} is the input body's content type.
*
* @param waitForResourcesSecs mirrors {@link ActorStartOptions#waitForResources(Long)}: {@code
* null} does not retry a start rejected for lack of resources (the default); {@link
@@ -73,18 +76,41 @@ public static CompletableFuture start(
Consumer applyParams,
String contentType,
Long waitForResourcesSecs) {
+ return start(ctx, input, applyParams, contentType, waitForResourcesSecs, null);
+ }
+
+ /**
+ * As {@link #start(ResourceContext, Object, Consumer, String, Long)}, with an explicit request
+ * timeout; {@code null} picks the {@link Timeout#MEDIUM} tier extended by the start's {@code
+ * waitForFinish} hold, so the client does not abort a request the API is still holding.
+ */
+ private static CompletableFuture start(
+ ResourceContext ctx,
+ Object input,
+ Consumer applyParams,
+ String contentType,
+ Long waitForResourcesSecs,
+ Timeout timeout) {
QueryParams params = new QueryParams();
applyParams.accept(params);
- byte[] body = input == null ? null : Json.toBytes(input);
+ byte[] body = Json.toRequestBody(input);
+ Timeout requestTimeout = timeout;
+ if (requestTimeout == null) {
+ HttpClientCore.WaitHold hold =
+ ctx.http.waitHold(Timeout.MEDIUM, params.getLong("waitForFinish"));
+ params.setLong("waitForFinish", hold.waitForFinishSecs());
+ requestTimeout = hold.timeout();
+ }
if (waitForResourcesSecs == null) {
- return ctx.postWithBody("runs", params, body, contentType, ActorRun.class);
+ return ctx.postWithBody("runs", params, body, contentType, ActorRun.class, requestTimeout);
}
long deadlineMillis =
waitForResourcesSecs == ActorStartOptions.WAIT_FOR_RESOURCES_INDEFINITELY
? Long.MAX_VALUE
: System.currentTimeMillis()
+ Math.min(Math.max(0, waitForResourcesSecs), MAX_WAIT_FOR_RESOURCES_SECS) * 1000L;
- return startRetryingForResources(ctx, params, body, contentType, deadlineMillis);
+ return startRetryingForResources(
+ ctx, params, body, contentType, requestTimeout, deadlineMillis);
}
/**
@@ -98,8 +124,9 @@ private static CompletableFuture startRetryingForResources(
QueryParams params,
byte[] body,
String contentType,
+ Timeout timeout,
long deadlineMillis) {
- return ctx.postWithBody("runs", params, body, contentType, ActorRun.class)
+ return ctx.postWithBody("runs", params, body, contentType, ActorRun.class, timeout)
.handle(
(run, error) -> {
if (error == null) {
@@ -125,7 +152,7 @@ private static CompletableFuture startRetryingForResources(
.thenCompose(
unused ->
startRetryingForResources(
- ctx, params, body, contentType, deadlineMillis));
+ ctx, params, body, contentType, timeout, deadlineMillis));
})
.thenCompose(future -> future);
}
@@ -144,7 +171,7 @@ public static CompletableFuture