diff --git a/CHANGELOG.md b/CHANGELOG.md index eebdecd..903a180 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,22 @@ # Changelog +## 0.3.0 + +Breaking: `RequestQueueClient` methods that previously returned a raw `JsonObject`/took an untyped +`object` now use typed models, matching the OpenAPI-documented response schemas and the typing the +sibling clients already apply to this same resource: + +- `ListAndLockHeadAsync` now returns `LockedRequestQueueHead` (was `JsonObject`). +- `ProlongRequestLockAsync` now returns `RequestLockInfo` (was `JsonObject`). +- `UnlockRequestsAsync` now returns `UnlockRequestsResult` (was `JsonObject`). +- `ListRequestsAsync` now returns `RequestQueueRequestsPage` (was `JsonObject`). +- `BatchDeleteRequestsAsync` now takes `IReadOnlyList` and returns + `BatchDeleteResult` (was `object requests` / `JsonObject`). +- `RequestQueueRequest` gained `RetryCount`/`LockExpiresAt` properties, populated on requests returned + by `ListAndLockHeadAsync`/`ListRequestsAsync`. +- `RequestQueueHead` and `LockedRequestQueueHead` gained the previously-missing `QueueModifiedAt` + field (present in the OpenAPI spec and the reference client, but not yet exposed by this client). + ## 0.2.0 - Bumped `ApifyClientVersion.ApiSpecVersion` to the Apify OpenAPI spec `v2-2026-08-05T133145Z` and the diff --git a/docs/models.md b/docs/models.md index 4615812..4c757b7 100644 --- a/docs/models.md +++ b/docs/models.md @@ -203,6 +203,8 @@ A read/write model (request input and response). Fields set to `null` are omitte | `UniqueKey` | `string?` | The key used to deduplicate the request within the queue. | | `Method` | `string?` | HTTP method (defaults to `GET` on the server). | | `UserData` | `JsonNode?` | Arbitrary user-defined JSON payload attached to the request. | +| `RetryCount` | `long?` | Number of times the request has been retried (assigned by the queue). | +| `LockExpiresAt` | `string?` | ISO 8601 lock expiry; only set on requests returned by `ListAndLockHeadAsync`. | ## `RequestQueueHead` @@ -212,8 +214,61 @@ The head (front) of a request queue. |---|---|---| | `Items` | `IReadOnlyList` | The requests at the head of the queue. | | `Limit` | `long` | The page-size limit that was applied. | +| `QueueModifiedAt` | `string?` | ISO 8601 timestamp of the last modification to the queue. | | `HadMultipleClients` | `bool` | `true` if more than one client has accessed the queue (concurrency hint). | +## `LockedRequestQueueHead` + +The result of `RequestQueueClient.ListAndLockHeadAsync()`: a batch of requests locked for exclusive +processing. Each item's `RequestQueueRequest.LockExpiresAt` holds its individual lock expiry. + +| Property | Type | Description | +|---|---|---| +| `Items` | `IReadOnlyList` | The locked requests. | +| `Limit` | `long` | The maximum number of requests requested. | +| `QueueModifiedAt` | `string?` | ISO 8601 timestamp of the last modification to the queue. | +| `HadMultipleClients` | `bool` | `true` if more than one client has accessed the queue. | +| `LockSecs` | `long` | The lock duration applied to every returned request, in seconds. | +| `QueueHasLockedRequests` | `bool?` | Whether the queue has any requests locked by any client. | +| `ClientKey` | `string?` | The client key used to acquire the locks. | + +## `RequestLockInfo` + +The result of `RequestQueueClient.ProlongRequestLockAsync()`. + +| Property | Type | Description | +|---|---|---| +| `LockExpiresAt` | `string?` | ISO 8601 timestamp the (possibly just-extended) lock expires at. | + +## `UnlockRequestsResult` + +The result of `RequestQueueClient.UnlockRequestsAsync()`. + +| Property | Type | Description | +|---|---|---| +| `UnlockedCount` | `long` | Number of requests that were unlocked. | + +## `RequestQueueRequestsPage` + +One cursor-paginated page of `RequestQueueClient.ListRequestsAsync()`. + +| Property | Type | Description | +|---|---|---| +| `Items` | `IReadOnlyList` | The requests in this page. | +| `Limit` | `long` | The page-size limit that was applied. | +| `ExclusiveStartId` | `string?` | Deprecated by the API in favor of `Cursor`/`NextCursor`. | +| `Cursor` | `string?` | The cursor that produced this page. | +| `NextCursor` | `string?` | Cursor to pass to fetch the next page, or `null` if this is the last page. | + +## `BatchDeleteResult` + +The aggregate result of `RequestQueueClient.BatchDeleteRequestsAsync()`. + +| Property | Type | Description | +|---|---|---| +| `ProcessedRequests` | `IReadOnlyList` | Requests successfully deleted. | +| `UnprocessedRequests` | `IReadOnlyList` | Requests that failed to delete and can be retried. | + ## `RequestQueueOperationInfo` The result of adding/updating a single request. diff --git a/docs/storages.md b/docs/storages.md index d45965c..4210d35 100644 --- a/docs/storages.md +++ b/docs/storages.md @@ -140,16 +140,18 @@ options set a stable `ClientKey` (required to manage locks the client created) a `UpdateRequestAsync(RequestQueueRequest request, bool forefront = false)` → `RequestQueueOperationInfo`; `DeleteRequestAsync(string id)` (no return value). - `ListHeadAsync(int? limit = null)` → `RequestQueueHead`; - `ListAndLockHeadAsync(int lockSecs, int? limit = null)`. + `ListAndLockHeadAsync(int lockSecs, int? limit = null)` → `LockedRequestQueueHead` (each item's + `RequestQueueRequest.LockExpiresAt`/`RetryCount` is populated). - `BatchAddRequestsAsync(IReadOnlyList requests, bool forefront = false, BatchAddRequestsOptions? options = null)` → `BatchAddResult` — auto-chunks by count (25) and payload size (~9 MiB) and retries unprocessed requests. Every request needs a non-empty `UniqueKey`. -- `BatchDeleteRequestsAsync(object requests)` → `JsonObject` — delete a batch of requests in one call - (`requests` is any JSON-serializable list of requests/keys to remove). -- `ListRequestsAsync(ListRequestsOptions? options = null)` → `JsonObject` and +- `BatchDeleteRequestsAsync(IReadOnlyList requests)` → `BatchDeleteResult` — delete a + batch of requests in one call; each entry identifies the request to delete via `Id` and/or `UniqueKey`. +- `ListRequestsAsync(ListRequestsOptions? options = null)` → `RequestQueueRequestsPage` and `PaginateRequestsAsync(PaginateRequestsOptions? options = null)` → `IAsyncEnumerable`. -- Lock management: `ProlongRequestLockAsync(string id, int lockSecs, bool forefront = false)` → `JsonObject`, - `DeleteRequestLockAsync(string id, bool forefront = false)`, `UnlockRequestsAsync()` → `JsonObject`. +- Lock management: `ProlongRequestLockAsync(string id, int lockSecs, bool forefront = false)` → + `RequestLockInfo`, `DeleteRequestLockAsync(string id, bool forefront = false)`, + `UnlockRequestsAsync()` → `UnlockRequestsResult`. - `WithClientKey(string clientKey)` → `RequestQueueClient` — returns a copy of this client bound to the given client key (a fluent alternative to passing `RequestQueueClientOptions.ClientKey` on `client.RequestQueue(id, options)`); the client key ties lock ownership to this client. diff --git a/src/Apify.Client/Apify.Client.csproj b/src/Apify.Client/Apify.Client.csproj index 482579e..979d0c2 100644 --- a/src/Apify.Client/Apify.Client.csproj +++ b/src/Apify.Client/Apify.Client.csproj @@ -6,7 +6,7 @@ Apify.Client - 0.2.0 + 0.3.0 Apify Apify Apify API client for .NET diff --git a/src/Apify.Client/ApifyClientVersion.cs b/src/Apify.Client/ApifyClientVersion.cs index dc895d4..6fff84f 100644 --- a/src/Apify.Client/ApifyClientVersion.cs +++ b/src/Apify.Client/ApifyClientVersion.cs @@ -14,7 +14,7 @@ public static class ApifyClientVersion /// The semantic version of this client library (see https://semver.org/). Changes to the public /// interface other than additive ones are considered breaking changes. /// - public const string ClientVersion = "0.2.0"; + public const string ClientVersion = "0.3.0"; /// /// The version of the Apify OpenAPI specification this client was generated and verified against. diff --git a/src/Apify.Client/Models/BatchDeleteResult.cs b/src/Apify.Client/Models/BatchDeleteResult.cs new file mode 100644 index 0000000..104774e --- /dev/null +++ b/src/Apify.Client/Models/BatchDeleteResult.cs @@ -0,0 +1,46 @@ +using System.Collections.Generic; +using System.Text.Json.Nodes; +using Apify.Client.Internal; + +namespace Apify.Client.Models; + +/// +/// The result of a batch request-delete: the requests that were successfully removed and the ones that +/// could not be (and can be retried). +/// +public sealed class BatchDeleteResult +{ + private BatchDeleteResult(IReadOnlyList processed, IReadOnlyList unprocessed) + { + ProcessedRequests = processed; + UnprocessedRequests = unprocessed; + } + + /// Builds a result from the decoded response object. + /// The decoded response object. + public static BatchDeleteResult FromData(JsonNode? data) + { + var obj = JsonValues.AsObject(data); + return new BatchDeleteResult(Hydrate(obj, "processedRequests"), Hydrate(obj, "unprocessedRequests")); + } + + private static List Hydrate(JsonObject obj, string key) + { + var items = new List(); + if (obj.TryGetPropertyValue(key, out var node) && node is JsonArray array) + { + foreach (var item in array) + { + items.Add(RequestQueueRequest.FromJsonObject(item as JsonObject ?? new JsonObject())); + } + } + + return items; + } + + /// The requests that were successfully deleted from the queue. + public IReadOnlyList ProcessedRequests { get; } + + /// The requests that failed to be deleted and can be retried. + public IReadOnlyList UnprocessedRequests { get; } +} diff --git a/src/Apify.Client/Models/LockedRequestQueueHead.cs b/src/Apify.Client/Models/LockedRequestQueueHead.cs new file mode 100644 index 0000000..f10db55 --- /dev/null +++ b/src/Apify.Client/Models/LockedRequestQueueHead.cs @@ -0,0 +1,69 @@ +using System.Collections.Generic; +using System.Text.Json.Nodes; +using Apify.Client.Internal; + +namespace Apify.Client.Models; + +/// A batch of requests from the head of a request queue, locked for exclusive processing. +public sealed class LockedRequestQueueHead +{ + private LockedRequestQueueHead( + IReadOnlyList items, + long limit, + string? queueModifiedAt, + bool hadMultipleClients, + long lockSecs, + bool? queueHasLockedRequests, + string? clientKey) + { + Items = items; + Limit = limit; + QueueModifiedAt = queueModifiedAt; + HadMultipleClients = hadMultipleClients; + LockSecs = lockSecs; + QueueHasLockedRequests = queueHasLockedRequests; + ClientKey = clientKey; + } + + /// Builds a locked head from the decoded response object. + /// The decoded response object. + public static LockedRequestQueueHead FromData(JsonNode? data) + { + var obj = JsonValues.AsObject(data); + var items = new List(); + foreach (var item in JsonValues.ObjectItems(obj)) + { + items.Add(RequestQueueRequest.FromJsonObject(item)); + } + + return new LockedRequestQueueHead( + items, + JsonValues.IntOr(obj, "limit", items.Count), + JsonValues.String(obj, "queueModifiedAt"), + JsonValues.BoolOr(obj, "hadMultipleClients", false), + JsonValues.IntOr(obj, "lockSecs", 0), + obj.ContainsKey("queueHasLockedRequests") ? JsonValues.BoolOr(obj, "queueHasLockedRequests", false) : null, + JsonValues.String(obj, "clientKey")); + } + + /// The locked requests from the head of the queue. Each carries its own LockExpiresAt. + public IReadOnlyList Items { get; } + + /// The maximum number of requests requested. + public long Limit { get; } + + /// ISO 8601 timestamp of the last modification to the queue. + public string? QueueModifiedAt { get; } + + /// Whether multiple clients have accessed the queue. + public bool HadMultipleClients { get; } + + /// The lock duration applied to every returned request, in seconds. + public long LockSecs { get; } + + /// Whether the queue has any requests locked by any client (this one or another). + public bool? QueueHasLockedRequests { get; } + + /// The client key used to acquire the locks. + public string? ClientKey { get; } +} diff --git a/src/Apify.Client/Models/RequestLockInfo.cs b/src/Apify.Client/Models/RequestLockInfo.cs new file mode 100644 index 0000000..0a8abec --- /dev/null +++ b/src/Apify.Client/Models/RequestLockInfo.cs @@ -0,0 +1,24 @@ +using System.Text.Json.Nodes; +using Apify.Client.Internal; + +namespace Apify.Client.Models; + +/// The result of prolonging a request lock: the new lock expiry. +public sealed class RequestLockInfo +{ + private RequestLockInfo(string? lockExpiresAt) + { + LockExpiresAt = lockExpiresAt; + } + + /// Builds a lock info from the decoded response object. + /// The decoded response object. + public static RequestLockInfo FromData(JsonNode? data) + { + var obj = JsonValues.AsObject(data); + return new RequestLockInfo(JsonValues.String(obj, "lockExpiresAt")); + } + + /// When the (possibly just-extended) lock expires (ISO-8601 string). + public string? LockExpiresAt { get; } +} diff --git a/src/Apify.Client/Models/RequestQueueHead.cs b/src/Apify.Client/Models/RequestQueueHead.cs index 6f3ea4d..90c1561 100644 --- a/src/Apify.Client/Models/RequestQueueHead.cs +++ b/src/Apify.Client/Models/RequestQueueHead.cs @@ -7,10 +7,11 @@ namespace Apify.Client.Models; /// The head (front) of a request queue. public sealed class RequestQueueHead { - private RequestQueueHead(IReadOnlyList items, long limit, bool hadMultipleClients) + private RequestQueueHead(IReadOnlyList items, long limit, string? queueModifiedAt, bool hadMultipleClients) { Items = items; Limit = limit; + QueueModifiedAt = queueModifiedAt; HadMultipleClients = hadMultipleClients; } @@ -28,6 +29,7 @@ public static RequestQueueHead FromData(JsonNode? data) return new RequestQueueHead( items, JsonValues.IntOr(obj, "limit", items.Count), + JsonValues.String(obj, "queueModifiedAt"), JsonValues.BoolOr(obj, "hadMultipleClients", false)); } @@ -37,6 +39,9 @@ public static RequestQueueHead FromData(JsonNode? data) /// The maximum number of requests requested. public long Limit { get; } + /// ISO 8601 timestamp of the last modification to the queue. + public string? QueueModifiedAt { get; } + /// Whether multiple clients have accessed the queue. public bool HadMultipleClients { get; } } diff --git a/src/Apify.Client/Models/RequestQueueRequest.cs b/src/Apify.Client/Models/RequestQueueRequest.cs index d4e12a3..065c53a 100644 --- a/src/Apify.Client/Models/RequestQueueRequest.cs +++ b/src/Apify.Client/Models/RequestQueueRequest.cs @@ -62,6 +62,36 @@ public string? Method set => SetString("method", value); } + /// + /// The number of times this request has been retried after a failed processing attempt (assigned by + /// the API; absent on create). + /// + public long? RetryCount + { + get => GetInt("retryCount"); + set + { + if (value is null) + { + ToJsonObject().Remove("retryCount"); + } + else + { + ToJsonObject()["retryCount"] = value.Value; + } + } + } + + /// + /// When this request's processing lock expires (ISO-8601 string). Only present on requests returned + /// by . + /// + public string? LockExpiresAt + { + get => GetString("lockExpiresAt"); + set => SetString("lockExpiresAt", value); + } + /// Arbitrary user-attached metadata. public JsonNode? UserData { diff --git a/src/Apify.Client/Models/RequestQueueRequestsPage.cs b/src/Apify.Client/Models/RequestQueueRequestsPage.cs new file mode 100644 index 0000000..b2f46f0 --- /dev/null +++ b/src/Apify.Client/Models/RequestQueueRequestsPage.cs @@ -0,0 +1,61 @@ +using System.Collections.Generic; +using System.Text.Json.Nodes; +using Apify.Client.Internal; + +namespace Apify.Client.Models; + +/// +/// A single, cursor-paginated page of a request queue's requests (as returned by +/// ). +/// +public sealed class RequestQueueRequestsPage +{ + private RequestQueueRequestsPage( + IReadOnlyList items, + long limit, + string? exclusiveStartId, + string? cursor, + string? nextCursor) + { + Items = items; + Limit = limit; + ExclusiveStartId = exclusiveStartId; + Cursor = cursor; + NextCursor = nextCursor; + } + + /// Builds a page from the decoded response object. + /// The decoded response object. + public static RequestQueueRequestsPage FromData(JsonNode? data) + { + var obj = JsonValues.AsObject(data); + var items = new List(); + foreach (var item in JsonValues.ObjectItems(obj)) + { + items.Add(RequestQueueRequest.FromJsonObject(item)); + } + + return new RequestQueueRequestsPage( + items, + JsonValues.IntOr(obj, "limit", items.Count), + JsonValues.String(obj, "exclusiveStartId"), + JsonValues.String(obj, "cursor"), + JsonValues.String(obj, "nextCursor")); + } + + /// The requests in this page. + public IReadOnlyList Items { get; } + + /// The maximum number of requests requested for this page. + public long Limit { get; } + + /// The ID of the last request of the previous page, if pagination was continued by ID. + /// Deprecated by the API in favor of /. + public string? ExclusiveStartId { get; } + + /// The cursor that produced this page, if pagination was continued by cursor. + public string? Cursor { get; } + + /// The cursor to pass to fetch the next page, or null if this is the last page. + public string? NextCursor { get; } +} diff --git a/src/Apify.Client/Models/UnlockRequestsResult.cs b/src/Apify.Client/Models/UnlockRequestsResult.cs new file mode 100644 index 0000000..07a0161 --- /dev/null +++ b/src/Apify.Client/Models/UnlockRequestsResult.cs @@ -0,0 +1,24 @@ +using System.Text.Json.Nodes; +using Apify.Client.Internal; + +namespace Apify.Client.Models; + +/// The result of releasing all of a client's request locks on a queue. +public sealed class UnlockRequestsResult +{ + private UnlockRequestsResult(long unlockedCount) + { + UnlockedCount = unlockedCount; + } + + /// Builds a result from the decoded response object. + /// The decoded response object. + public static UnlockRequestsResult FromData(JsonNode? data) + { + var obj = JsonValues.AsObject(data); + return new UnlockRequestsResult(JsonValues.IntOr(obj, "unlockedCount", 0)); + } + + /// The number of requests that were unlocked. + public long UnlockedCount { get; } +} diff --git a/src/Apify.Client/Resources/RequestQueueClient.cs b/src/Apify.Client/Resources/RequestQueueClient.cs index 6277d76..29266e8 100644 --- a/src/Apify.Client/Resources/RequestQueueClient.cs +++ b/src/Apify.Client/Resources/RequestQueueClient.cs @@ -154,17 +154,18 @@ public async Task DeleteRequestAsync(string id, CancellationToken cancellationTo /// /// Atomically returns and locks up to requests from the head of the queue for - /// seconds. Returns the raw locked-head object. + /// seconds. /// /// How long to lock the returned requests, in seconds. /// The maximum number of requests to lock. /// A token to cancel the request. - public Task ListAndLockHeadAsync(int lockSecs, int? limit = null, CancellationToken cancellationToken = default) + public async Task ListAndLockHeadAsync(int lockSecs, int? limit = null, CancellationToken cancellationToken = default) { var q = new QueryParams(); q.AddInt("lockSecs", lockSecs).AddInt("limit", limit); ApplyClientKey(q); - return _ctx.PostWithBodyAsync("head/lock", q, null, "", cancellationToken); + var data = await _ctx.PostWithBodyAsync("head/lock", q, null, "", cancellationToken).ConfigureAwait(false); + return LockedRequestQueueHead.FromData(data); } /// @@ -279,11 +280,7 @@ private async Task DispatchChunkAsync( /// private static List SliceByByteLength(List requests, int maxByteLength, int startIndex) { - var payloads = new List(requests.Count); - foreach (var r in requests) - { - payloads.Add(r.ToJsonObject()); - } + var payloads = ToPayload(requests); if (Encoding.UTF8.GetByteCount(Json.Encode(payloads)) < maxByteLength) { @@ -371,13 +368,7 @@ private async Task BatchAddChunkAsync(List var q = new QueryParams(); q.AddBool("forefront", forefront); ApplyClientKey(q); - var payload = new List(requests.Count); - foreach (var r in requests) - { - payload.Add(r.ToJsonObject()); - } - - var data = await _ctx.PostWithBodyAsync("requests/batch", q, Json.Encode(payload), ResourceContext.ContentTypeJson, cancellationToken).ConfigureAwait(false); + var data = await _ctx.PostWithBodyAsync("requests/batch", q, Json.Encode(ToPayload(requests)), ResourceContext.ContentTypeJson, cancellationToken).ConfigureAwait(false); var processed = new List(); if (data.TryGetPropertyValue("processedRequests", out var pNode) && pNode is JsonArray pArray) @@ -437,22 +428,36 @@ private static Task SleepBackoffAsync(int attempt, int minDelayMillis, Cancellat } /// - /// Deletes multiple requests in a single call. Each entry identifies a request (e.g. by id or - /// uniqueKey). Returns the raw batch result. + /// Deletes multiple requests in a single call. Each entry must have its + /// and/or set to identify the request to delete; other fields + /// are ignored. /// - /// A JSON-serializable list identifying the requests to delete. + /// The requests to delete, identified by Id and/or UniqueKey. /// A token to cancel the request. - public Task BatchDeleteRequestsAsync(object requests, CancellationToken cancellationToken = default) + public async Task BatchDeleteRequestsAsync(IReadOnlyList requests, CancellationToken cancellationToken = default) { var q = new QueryParams(); ApplyClientKey(q); - return _ctx.DeleteWithBodyAsync("requests/batch", q, requests, cancellationToken); + var data = await _ctx.DeleteWithBodyAsync("requests/batch", q, ToPayload(requests), cancellationToken).ConfigureAwait(false); + return BatchDeleteResult.FromData(data); + } + + /// Encodes a list of requests as their raw JSON objects, for a batch request body. + private static List ToPayload(IReadOnlyCollection requests) + { + var payload = new List(requests.Count); + foreach (var r in requests) + { + payload.Add(r.ToJsonObject()); + } + + return payload; } - /// Lists the queue's requests with pagination. Returns the raw response. + /// Lists the queue's requests with pagination. /// Optional listing filters and pagination. /// A token to cancel the request. - public async Task ListRequestsAsync(ListRequestsOptions? options = null, CancellationToken cancellationToken = default) + public async Task ListRequestsAsync(ListRequestsOptions? options = null, CancellationToken cancellationToken = default) { options ??= new ListRequestsOptions(); options.Validate(); @@ -460,18 +465,18 @@ public async Task ListRequestsAsync(ListRequestsOptions? options = n options.AppendTo(q); ApplyClientKey(q); var data = await _ctx.GetResourceRequiredAsync("requests", q, cancellationToken).ConfigureAwait(false); - return data as JsonObject ?? new JsonObject(); + return RequestQueueRequestsPage.FromData(data); } /// /// Extends the lock on a request by seconds. If - /// is true, the request is moved to the front when its lock expires. Returns the raw response. + /// is true, the request is moved to the front when its lock expires. /// /// The request ID. /// How much longer to hold the lock, in seconds. /// Whether to move the request to the front when the lock expires. /// A token to cancel the request. - public async Task ProlongRequestLockAsync(string id, int lockSecs, bool forefront = false, CancellationToken cancellationToken = default) + public async Task ProlongRequestLockAsync(string id, int lockSecs, bool forefront = false, CancellationToken cancellationToken = default) { var q = new QueryParams(); q.AddInt("lockSecs", lockSecs).AddBool("forefront", forefront); @@ -479,7 +484,7 @@ public async Task ProlongRequestLockAsync(string id, int lockSecs, b var url = _ctx.MergedParams(q).ApplyToUrl(_ctx.SubUrl("requests/" + ResourceContext.EncodePathSegment(id) + "/lock")); using var response = await _http.CallAsync(HttpMethod.Put, url, null, "", _timeout, cancellationToken: cancellationToken).ConfigureAwait(false); var data = Json.DecodeData(await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false)); - return data as JsonObject ?? new JsonObject(); + return RequestLockInfo.FromData(data); } /// @@ -505,13 +510,14 @@ public async Task DeleteRequestLockAsync(string id, bool forefront = false, Canc } } - /// Releases all locks the client holds on this queue's requests. Returns the raw response. + /// Releases all locks the client holds on this queue's requests. /// A token to cancel the request. - public Task UnlockRequestsAsync(CancellationToken cancellationToken = default) + public async Task UnlockRequestsAsync(CancellationToken cancellationToken = default) { var q = new QueryParams(); ApplyClientKey(q); - return _ctx.PostWithBodyAsync("requests/unlock", q, null, "", cancellationToken); + var data = await _ctx.PostWithBodyAsync("requests/unlock", q, null, "", cancellationToken).ConfigureAwait(false); + return UnlockRequestsResult.FromData(data); } /// @@ -553,22 +559,19 @@ public async IAsyncEnumerable PaginateRequestsAsync( }, cancellationToken).ConfigureAwait(false); - var items = page.TryGetPropertyValue("items", out var itemsNode) && itemsNode is JsonArray array - ? array - : new JsonArray(); - if (items.Count == 0) + if (page.Items.Count == 0) { yield break; } - foreach (var item in items) + foreach (var item in page.Items) { - yield return RequestQueueRequest.FromJsonObject(item as JsonObject ?? new JsonObject()); + yield return item; } - iterated += items.Count; + iterated += page.Items.Count; - nextCursor = JsonValues.String(page, "nextCursor"); + nextCursor = page.NextCursor; if ((limit is not null && iterated >= limit.Value) || string.IsNullOrEmpty(nextCursor)) { yield break; diff --git a/tests/Apify.Client.Tests/Integration/RequestQueueIntegrationTests.cs b/tests/Apify.Client.Tests/Integration/RequestQueueIntegrationTests.cs index 3106671..9f53fcc 100644 --- a/tests/Apify.Client.Tests/Integration/RequestQueueIntegrationTests.cs +++ b/tests/Apify.Client.Tests/Integration/RequestQueueIntegrationTests.cs @@ -128,15 +128,27 @@ public async Task RequestQueueLockLifecycle() { var queue = client.RequestQueue(rq.Id!).WithClientKey("dotnet-test-client-key"); var info = await queue.AddRequestAsync(new RequestQueueRequest("https://lock.example.com", "lock")); - Assert.True((await queue.ListRequestsAsync(new ListRequestsOptions())).ContainsKey("items")); + Assert.NotEmpty((await queue.ListRequestsAsync(new ListRequestsOptions())).Items); await queue.ListRequestsAsync(new ListRequestsOptions { Filter = new[] { ListRequestsOptions.FilterLocked, ListRequestsOptions.FilterPending }, }); - Assert.True((await queue.ListAndLockHeadAsync(60, 10)).ContainsKey("items")); - await queue.ProlongRequestLockAsync(info.RequestId!, 30); + + var locked = await queue.ListAndLockHeadAsync(60, 10); + Assert.NotEmpty(locked.Items); + Assert.Equal(60, locked.LockSecs); + Assert.NotNull(locked.Items[0].LockExpiresAt); + + var prolonged = await queue.ProlongRequestLockAsync(info.RequestId!, 30); + Assert.NotNull(prolonged.LockExpiresAt); + await queue.DeleteRequestLockAsync(info.RequestId!); - await queue.UnlockRequestsAsync(); + + var unlocked = await queue.UnlockRequestsAsync(); + Assert.True(unlocked.UnlockedCount >= 0); + + var deleted = await queue.BatchDeleteRequestsAsync(new[] { new RequestQueueRequest { Id = info.RequestId } }); + Assert.Equal(info.RequestId, Assert.Single(deleted.ProcessedRequests).Id); } finally { diff --git a/tests/Apify.Client.Tests/Unit/RequestShapeTests.cs b/tests/Apify.Client.Tests/Unit/RequestShapeTests.cs index 2e8ef0e..384fd7c 100644 --- a/tests/Apify.Client.Tests/Unit/RequestShapeTests.cs +++ b/tests/Apify.Client.Tests/Unit/RequestShapeTests.cs @@ -2,6 +2,7 @@ using System.Globalization; using System.Threading.Tasks; using System.Text.Json.Nodes; +using Apify.Client.Models; using Apify.Client.Options; using Xunit; @@ -401,4 +402,110 @@ public async Task SetRecordSendsRawBytesWithVerbatimContentType() Assert.DoesNotContain("charset", request.Header("Content-Type"), StringComparison.OrdinalIgnoreCase); Assert.Equal(bytes, request.BodyBytes); } + + [Fact] + public async Task ListHeadDecodesRequestQueueHead() + { + var transport = new MockTransport().QueueResponse(200, """ + {"data":{"limit":3,"queueModifiedAt":"2018-03-14T23:00:00.000Z","hadMultipleClients":false, + "items":[{"id":"r1","uniqueKey":"k1","url":"https://example.com"}]}} + """); + var head = await Client(transport).RequestQueue("q1").ListHeadAsync(3); + + Assert.Equal(3, head.Limit); + Assert.Equal("2018-03-14T23:00:00.000Z", head.QueueModifiedAt); + Assert.False(head.HadMultipleClients); + Assert.Equal("r1", Assert.Single(head.Items).Id); + } + + [Fact] + public async Task ListAndLockHeadDecodesLockedRequestQueueHead() + { + var transport = new MockTransport().QueueResponse(200, """ + {"data":{"limit":2,"queueModifiedAt":"2018-03-14T23:00:00.000Z","hadMultipleClients":true, + "lockSecs":60,"queueHasLockedRequests":true,"clientKey":"client-one", + "items":[{"id":"r1","uniqueKey":"k1","url":"https://example.com","method":"GET", + "retryCount":0,"lockExpiresAt":"2022-06-14T23:00:00.000Z"}]}} + """); + var head = await Client(transport).RequestQueue("q1").ListAndLockHeadAsync(60, 2); + + var request = transport.LastRequest; + Assert.Equal("POST", request.Method); + Assert.Contains("/request-queues/q1/head/lock", request.Uri, StringComparison.Ordinal); + Assert.Contains("lockSecs=60", request.Uri, StringComparison.Ordinal); + Assert.Equal(2, head.Limit); + Assert.Equal("2018-03-14T23:00:00.000Z", head.QueueModifiedAt); + Assert.Equal(60, head.LockSecs); + Assert.True(head.HadMultipleClients); + Assert.True(head.QueueHasLockedRequests); + Assert.Equal("client-one", head.ClientKey); + var item = Assert.Single(head.Items); + Assert.Equal("r1", item.Id); + Assert.Equal("k1", item.UniqueKey); + Assert.Equal(0, item.RetryCount); + Assert.Equal("2022-06-14T23:00:00.000Z", item.LockExpiresAt); + } + + [Fact] + public async Task ProlongRequestLockDecodesLockExpiresAt() + { + var transport = new MockTransport().QueueResponse(200, "{\"data\":{\"lockExpiresAt\":\"2022-06-14T23:00:00.000Z\"}}"); + var info = await Client(transport).RequestQueue("q1").ProlongRequestLockAsync("r1", 30); + + var request = transport.LastRequest; + Assert.Equal("PUT", request.Method); + Assert.Contains("/request-queues/q1/requests/r1/lock", request.Uri, StringComparison.Ordinal); + Assert.Contains("lockSecs=30", request.Uri, StringComparison.Ordinal); + Assert.Equal("2022-06-14T23:00:00.000Z", info.LockExpiresAt); + } + + [Fact] + public async Task UnlockRequestsDecodesUnlockedCount() + { + var transport = new MockTransport().QueueResponse(200, "{\"data\":{\"unlockedCount\":3}}"); + var result = await Client(transport).RequestQueue("q1").UnlockRequestsAsync(); + + Assert.Equal("POST", transport.LastRequest.Method); + Assert.Contains("/request-queues/q1/requests/unlock", transport.LastRequest.Uri, StringComparison.Ordinal); + Assert.Equal(3, result.UnlockedCount); + } + + [Fact] + public async Task ListRequestsDecodesCursorPage() + { + var transport = new MockTransport().QueueResponse(200, """ + {"data":{"limit":2,"cursor":"c0","nextCursor":"c1", + "items":[{"id":"r1","uniqueKey":"k1","url":"https://example.com"}]}} + """); + var page = await Client(transport).RequestQueue("q1").ListRequestsAsync(); + + Assert.Equal(2, page.Limit); + Assert.Equal("c0", page.Cursor); + Assert.Equal("c1", page.NextCursor); + Assert.Equal("r1", Assert.Single(page.Items).Id); + } + + [Fact] + public async Task BatchDeleteRequestsSendsIdentifiersAndDecodesResult() + { + var transport = new MockTransport().QueueResponse(200, """ + {"data":{"processedRequests":[{"id":"r1","uniqueKey":"k1"}], + "unprocessedRequests":[{"uniqueKey":"k2","url":"https://example.com/2"}]}} + """); + var result = await Client(transport).RequestQueue("q1").BatchDeleteRequestsAsync(new[] + { + new RequestQueueRequest { Id = "r1" }, + new RequestQueueRequest { UniqueKey = "k2" }, + }); + + var request = transport.LastRequest; + Assert.Equal("DELETE", request.Method); + Assert.Contains("/request-queues/q1/requests/batch", request.Uri, StringComparison.Ordinal); + var sent = JsonNode.Parse(request.Body)!.AsArray(); + Assert.Equal("r1", sent[0]!["id"]!.GetValue()); + Assert.Equal("k2", sent[1]!["uniqueKey"]!.GetValue()); + + Assert.Equal("r1", Assert.Single(result.ProcessedRequests).Id); + Assert.Equal("k2", Assert.Single(result.UnprocessedRequests).UniqueKey); + } }