Obvs Observable Microservice Bus

Obvs uses a convention-based messaging system. To organize messages, you should define a root message interface that identifies messages belonging to your specific service. You then implement specific message types (Commands, Events, Requests, Responses) by inheriting from both your service's root interface and the appropriate Obvs marker interface.

• IMessage: The base interface for all messages.
• ICommand: For actions that trigger a state change.
• IEvent: For notifications of things that have happened.
• IRequest: For synchronous-style requests.
• IResponse: For the result of a request.

// 1. Define a root message type for your service
public interface IMyServiceMessage : IMessage { }

// 2. Create specific message types
public class MyCommand : IMyServiceMessage, ICommand { }
public class MyEvent : IMyServiceMessage, IEvent { }
public class MyRequest : IMyServiceMessage, IRequest { }
public class MyResponse : IMyServiceMessage, IResponse { }

You can declare a new IServiceBus using the ServiceBus.Configure() fluent builder. This allows you to specify the transport (e.g., ActiveMQ), the message contract scope, naming, queue/topic usage, connection details, and serialization format.

Key configuration methods:

• .WithActiveMQEndpoints<T>(): Configures endpoints for messages implementing T.
• .Named(string): Sets the service name.
• .UsingQueueFor<T>(): Specifies that a specific message type (like ICommand) should use a queue.
• .ConnectToBroker(string): Provides the connection string.
• .SerializedAsJson(): Sets the serialization format.
• .AsClientAndServer(): Configures the bus to both send and receive messages.
• .WithEndpoints(IServiceEndpointClient): Adds custom endpoints for external integrations.

IServiceBus serviceBus = ServiceBus.Configure()
    .WithActiveMQEndpoints<IMyServiceMessage>()
        .Named("MyService")
        .UsingQueueFor<ICommand>()
        .ConnectToBroker("tcp://localhost:61616")
        .SerializedAsJson()
        .AsClientAndServer()
    .Create();

Obvs supports a request/response pattern using IRequest and IResponse.

1. Handling a Request: Subscribe to the .Requests stream and use serviceBus.ReplyAsync(request, response) to send a reply.
2. Getting a Response: Use serviceBus.GetResponses(request) which returns an IObservable<IResponse>. You can then use Rx operators like .OfType<T>(), .Take(1), and .Timeout() to manage the response lifecycle.

// 1. Server-side: Handle the request and reply
serviceBus.Requests
      .OfType<MyRequest>()
      .Subscribe(request => serviceBus.ReplyAsync(request, new MyResponse()));

// 2. Client-side: Send request and wait for response
serviceBus.GetResponses(new MyRequest())
      .OfType<MyResponse>()
      .Take(1)
      .Timeout(TimeSpan.FromSeconds(1))
      .Subscribe(
          r => Console.WriteLine("Received a response!"), 
          err => Console.WriteLine("Oh no!")
      );

To wrap external APIs or non-Obvs messaging systems into your bus, implement the IServiceEndpointClient interface. This allows you to standardize messaging semantics by treating external integrations as custom endpoints.

You must implement:

• Events: An IObservable<IEvent> representing incoming external events.
• SendAsync(ICommand): Logic to call the external system for commands.
• GetResponses(IRequest): Logic to wrap an external API call into an IObservable<IResponse>.
• CanHandle(IMessage): A predicate to determine if this endpoint should process a specific message type.

public class MyCustomEndpoint : IServiceEndpointClient
{
    Type _serviceType = typeof(IMyCustomServiceMessage);

    public IObservable<IEvent> Events
    {
        get { /* subscribe to external MQ broker */ }
    }

    public Task SendAsync(ICommand command)
    {
        /* call external API */
    }

    public IObservable<IResponse> GetResponses(IRequest request)
    {
        /* call external API and wrap response in observable */
    }

    public bool CanHandle(IMessage message)
    {
        return _serviceType.IsInstanceOfType(message);
    }
}

// Register the endpoint during configuration
IServiceBus serviceBus = ServiceBus.Configure()
    .WithActiveMQEndpoints<IMyServiceMessage>()
    // ... other config ...
    .WithEndpoints(new MyCustomEndpoint())
    .Create();