Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/CosmosDbStorage.cs
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ private CosmosDbStorage(string databaseName, string containerName, CosmosDbStora

private void ConfigureCosmosClientOptions(CosmosClientOptions cosmosClientOptions)
{
cosmosClientOptions.ConnectionMode = ConnectionMode.Direct;
cosmosClientOptions.ApplicationName ??= "Hangfire";
cosmosClientOptions.Serializer = new CosmosJsonSerializer(settings);
cosmosClientOptions.MaxRetryAttemptsOnRateLimitedRequests ??= 9;
Expand Down
57 changes: 33 additions & 24 deletions src/Queue/JobQueueMonitoringApi.cs
Original file line number Diff line number Diff line change
Expand Up @@ -45,40 +45,49 @@ public IEnumerable<string> GetQueues()

public int GetEnqueuedCount(string queue)
{
QueryDefinition sql = new QueryDefinition("SELECT TOP 1 VALUE COUNT(1) FROM doc WHERE doc.name = @name AND NOT IS_DEFINED(doc.fetched_at)")
QueryDefinition sql = new QueryDefinition("SELECT TOP 1 VALUE COUNT(1) FROM doc WHERE doc.name = @name AND (NOT IS_DEFINED(doc.fetched_at) OR IS_NULL(doc.fetched_at))")
.WithParameter("@name", queue);

return storage.Container.GetItemQueryIterator<int>(sql, requestOptions: new QueryRequestOptions { PartitionKey = partitionKey })
.ToQueryResult()
.FirstOrDefault();
}

public IEnumerable<string> GetEnqueuedJobIds(string queue, int from, int perPage) => storage.Container.GetItemLinqQueryable<Documents.Queue>(requestOptions: new QueryRequestOptions { PartitionKey = partitionKey })
.Where(q => q.Name == queue && q.FetchedAt.IsDefined() == false)
.OrderBy(q => q.CreatedOn)
.Skip(from).Take(perPage)
.Select(q => q.JobId)
.ToQueryResult()
.ToList();
public IEnumerable<string> GetEnqueuedJobIds(string queue, int from, int perPage)
{
QueryDefinition sql = new QueryDefinition("SELECT VALUE doc.job_id FROM doc WHERE doc.name = @name AND (NOT IS_DEFINED(doc.fetched_at) OR IS_NULL(doc.fetched_at)) ORDER BY doc.created_on OFFSET @offset LIMIT @limit")
.WithParameter("@name", queue)
.WithParameter("@offset", from)
.WithParameter("@limit", perPage);

return storage.Container.GetItemQueryIterator<string>(sql, requestOptions: new QueryRequestOptions { PartitionKey = partitionKey }).ToQueryResult();
}

public IEnumerable<string> GetFetchedJobIds(string queue, int from, int perPage) => storage.Container.GetItemLinqQueryable<Documents.Queue>(requestOptions: new QueryRequestOptions { PartitionKey = partitionKey })
.Where(q => q.Name == queue && q.FetchedAt.IsDefined())
.OrderBy(q => q.CreatedOn)
.Skip(from).Take(perPage)
.Select(q => q.JobId)
.ToQueryResult()
.ToList();
public IEnumerable<string> GetFetchedJobIds(string queue, int from, int perPage)
{
QueryDefinition sql = new QueryDefinition("SELECT VALUE doc.job_id FROM doc WHERE doc.name = @name AND IS_DEFINED(doc.fetched_at) AND NOT IS_NULL(doc.fetched_at) ORDER BY doc.created_on OFFSET @offset LIMIT @limit")
.WithParameter("@name", queue)
.WithParameter("@offset", from)
.WithParameter("@limit", perPage);

return storage.Container.GetItemQueryIterator<string>(sql, requestOptions: new QueryRequestOptions { PartitionKey = partitionKey }).ToQueryResult();
}

public (int? EnqueuedCount, int? FetchedCount) GetEnqueuedAndFetchedCount(string queue)
public (int? EnqueuedCount, int? FetchedCount) GetEnqueuedAndFetchedCount(string queue)
{
(int EnqueuedCount, int FetchedCount) result = storage.Container.GetItemLinqQueryable<Documents.Queue>(requestOptions: new QueryRequestOptions { PartitionKey = partitionKey })
.Where(q => q.Name == queue)
.Select(q => new { q.Name, EnqueuedCount = q.FetchedAt.IsDefined() ? 0 : 1, FetchedCount = q.FetchedAt.IsDefined() ? 1 : 0 })
.ToQueryResult()
.GroupBy(q => q.Name)
.Select(v => (EnqueuedCount: v.Sum(q => q.EnqueuedCount), FetchedCount: v.Sum(q => q.FetchedCount)))
.FirstOrDefault();
QueryDefinition sqlQueued = new QueryDefinition("SELECT VALUE COUNT(1) FROM doc WHERE doc.name = @name AND (NOT IS_DEFINED(doc.fetched_at) OR IS_NULL(doc.fetched_at))")
.WithParameter("@name", queue);
QueryDefinition sqlFetched = new QueryDefinition("SELECT VALUE COUNT(1) FROM doc WHERE doc.name = @name AND IS_DEFINED(doc.fetched_at) AND NOT IS_NULL(doc.fetched_at)")
.WithParameter("@name", queue);

return result;
int queued = storage.Container.GetItemQueryIterator<int>(sqlQueued, requestOptions: new QueryRequestOptions { PartitionKey = partitionKey })
.ToQueryResult()
.FirstOrDefault();

int fetched = storage.Container.GetItemQueryIterator<int>(sqlFetched, requestOptions: new QueryRequestOptions { PartitionKey = partitionKey })
.ToQueryResult()
.FirstOrDefault();

return new ValueTuple<int?, int?>(queued, fetched);
}
}