Isoftforge.Apps.Client.Queue 1.0.6

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 (publisher + always-on consumer)

Register the client, a strongly-typed publisher and a background consumer (always connected) together:

var builder = WebApplication.CreateBuilder(args);

// Base client: required by publishers and consumers. Inject IQueueClient directly if you prefer.
builder.Services.AddQueueClient(builder.Configuration.GetSection("Queue"));

// Publisher: a strongly-typed class you inject to publish messages.
builder.Services.AddQueuePublisher<OrdersPublisher>();

// Consumer: 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;
    options.ClientName = "payments-worker"; // shown on the portal Connections page
});

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

Client name

The client name is configured once per client in Queue:ClientName and applies to every request. It labels the connection on the portal Connections page, where the publisher and consumer of the same app appear as two separate connections (suffixed -publisher and -consumer). Because the name is sent on every request (heartbeat, publish, consume, ack, nack), the always-on connection and your actual publishes always map to the same connection — no duplicates.

If ClientName is not set, connections show as http-publisher / http-consumer.

Publisher class

Derive from QueuePublisher and expose your own public methods that call the protected PublishAsync.

using Isoftforge.Apps.Client.Queue;

public sealed class OrdersPublisher : QueuePublisher
{
    public OrdersPublisher(IQueueClient client) : base(client) { }

    public Task<PublishResult> PublishOrderAsync(string json, CancellationToken ct = default)
        => PublishAsync("orders", json, ct);
}

Inject and use it:

public sealed class OrdersController(OrdersPublisher publisher)
{
    public Task Create(string json) => publisher.PublishOrderAsync(json);
}

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.DeliveryUid);       // confirm
    // or: await client.NackAsync(message.DeliveryUid, 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
    }
}

Connections & channels

Publishers and consumers registered with AddQueuePublisher<T>() / AddQueueConsumer<T>() show up on the portal Connections page while the app is running, each with an automatically created channel.

• Graceful shutdown removes the connection immediately (DisconnectAsync is called for you).
• Crash / hard kill is handled by the server: connections that stop refreshing are swept after ~60s.
• Channel pause (portal Channels page) is enforced: a paused Consumer channel receives no messages, and a paused Publisher channel rejects PublishAsync (throws QueueClientException with StatusCode == 409) until it is set back to Active.

You can also drive this manually:

await client.HeartbeatAsync(QueueRole.Publisher);   // keep a connection visible
await client.DisconnectAsync(QueueRole.Publisher);  // remove it

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, HeartbeatAsync, DisconnectAsync
• 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.