YMJake.Aspire.Apache.RocketMQ 1.2.8

YMJake.Aspire.Apache.RocketMQ

Client-side helpers for using the published YMJake.RocketMQ.Client and YMJake.RocketMQ.Client.OpenTelemetry packages from .NET Aspire apps. The package can automatically wire RocketMQ health checks, logging, tracing, and metrics unless disabled in settings.

Recommended usage

Use rocketmq as the Aspire connection name so the client binds to ConnectionStrings:rocketmq.

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();

If you are not using Aspire, you can configure the connection string directly:

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

var builder = WebApplication.CreateBuilder(args);

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

var app = builder.Build();

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 standalone"))
        .Build();

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

app.Run();

Consumer APIs

The package also exposes consumer-oriented registrations:

• AddRocketMQConsumer(...) for SimpleConsumer
• AddRocketMQPushConsumer(...) for PushConsumer
• AddRocketMQLitePushConsumer(...) for LitePushConsumer
• AddRocketMQLiteSimpleConsumer(...) for LiteSimpleConsumer is currently experimental and not recommended for RocketMQ 5.5.

Keyed overloads are available for multi-connection scenarios.

Configuration

Supported configuration keys:

• 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

DisableLogging is honored by the client package and switches the underlying RocketMQ logger factory to a null logger for the process. DisableHealthChecks, DisableTracing, and DisableMetrics are also honored and control the automatic Aspire wiring for those integrations.

Example configuration:

{
  "Aspire": {
    "Apache": {
      "RocketMQ": {
        "Producer": {
          "rocketmq": {
            "EnableSsl": false,
            "MaxStartupAttempts": 10
          }
        },
        "Consumer": {
          "rocketmq": {
            "EnableSsl": false,
            "MaxStartupAttempts": 10
          }
        }
      }
    }
  }
}

For standalone apps, set ConnectionString directly on RocketMQProducerSettings or RocketMQConsumerSettings instead of using configuration binding. The connection string must point to the RocketMQ 5.x proxy gRPC endpoint, not the NameServer remoting endpoint.

Demo flow

The demo app uses the manual pull model:

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

Notes

• RocketMQ 5.5 cluster mode has been verified with Producer, SimpleConsumer, PushConsumer, LiteProducer, and LitePushConsumer.
• AddRocketMQLitePushConsumer(...) requires a LITE topic and a consumer group configured with lite.bind.topic and lite.sub.model.
• RocketMQ 5.5 local mode is not suitable for Lite consumer validation. LitePushConsumer fails there because the server-side local mode does not initialize LiteSubscriptionService.
• AddRocketMQLiteSimpleConsumer(...) currently maps to LiteSimpleConsumer, but RocketMQ 5.5 proxy still throws server-side errors for LITE_SIMPLE_CONSUMER; treat it as experimental/unsupported for now.
• MaxStartupAttempts is surfaced because the underlying client fetches topic routes during startup.
• Health checks, tracing, and metrics are wired automatically by this package unless you disable them in settings. The underlying integration packages are still used internally.
• AddRocketMQConsumer(...) uses SimpleConsumer, so its Receive(...) calls produce the visible receive trace shown in Aspire.
• AddRocketMQPushConsumer(...) uses PushConsumer, and its Consume(...) callback now emits an independent consumer span linked back to the producer span.
• AddRocketMQLitePushConsumer(...) follows the same linked-consumer span model as PushConsumer, while keeping the lite push subscription flow.

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.