YMJake.Aspire.Hosting.RocketMQ 1.0.2

Aspire RocketMQ libraries

Aspire helpers for RocketMQ hosting and client wiring.

Packages

• YMJake.Aspire.Hosting.RocketMQ: AppHost extensions for running a RocketMQ NameServer-led cluster, broker, and optional dashboard in .NET Aspire.
• YMJake.Aspire.Apache.RocketMQ: app-side helpers for registering RocketMQ producers and consumers from Aspire connection strings.
• YMJake.AspNetCore.HealthChecks.RocketMQ: ASP.NET Core health checks for probing RocketMQ connectivity from application code.

Hosting

The hosting package is designed around the RocketMQ NameServer-led deployment path:

Note: AddRocketMQNameServer("rocketmq-nameserver") uses rocketmq-nameserver as the root Aspire resource name and connection-string key. That root resource represents the RocketMQ NameServer-led cluster entry point, with broker and proxy wired underneath it.

Why this shape? RocketMQ clients first resolve routes through NameServer, brokers register themselves there, and proxy depends on the broker/name server pair. Making NameServer the root resource keeps the Aspire graph aligned with RocketMQ's startup and routing model.

using Aspire.Hosting.RocketMQ;
using Projects;

var builder = DistributedApplication.CreateBuilder(args);
var brokerConfigPath = Path.Combine("..", "rocketmq", "conf", "broker.conf");

var proxy = builder.AddRocketMQProxy(brokerConfigPath);

var rocketmq = builder.AddRocketMQNameServer("rocketmq-nameserver")
    .WithProxy(proxy)
    .WithDataBindMount("../.containers/rocketmq")
    .WithJvmArgs(
        nameServerArgs: "-Xms256m -Xmx512m",
        brokerArgs: "-Xms256m -Xmx512m");

builder.AddProject<Aspire_RocketMQ_AspNet_Demo>("api")
    .WaitFor(proxy)
    .WithReference(rocketmq);

builder.Build().Run();

WithProxy(...) is intentionally the only proxy wiring API. We do not expose a separate WithBrokerConfig(...) entry point because RocketMQ local mode depends on a single broker broadcast address and fixed networking assumptions that do not fit Aspire's dynamic port model.

Client

The client package now uses a simple, test-friendly flow:

using Aspire.RocketMQ.Client;
using RocketMQ.Client.Core;
using System.Text;
using System.Collections.Generic;

var builder = WebApplication.CreateBuilder(args);

builder.AddRocketMQProducer(
    "rocketmq",
    settings =>
    {
        settings.EnableSsl = false;
        settings.MaxStartupAttempts = 10;
    },
    producer => producer.SetTopics("AspireRocketMQSmoke"));

builder.AddRocketMQConsumer(
    "rocketmq",
    settings =>
    {
        settings.EnableSsl = false;
        settings.MaxStartupAttempts = 10;
    },
    consumer => consumer
        .SetConsumerGroup("aspire-consumer-group")
        .SetSubscriptionExpression(new Dictionary<string, FilterExpression>
        {
            ["AspireRocketMQSmoke"] = new FilterExpression("*")
        })
        .SetAwaitDuration(TimeSpan.FromSeconds(3)));

var app = builder.Build();

app.MapGet("/rocketmq", (RocketMQProducerSettings rocketmq) => new
{
    rocketmq.Endpoints,
    rocketmq.RequestTimeout,
    rocketmq.EnableSsl,
    rocketmq.Namespace,
    rocketmq.MaxStartupAttempts
});

app.MapPost("/rocketmq/send", async (Producer producer) =>
{
    var message = new Message.Builder()
        .SetTopic("AspireRocketMQSmoke")
        .SetKeys(Guid.NewGuid().ToString("N"))
        .SetBody(Encoding.UTF8.GetBytes("hello from aspire"))
        .Build();

    var receipt = await producer.Send(message);
    return Results.Ok(new { receipt.MessageId, message.Topic });
});

app.MapPost("/rocketmq/consume", async (SimpleConsumer consumer, int? maxMessageNum, int? invisibleDurationMs) =>
{
    var safeInvisibleDurationMs = Math.Max(invisibleDurationMs.GetValueOrDefault(10_000), 10_000);
    var messages = await consumer.Receive(
        maxMessageNum.GetValueOrDefault(1),
        TimeSpan.FromMilliseconds(safeInvisibleDurationMs));

    return Results.Ok(messages.Select(message => new
    {
        message.MessageId,
        message.Topic,
        message.Keys,
        Body = Encoding.UTF8.GetString(message.Body)
    }));
});

app.Run();

Client registrations

• AddRocketMQProducer(...)
• AddRocketMQConsumer(...)
• AddRocketMQPushConsumer(...)
• AddRocketMQLitePushConsumer(...)

Keyed overloads are available for multi-connection scenarios.

Configuration

• ConnectionStrings:{name}
• Aspire:Apache:RocketMQ:Producer:{name}:RequestTimeout
• Aspire:Apache:RocketMQ:Producer:{name}:EnableSsl
• Aspire:Apache:RocketMQ:Producer:{name}:DisableHealthChecks
• Aspire:Apache:RocketMQ:Producer:{name}:DisableTracing
• Aspire:Apache:RocketMQ:Producer:{name}:DisableMetrics
• Aspire:Apache:RocketMQ:Producer:{name}:DisableLogging
• Aspire:Apache:RocketMQ:Producer:{name}:Namespace
• Aspire:Apache:RocketMQ:Producer:{name}:MaxStartupAttempts
• Aspire:Apache:RocketMQ:Consumer:{name}:RequestTimeout
• Aspire:Apache:RocketMQ:Consumer:{name}:EnableSsl
• Aspire:Apache:RocketMQ:Consumer:{name}:DisableHealthChecks
• Aspire:Apache:RocketMQ:Consumer:{name}:DisableTracing
• Aspire:Apache:RocketMQ:Consumer:{name}:DisableMetrics
• Aspire:Apache:RocketMQ:Consumer:{name}:DisableLogging
• Aspire:Apache:RocketMQ:Consumer:{name}:Namespace
• Aspire:Apache:RocketMQ:Consumer:{name}:MaxStartupAttempts

Demo flow

1. POST /rocketmq/send publishes a message.
2. POST /rocketmq/consume?maxMessageNum=1 pulls messages.
3. invisibleDurationMs must be at least 10000.

Notes

• The underlying RocketMQ client fetches topic routes during startup, so MaxStartupAttempts is surfaced as a first-class setting.
• The Aspire client package automatically wires RocketMQ health checks, tracing, and metrics unless disabled in settings. The underlying health check and OpenTelemetry packages are still used internally.

Deployment plan

• The current hosting health checks use 127.0.0.1 because they are designed for local Aspire/AppHost development.
• For Docker or Kubernetes deployments, the broker and proxy probe hosts should be made configurable so they can target service DNS names or in-cluster addresses instead of loopback.
• The next release should keep the local defaults but expose an override path for non-local deployments.

Health checks

The health check package is application-facing and probes RocketMQ by starting a short-lived client instance. By default it only verifies connectivity and route fetch, not consumer subscription state. It accepts a direct ConnectionString or a ConnectionStringFactory, so it works both with and without Aspire. It is separate from hosting readiness checks:

• YMJake.Aspire.Hosting.RocketMQ covers container readiness and startup ordering.
• YMJake.AspNetCore.HealthChecks.RocketMQ covers app-level RocketMQ dependency checks.