Transports NATS

NATS Transport

Request and reply between services over NATS: one subject for each distributed key, and the options of the NATS client as they are.

Installation

dotnet add package FxMap.Nats

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. NatsOpts gives you the options of the NATS client:

builder.Services.AddFxMap(cfg =>
{
    cfg.AddEntitiesFromAssemblyContaining<SomeEntityAssemblyMarker>();
    cfg.AddProfilesFromAssemblyContaining<SomeProfileAssemblyMarker>();
    cfg.AddNats(c => c.NatsOpts(opts => opts.Url = "nats://localhost:4222"));
});

More options

cfg.AddNats(c =>
{
    c.TopicPrefix("billing");               // optional, see Naming
    c.NatsOpts(opts =>
    {
        opts.Url = "nats://localhost:4222";
        opts.MaxReconnectRetry = 5;
        opts.ConnectTimeout = TimeSpan.FromSeconds(5);
        // ... any other property of FxNatsOpts
    });
});

FxNatsOpts exposes the properties of the NATS client options: Url, Name, AuthOpts, TlsOpts, MaxReconnectRetry, ReconnectWaitMin and ReconnectWaitMax, PingInterval, ConnectTimeout and more. Each starts with the default of the NATS client.

Authentication and TLS

Credentials and TLS are the AuthOpts and TlsOpts of the NATS client. FxMap does not wrap them:

cfg.AddNats(c => c.NatsOpts(opts =>
{
    opts.Url = "tls://nats.example.com:4222";
    opts.AuthOpts = new NatsAuthOpts { Username = "svc-orders", Password = builder.Configuration["Nats:Password"] };
    opts.TlsOpts = new NatsTlsOpts { Mode = TlsMode.Require };
}));

The names and the values of these types belong to the NATS client library (NATS.Net): see its documentation for tokens, NKeys, credentials files and certificates.

Server: the service that owns the data

Register the entities and a data provider as usual, and add NATS. The service subscribes to the subject of each distributed key it owns, and no endpoint is needed:

builder.Services.AddFxMap(cfg =>
    {
        cfg.AddEntitiesFromAssemblyContaining<UserConfig>();
        cfg.AddNats(c => c.NatsOpts(opts => opts.Url = "nats://localhost:4222"));
    })
    .AddEntityFrameworkCore(c => c.AddDbContexts(typeof(UserDbContext)));

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 with the request and reply of NATS to the subject of the key, with the headers of the IContext and the trace context.
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.

Naming

FxMap uses one subject for each distributed key: fxmap-{namespace}-{type name}, in lower case, for example fxmap-myapp.contracts-userdistributedkey. With TopicPrefix("billing") the subject becomes billing-fxmap-{namespace}-{type name}, which keeps the services of two environments or two teams apart on one NATS server. Do not use these subjects for anything else.

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 of the subjects run under the supervisor, which restarts a server that fails. Since 4.1.5 a server can be started again after it was stopped.

One request, then the owner

This is plain request and reply of NATS core, with no stream: a request that nobody answers ends at the timeout and is not stored. FxMap does not use JetStream.

A known limit

A request sent in the few milliseconds after a server was stopped can reach its subscription before the NATS server has processed the unsubscribe, and nobody answers it: it ends at the request timeout. It matters only when a server is stopped and started again at once, as the supervisor does with OneForAll and RestForOne.

---

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