Transports RabbitMQ

RabbitMQ Transport

Request and reply between services over RabbitMQ: one exchange for each distributed key, one shared connection with automatic recovery, and a pool of channels.

Installation

dotnet add package FxMap.RabbitMq

Every service that maps data needs the core FxMap package, and the owning service also needs a data provider. All FxMap.* packages must use the same version.

Configuration

The same call is used on the service that needs data and on the service that owns it. Host takes the host and the virtual host:

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddProfilesFromAssemblyContaining<SomeProfileAssemblyMarker>();
 
    // Local development
    cfg.AddRabbitMq(config => config.Host("localhost", "/"));
});

Credentials, port and channel pool

cfg.AddRabbitMq(config =>
{
    config.Host("rabbit.internal", "/", 5672, c =>
    {
        c.UserName("SomeUserName");
        c.Password(builder.Configuration["RabbitMq:Password"]);
    });
    config.ChannelPoolSize(10);   // default 5
});

TLS

Ssl(...) is part of the credentials. It gives you the SslOption of the RabbitMQ client:

cfg.AddRabbitMq(config =>
{
    config.Host("rabbit.example.com", "/", 5671, c =>
    {
        c.UserName("SomeUserName");
        c.Password(builder.Configuration["RabbitMq:Password"]);
        c.Ssl(o =>
        {
            o.Enabled = true;
            o.ServerName = "rabbit.example.com";
        });
    });
});

Server: the service that owns the data

Register the entities and a data provider as usual, and add RabbitMQ. The service consumes the requests of each distributed key it owns, and no endpoint is needed:

builder.Services.AddFxMap(cfg =>
    {
        cfg.AddEntitiesFromAssemblyContaining<UserConfig>();
        cfg.AddRabbitMq(config => config.Host("localhost", "/"));
    })
    .AddEntityFrameworkCore(c => c.AddDbContexts(typeof(UserDbContext)));

Connection and channels

The client and the server of a process share one RabbitMQ connection, opened lazily and with automatic recovery, and a pool of channels (ChannelPoolSize, default 5). The server opens one consumer for each pooled channel, and the concurrency is driven by the native consumerDispatchConcurrency of RabbitMQ.

Naming

FxMap uses exchanges named FxMap-{namespace}:{type name} and creates the queue fxmap-rpc-queue-[application friendly name]. Do not use these names for anything else.

Recovery after an outage

Since 4.1.5 the consumer works again after the connection to the broker was lost. The client chooses the name of its reply queue (fxmap-reply-<guid>, exclusive and auto-delete), so the recovery declares the same queue again. Before 4.1.5 the broker chose the name, and a recovered connection declared a queue with another name while the requests kept the old one in ReplyTo: every request timed out until the process restarted.

The consumer works again about 10 seconds after the broker is back, which is the recovery interval of the connection. A server whose start failed half way no longer ends with two consumers of the same queue when it is started again.

How a request travels

1. Each request holds the ids and the expressions for one distributed key and one level. A hundred objects that need a UserName make one request.
2. The client publishes it to the exchange of the key, with the headers of the IContext, and waits on its reply queue.
3. The owner answers with the values, and with the expression errors it isolated (4.1.3+).
4. If nobody answers within the request timeout (30 seconds by default, cfg.SetRequestTimeOut(...)), the request fails. See Retry Policy to retry it.

Complete example

// Shared contract
public sealed class UserDistributedKey : IDistributedKey;
 
// Owner (users service): the entity and how it is exposed
public class UserConfig : EntityConfigureOf<User>
{
    protected override void Configure()
    {
        Id(x => x.Id);
        DefaultProperty(x => x.Name);
        UseDistributedKey<UserDistributedKey>();
    }
}
 
// Client (orders service): the DTO and where its values come from
public class OrderResponse
{
    public string Id { get; set; }
    public string UserId { get; set; }
    public string? UserName { get; set; }
    public string? UserEmail { get; set; }
}
 
public class OrderResponseProfile : ProfileOf<OrderResponse>
{
    protected override void Configure() =>
        UseDistributedKey<UserDistributedKey>()
            .Of(x => x.UserId)
            .For(x => x.UserName)
            .For(x => x.UserEmail, "Email");
}
 
// In an endpoint of the orders service
await distributedMapper.MapDataAsync(orders);

Good to know

Supervised servers

The consumers run under the supervisor, which restarts a server that fails.

The same on both sides

The client and the owner must use the same transport and the same virtual host. The profiles and the entity configs do not change if you switch to another transport.

Keep secrets out of code

Read the user name and the password from configuration or a secret store, as in the examples above.

---

Source: https://fxmapper.net/docs/transports/rabbitmq/ — the website is always the most up to date version.