Obvs: an observable microservice bus

observable services, obviously

Obvs is a .NET library for messaging between services. Each service has commands, events, requests and responses. You choose its transport, such as RabbitMQ, Kafka, NATS, Azure Service Bus or Amazon SQS. Your code handles its messages, or reads them as Reactive Extensions streams.

Quick start

dotnet add package Obvs
dotnet add package Obvs.RabbitMQ

Define your service's messages:

// Every message of a service implements its interface, and is one of ICommand, IEvent, IRequest or IResponse.
// A request names the response that answers it.
public interface IOrderMessage : IMessage { }

public record PlaceOrder(string OrderId) : IOrderMessage, ICommand;

public record OrderPlaced(string OrderId) : IOrderMessage, IEvent;

public record GetQuote(string Sku) : IOrderMessage, IRequest<Quote>
{
    public string RequestId { get; set; }
    public string RequesterId { get; set; }
}

public record Quote(string Sku, decimal Price) : IOrderMessage, IResponse
{
    public string RequestId { get; set; }
    public string RequesterId { get; set; }
}

Host the service: it handles commands and requests, and publishes events.

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddObvs(obvs => obvs
    .AddService<IOrderMessage>("orders", service => service
        .UseRabbitMq(options => options.Uri = new Uri("amqp://localhost:5672"))
        .HandleCommand<PlaceOrder>(async (command, context, ct) =>
        {
            // ... place the order, then tell whoever's interested
            await context.PublishAsync(new OrderPlaced(command.OrderId), ct);
        })
        .HandleRequest<GetQuote, Quote>((request, context, ct) =>
            Task.FromResult(new Quote(request.Sku, 9.99m)))));

await builder.Build().RunAsync();

Then send it commands and requests from another app, a client of the service:

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddObvs(obvs => obvs
    .AddService<IOrderMessage>("orders", service => service
        .UseRabbitMq(options => options.Uri = new Uri("amqp://localhost:5672"))
        .AsClient()));

using var app = builder.Build();
await app.StartAsync();

var client = app.Services.GetRequiredService<IServiceBusClient>();

await client.SendAsync(new PlaceOrder("o-1"));

var quote = await client.RequestAsync(new GetQuote("sku-1"), TimeSpan.FromSeconds(5));

Everything on the bus is an Rx stream: Commands, Events, Requests and GetResponses(request). Compose them with Rx operators, and process them with SubscribeAsync, which awaits your code and settles each message. Handlers are the same thing, registered at startup with a DI scope per message, and RequestAsync is a request with one response.

Next steps

• The Obvs site, with what's new in v7.
• Common tasks: add a service, handle and send messages, requests, errors, tests, tracing.
• Testing with an in-memory broker.
• Transports: RabbitMQ, Azure Service Bus, Kafka (and Confluent Cloud, Amazon MSK, Azure Event Hubs), Amazon SQS and SNS, NATS and JetStream, NetMQ.
• Configuration errors and analyzers.
• Service contracts as AsyncAPI, and an MCP server for agents.
• Migrating from v6.
• llms.txt, for agents, and AGENTS.md, for working on Obvs itself.

Packages

Package
Obvs	The bus, hosting with AddObvs, handlers, System.Text.Json serialization, OpenTelemetry, the analyzers and the [ObvsMessages] generator
Obvs.RabbitMQ, Obvs.Kafka, Obvs.NATS, Obvs.ActiveMQ, Obvs.AzureSB, Obvs.AmazonSqs, Obvs.NetMQ	Transports. Obvs.NATS covers core NATS and JetStream
Obvs.Serialization.Json, .MessagePack, .ProtoBuf	Other serializers: Newtonsoft JSON, MessagePack-CSharp and protobuf-net. Any serializer can be gzipped with Compressed()
Obvs.Testing	An in-memory broker and a test harness
Obvs.Mcp	An MCP server for development buses

The XML, NetJSON, YAML and MsgPack.Cli serializers, Obvs.Monitoring.* and Obvs.Logging.NLog stay on v6.

Versions

• v7 - hosting with dependency injection, handlers, RequestAsync, System.Text.Json, OpenTelemetry, Azure Service Bus and Amazon SQS, analyzers, a test kit, contracts and an MCP server. Supports netstandard2.0, net472 and net10.0
• v6 - System.Reactive, supports netstandard2.0, net472 and net10.0 (net5.0 in early 6.x). The transports moved into this repo
• v5 - System.Reactive 4.1, supports netstandard2.0 and net472
• v4 - System.Reactive 3.1.1, supports netstandard1.6 and net452

Examples

Examples/ has a client and a service on ActiveMQ, traced across both apps with OpenTelemetry's console exporter:

cd Examples
docker compose up        # ActiveMQ and the service
dotnet run --project Client