[christopherread]

Next v7 increment: UseActiveMq() for #84, following UseRabbitMq() (#112), UseNats() (#118) and UseKafka() (#119). It also carries message headers, including trace context, over ActiveMQ (#90). Targets v7.

API

builder.Services.AddObvs(obvs => obvs
    .AddService<IOrderMessage>("orders", service => service
        .UseActiveMq(builder.Configuration.GetSection("Obvs:Orders:ActiveMq"))   // or .UseActiveMq(o => o.BrokerUri = "activemq:tcp://localhost:61616")
        .SerializedWith(new JsonMessageSerializer(), new JsonMessageDeserializerFactory(typeof(JsonMessageDeserializer<>)))));

• Options: ActiveMqOptions has BrokerUri (required), UserName and Password. Each service's settings are named options keyed by the service name: IOptionsMonitor<ActiveMqOptions>.Get("orders").
• Startup checks: a missing broker URI or serializer fails at startup, inside the same ObvsConfigurationException as any other problem. The message names the configuration key to set (for example Obvs:Orders:ActiveMq:BrokerUri) or the method to call.
• Topics only: every subscriber on a topic gets every message, so each bus receives every message. v6's option to send chosen message types through queues isn't exposed yet.
• Client IDs include the bus's instance ID. On .NET Framework, v6 builds the client ID from the process, user and service. Two buses for the same service in one process would get the same ID, and the broker refuses a duplicate. On .NET 10 the client ID already includes a random GUID, and the tests only run there, so this change isn't covered by a test.

A transport bug this exposed and its fix

Deadlock when disposing the bus after receiving a message. It's the same bug as in RabbitMQ (#113) and Kafka (#119, #120).

• Cause: messages reached subscribers on the session's dispatch thread. Code that ran after a message arrived, including the code after awaiting one, therefore ran on that thread. Disposing the bus there stops the connection, which waits for that same thread forever.
• Effect: every broker test hung when disposing its host. A stack dump showed Connection.Stop() → SessionExecutor.Stop() waiting on the dispatch thread it was running on.
• Fix: messages are still deserialized and acknowledged on the dispatch thread, then handed to subscribers on the task pool, in order. This also applies to v6's fluent setup.
• Existing tests changed: five tests in TestMessageSource.cs asserted OnNext straight after raising a fake message, which assumed delivery on the listener's thread. They now wait up to 5 seconds for delivery and then make the same assertion. Their assertions about deserializing and acknowledging are unchanged, since those still happen on the listener's thread.

Headers over ActiveMQ

• Sending: each Obvs header becomes a message property under its own name, so other tools can read traceparent too. Obvs's own type-name property is never overwritten.
• Receiving: every text property except the type name becomes a header on the received message. Properties set through v6's property-provider option therefore show up as headers too.
• Header names: the broker and client accept names that aren't Java identifiers, such as x-tenant.
• Compatibility: v6 receivers ignore the extra properties, so v6 and v7 services can still exchange messages.

CI

• Broker: adds an apache/activemq-classic:6.2.0 service container. 6.2.0 is the newest Classic image tag on Docker Hub; locally I used Homebrew's 6.3.2.
• Tests: runs the ActiveMQ broker tests alongside the other transports.

Tests (written first)

• Configuration (no broker): 5 tests in Obvs.ActiveMQ.Tests/Hosting/TestUseActiveMq.cs. They cover:
  • options from configuration, including credentials;
  • options from code;
  • a missing broker URI, in configuration and in code;
  • a missing serializer.
• End to end (Explicit, real broker): 7 tests in TestActiveMqHosting.cs. They cover:
  • commands and events;
  • request and response;
  • every bus in a process receiving every message;
  • two hosts in one process;
  • disposing the bus from a message callback;
  • trace continuation;
  • headers the app sets.
• Red first:
  • All 12 failed with NotImplementedException before implementation.
  • Once UseActiveMq worked, every broker test hung on the deadlock above.
  • After the fix, the header and trace tests still failed on a missing x-tenant header until headers were carried.
• Results:
  • All 7 broker tests passed 3 runs in a row against a local ActiveMQ 6.3.2, taking about a second per run.
  • The 24 ActiveMQ unit tests passed 5 runs in a row.
  • A clean build of the solution has 0 warnings, and all 220 unit tests pass.

Not in this increment

• UseNetMq() with header support, which is the last transport.
• Exposing ActiveMQ queues and acknowledgement modes in ActiveMqOptions.

🤖 Generated with Claude Code