Isoftforge.Apps.Client.Queue 1.0.2

Isoftforge.Apps.Client.Queue

.NET client for the isoftforge Queue satellite API. Publish and consume messages over HTTP using only a base URL and an API key — no broker host, port or extra environment configuration required.

Configuration

var client = QueueClient.Create(
    baseUrl: "https://apps-api.isoftforge.it/api/queue",
    apiKey: "ifc_your-api-key");

Or with dependency injection:

services.AddQueueClient(configuration.GetSection("Queue"));

{
  "Queue": {
    "BaseUrl": "https://apps-api.isoftforge.it/api/queue",
    "ApiKey": "ifc_your-api-key"
  }
}

• BaseUrl — hub proxy https://apps-api.isoftforge.it/api/queue or dedicated host https://queue-api.isoftforge.it/api
• ApiKey — organization credential from Queue → Credentials (header X-API-Key)

The API key identifies your organization and resolves the queue environment automatically, so only the URL and API key are needed.

Program.cs (publish + always-on consumer)

Register the client (for publishing) and a background consumer (always connected) together:

var builder = WebApplication.CreateBuilder(args);

// Publish side: inject IQueueClient anywhere to publish messages.
builder.Services.AddQueueClient(builder.Configuration.GetSection("Queue"));

// Consumer side: a strongly-typed class that stays connected for the app lifetime.
builder.Services.AddQueueConsumer<OrdersConsumer>();

// ...or an inline consumer without a dedicated class:
builder.Services.AddQueueConsumer("payments", async (message, services, ct) =>
{
    var db = services.GetRequiredService<AppDbContext>();
    await db.SaveIncomingAsync(message.Body, ct);
}, options =>
{
    options.MaxMessages = 20;
    options.PollInterval = TimeSpan.FromSeconds(2);
    options.RequeueOnError = true;
});

var app = builder.Build();
app.Run();

Consumer class

Derive from QueueConsumerService, set the queue name and implement the handler. A fresh DI scope is created per message, so you can resolve scoped services (like a database context).

using Isoftforge.Apps.Client.Queue;
using Microsoft.Extensions.DependencyInjection;

public sealed class OrdersConsumer : QueueConsumerService
{
    public OrdersConsumer(IQueueClient client, IServiceScopeFactory scopeFactory)
        : base(client, scopeFactory)
    {
    }

    protected override string QueueName => "orders";

    protected override ConsumeLoopOptions ConfigureOptions() => new()
    {
        MaxMessages = 20,
        PollInterval = TimeSpan.FromSeconds(2),
        RequeueOnError = true
    };

    protected override async Task HandleAsync(QueueMessage message, IServiceProvider services, CancellationToken ct)
    {
        var db = services.GetRequiredService<AppDbContext>();
        await db.SaveOrderAsync(message.Body, ct);
        // Returning normally acknowledges the message; throwing nacks/requeues it.
    }
}

Publish

await client.PublishAsync("orders", "{ \"id\": 123 }");

await client.PublishAsync(new PublishRequest
{
    QueueName = "orders",
    Body = "{ \"id\": 123 }",
    ContentType = "application/json"
});

Consume (pull once)

var messages = await client.ConsumeAsync("orders", new ConsumeOptions { MaxMessages = 10 });
foreach (var message in messages)
{
    Console.WriteLine(message.Body);
    await client.AckAsync(message.DeliveryTag);       // confirm
    // or: await client.NackAsync(message.DeliveryTag, requeue: true);
}

Consume (continuous consumer)

ConsumeLoopAsync polls the queue and calls your handler for each message. It acknowledges automatically after the handler succeeds and nacks (requeues) when it throws. It runs until the cancellation token is cancelled.

var cts = new CancellationTokenSource();

await client.ConsumeLoopAsync("orders", async (message, ct) =>
{
    await ProcessAsync(message.Body, ct);
}, new ConsumeLoopOptions
{
    MaxMessages = 20,
    PollInterval = TimeSpan.FromSeconds(2),
    RequeueOnError = true
}, cts.Token);

Run it as a hosted background service:

public sealed class OrdersConsumer : BackgroundService
{
    private readonly IQueueClient _queue;
    public OrdersConsumer(IQueueClient queue) => _queue = queue;

    protected override Task ExecuteAsync(CancellationToken stoppingToken) =>
        _queue.ConsumeLoopAsync("orders", HandleAsync, new ConsumeLoopOptions(), stoppingToken);

    private async Task HandleAsync(QueueMessage message, CancellationToken ct)
    {
        // process message.Body
    }
}

Queue management

var queues = await client.ListQueuesAsync();
await client.CreateQueueAsync(new CreateQueueRequest { Name = "orders", Durable = true });

Validation

var validation = await client.ValidateCredentialsAsync();
if (!validation.Valid) throw new InvalidOperationException(validation.Error);

API surface

• IQueueClient — ValidateCredentialsAsync, ListQueuesAsync, CreateQueueAsync, PublishAsync, ConsumeAsync, AckAsync, NackAsync, ConsumeLoopAsync
• QueueClientException — thrown on HTTP errors; includes StatusCode and the API error code when available

Consuming happens over HTTP (polling). For high-throughput streaming you can still connect any AMQP 0-9-1 client to the broker on port 5672.