Transports Azure Service Bus

Azure Service Bus Transport

Request and reply between services over session-enabled Azure Service Bus queues: FxMap creates a request queue and a reply queue for each distributed key.

Standard or Premium tier

This transport needs the Standard or the Premium tier of Azure Service Bus, because it uses sessions. Premium is recommended for production.

Installation

dotnet add package FxMap.Azure.ServiceBus

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:

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddProfilesFromAssemblyContaining<SomeProfileAssemblyMarker>();
    cfg.AddAzureServiceBus(c =>
    {
        c.Host(builder.Configuration["ServiceBus:ConnectionString"]);   // connection string
        c.TopicPrefix("staging");        // optional
        c.MaxConcurrentSessions(16);     // optional, at least 1 (default 8)
    });
});

• Host takes the connection string, and optionally an action that sets the ServiceBusClientOptions of the Azure client (for example a retry policy or a proxy).
• TopicPrefix prepends "{prefix}-" to the names of the queues, to keep environments apart in one namespace.
• MaxConcurrentSessions is the number of sessions that a server processes at the same time.
• Only a connection string is accepted. Read it from configuration or a secret store, not from code.

Server: the service that owns the data

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

builder.Services.AddFxMap(cfg =>
    {
        cfg.AddEntitiesFromAssemblyContaining<UserConfig>();
        cfg.AddAzureServiceBus(c => c.Host(builder.Configuration["ServiceBus:ConnectionString"]));
    })
    .AddEntityFrameworkCore(c => c.AddDbContexts(typeof(UserDbContext)));

Naming and sessions

For every distributed key FxMap creates session-enabled queues at startup: fxmap-request-{namespace}-{type name} for the requests and fxmap-reply-{namespace}-{type name} for the replies, with dots replaced by hyphens and in lower case. Do not use these names for anything else.

Because the queues are created at startup, the connection string must belong to an identity that is allowed to manage entities in the namespace.

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 sends it to the request queue of the key. Both queues are session-enabled.
3. The owner answers on the reply queue of the key, 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 servers run under the supervisor, which restarts a server that fails.

The connection is closed with the container

Before 4.1.4 the AMQP connection of the namespace was never closed when the container was disposed (a host that stops, a test). A namespace or an emulator has a quota of connections, and a run of tests could end with ConnectionsQuotaExceeded. The client is now disposed with the container.

---

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