Queues (Send)

using MassTransit;
using MassTransit.KubeMQTransport;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();
    x.UsingKubeMQ((context, cfg) =>
    {
        cfg.Host("localhost", 50000);
        cfg.ReceiveEndpoint("send-basic-queue", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

builder.Services.AddHostedService<BasicSendSender>();

var host = builder.Build();
await host.RunAsync();

// --- Message type ---
public record OrderMessage(Guid OrderId, string ProductName, decimal Amount);

// --- Consumer ---
public class OrderConsumer(ILogger<OrderConsumer> logger) : IConsumer<OrderMessage>
{
    public Task Consume(ConsumeContext<OrderMessage> context)
    {
        logger.LogInformation(
            "Received order: OrderId={OrderId}, Product={Product}, Amount={Amount:C}",
            context.Message.OrderId,
            context.Message.ProductName,
            context.Message.Amount);
        return Task.CompletedTask;
    }
}

// --- Background sender ---
public class BasicSendSender(
    ILogger<BasicSendSender> logger,
    IHostApplicationLifetime lifetime) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await Task.Delay(2000, stoppingToken);

        var rider = KubeMQRiderAccessor.Current
            ?? throw new InvalidOperationException("KubeMQ rider not started.");

        var endpoint = await rider.GetSendEndpoint(
            new Uri("kubemq://localhost:50000/send-basic-queue"), stoppingToken);

        var order = new OrderMessage(Guid.NewGuid(), "Widget", 29.99m);
        await endpoint.Send(order, stoppingToken);
        logger.LogInformation("Message sent successfully.");

        await Task.Delay(5000, stoppingToken);
        lifetime.StopApplication();
    }
}

var endpoint = await rider.GetSendEndpoint(
    new Uri("kubemq://localhost:50000/send-delayed-queue"), stoppingToken);

var reminder = new ReminderMessage(
    Guid.NewGuid(), "Follow up with client", DateTimeOffset.UtcNow.AddSeconds(5));

// Set the delay on the send context — transport maps this to QueueMessage.DelaySeconds
await endpoint.Send(reminder, context =>
{
    context.Delay = TimeSpan.FromSeconds(5);
}, stoppingToken);

// Send
await endpoint.Send(invoice, context =>
{
    context.Headers.Set("tenant-id", "acme-corp");
    context.Headers.Set("x-correlation-hint", "batch-2024-q1");
}, stoppingToken);

// Receive
public class HeaderConsumer(ILogger<HeaderConsumer> logger) : IConsumer<InvoiceMessage>
{
    public Task Consume(ConsumeContext<InvoiceMessage> context)
    {
        var tenantId = context.Headers.Get<string>("tenant-id") ?? "(not set)";
        logger.LogInformation("Received invoice; tenant-id={TenantId}", tenantId);
        return Task.CompletedTask;
    }
}

public record BatchOrderMessage(Guid OrderId, string ProductName, decimal Amount, int Sequence);

public class OrderBatchConsumer(ILogger<OrderBatchConsumer> logger)
    : IConsumer<Batch<BatchOrderMessage>>
{
    public Task Consume(ConsumeContext<Batch<BatchOrderMessage>> context)
    {
        logger.LogInformation("Received batch of {Count} messages", context.Message.Length);

        foreach (var msg in context.Message)
        {
            logger.LogInformation(
                "  Batch item: Seq={Sequence}, OrderId={OrderId}",
                msg.Message.Sequence, msg.Message.OrderId);
        }

        return Task.CompletedTask;
    }
}

// Configures batch size and time limits
public class OrderBatchConsumerDefinition : ConsumerDefinition<OrderBatchConsumer>
{
    protected override void ConfigureConsumer(
        IReceiveEndpointConfigurator endpointConfigurator,
        IConsumerConfigurator<OrderBatchConsumer> consumerConfigurator,
        IRegistrationContext context)
    {
        consumerConfigurator.Options<BatchOptions>(options => options
            .SetMessageLimit(10)
            .SetTimeLimit(TimeSpan.FromSeconds(5)));
    }
}

x.UsingKubeMQ((context, cfg) =>
{
    cfg.Host("localhost", 50000);

    // Two endpoints on the same channel compete for messages.
    // KubeMQ delivers each message to exactly one member.
    cfg.ReceiveEndpoint("competing-consumers-queue", e =>
    {
        e.ConfigureKubeMQ(t => { });
    });

    cfg.ReceiveEndpoint("competing-consumers-queue", e =>
    {
        e.ConfigureKubeMQ(t => { });
    });
});

x.UsingKubeMQ((context, cfg) =>
{
    cfg.Host("localhost", 50000);

    // Default weights: high=3, normal=2, low=1
    cfg.UsePriorityQueues();

    cfg.ReceiveEndpoint("pq-basic-orders", e =>
    {
        e.ConfigureKubeMQ(t => { });
    });
});

// Send to a specific priority level by addressing the suffixed channel
var highEndpoint = await rider.GetSendEndpoint(
    new Uri("kubemq://localhost:50000/pq-basic-orders_high"), stoppingToken);
await highEndpoint.Send(new PriorityOrder(Guid.NewGuid(), "HIGH", "HighItem", 100.00m), stoppingToken);

// high-priority channel is polled 5x per cycle vs 1x for low
cfg.UsePriorityQueues(p => p.SetWeights(highWeight: 5, normalWeight: 3, lowWeight: 1));

cfg.ReceiveEndpoint("order-processing", e =>
{
    e.PollTimeoutSeconds = 10;   // wait up to 10s per poll
    e.MaxPollMessages = 64;      // fetch up to 64 messages per poll
});