Isoftforge.Apps.Client.Queue 1.0.3

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

Every connector can set its own client name, shown on the portal Connections page (publisher and consumer appear as separate connections). Set it in one of three ways:

• Globally in configuration (Queue:ClientName) — applies to every request.
• Per consumer/publisher by overriding the ClientName property (see below).
• Per call via ConsumeLoopOptions.ClientName / PublishRequest.ClientName.

Publisher class

Derive from QueuePublisher, optionally override ClientName, 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) { }

    protected override string? ClientName => "orders-service";

    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 string? ClientName => "orders-service";

    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.