Beginning with NATS
In the world of distributed systems, event driven architecture, EDA, has emerged as a powerful paradigm for building scalable and resilient applications. At the heart of many EDA implementations lies a robust messaging system, and NATS is a compelling choice known for its speed, simplicity, and reliability.
In a crowded landscape of message brokers and event buses like Kafka, Event Hub, SQS, etc. NATS carves a unique niche by focusing on fundamental application connectivity. While offering straightforward, location transparent Pub/Sub as a core capability, NATS goes beyond simple message passing. It provides a versatile foundation for implementing complex messaging patterns, including basic Fan-Out/In, Work Queues, Request/Reply semantics, and much more. This flexibility allows us to rethink traditional communication methods, offering an elegant and often simpler alternative to REST based APIs and traditional complex ingress, and this is just scratching the surface of Core NATS before delving into its advanced features like JetStream persistence and other higher-level abstractions.
Where do we start?
Today, we'll explore how to integrate NATS Core into your .NET applications using C#. We'll walk through the process of setting up a console application, configuring the NATS connection using dependency injection, and then demonstrate the foundational patterns of Pub/Sub and Request/Reply.
We’re going to accomplish three main goals:
Stand up a C# console project with our NuGet dependencies
Integrate with the NATS server via the C# NATS client
Demo Pub/Sub and Request/Reply
Setting Up Your .NET Console Project
First things first, let's create a new .NET console application. Open your terminal or command prompt and execute the following command
dotnet new console -n NATSUnleashed.CSharpIntro
Next, we’ll need to add the official NATS.NET client library to our project along with the serializer package and usual hosting components. Run these commands
dotnet add package NATS.Client.Core
dotnet add package NATS.Client.Serializers.Json
dotnet add package Microsoft.Extensions.Hosting
Finally, open the solution and we’ll get started!
Creating the NatsService
We want a NATS service that will allow us to run our app as any of the following actors:
publisher - Publishes to NATS Server on a given subject
subscriber - Subscribes to the same subject as the publisher
requestor - Makes requests to NATS server and expects a reply
responder - Responds to requests from the requestor
NatsService - Config Options, Message Schema, Extensions
Configuration
First, we’ll create an options record to bind various configuration settings to.
public sealed record NatsConfig
{
public string NatsUrl { get; set; } = default!;
public string AppType { get; set; } = default!;
public string Subject { get; set; } = default!;
public string? QueueGroup { get; set; } = default!;
public void Deconstruct(
out string natsUrl,
out string appType,
out string subject,
out string? queueGroup)
=> (natsUrl, appType, subject, queueGroup) = (NatsUrl, AppType, Subject, QueueGroup);
}
Simply a record that captures the server URL, the type of actor to run, our subject and queue group.
Message Schema
Next, a simple message schema to pass around
public sealed record ExampleMessage(
int Id,
string Message);
Extensions - Message Builders
Finally, a set of helpers to build messages with distinct random Id’s and static content
public static class Extensions
{
public static ExampleMessage NextPubSubMessage(
this Random random)
=> new(random.Next(100, 199), "Hello World");
public static ExampleMessage NextRequestMessage(
this Random random)
=> new(random.Next(200, 299), "Anyone Home?");
public static ExampleMessage NextReplyMessage(
this Random random)
=> new(random.Next(300, 399), "I'm home!");
}
With the basics in place, we can implement the core NatsService
NatsService - The Background Service
Our NatsService will be a BackgroundService that becomes configured as any one of our actor variations. We’ll switch on the AppType config to drive our behavior
public sealed class NatsService(
ILogger<NatsService> logger,
INatsConnection connection,
IOptions<NatsConfig> options)
: BackgroundService
{
private readonly ILogger<NatsService> _logger = logger;
private readonly INatsConnection _connection = connection;
private readonly IOptions<NatsConfig> _options = options;
protected override Task ExecuteAsync(CancellationToken stoppingToken)
=> _options.Value.AppType switch
{
AppBuilding.Publisher => PublisherAsync(stoppingToken),
AppBuilding.Subscriber => SubscriberAsync(stoppingToken),
AppBuilding.Requestor => RequestorAsync(stoppingToken),
AppBuilding.Responder => ResponderAsync(stoppingToken),
_ => throw new NotSupportedException($"App type {_options.Value.AppType} is not supported.")
};
}
Then we can implement each behavior variation.
Publishing
private async Task PublisherAsync(CancellationToken cancelToken)
{
var (_, _, subject, _) = _options.Value;
_logger.LogInformation("Publishing to {Subject}", subject);
while (!cancelToken.IsCancellationRequested)
{
var message = Random.Shared.NextPubSubMessage();
await _connection.PublishAsync(
subject,
message,
cancellationToken: cancelToken);
await Task.Delay(TimeSpan.FromSeconds(3), cancelToken);
}
}
Subscribing
private async Task SubscriberAsync(CancellationToken cancelToken)
{
var (_, _, subject, group) = _options.Value;
_logger.LogInformation("Subscribing to {Subject}", subject);
await foreach (var msg in _connection.SubscribeAsync<ExampleMessage>(
subject,
queueGroup: group,
cancellationToken: cancelToken))
_logger.LogInformation("Pub/Sub - Received [{Message}]", msg.Data);
}
Requesting
private async Task RequestorAsync(CancellationToken cancelToken)
{
var (_, _, subject, _) = _options.Value;
_logger.LogInformation("Requesting to {Subject}", subject);
while (!cancelToken.IsCancellationRequested)
{
var message = Random.Shared.NextRequestMessage();
var reply = await _connection.RequestAsync<ExampleMessage, ExampleMessage>(
subject,
message,
cancellationToken: cancelToken);
_logger.LogInformation("Req/Rep - Reply [{Reply}]", reply.Data);
await Task.Delay(TimeSpan.FromSeconds(3), cancelToken);
}
}
Responding
private async Task ResponderAsync(CancellationToken cancelToken)
{
var (_, _, subject, _) = _options.Value;
_logger.LogInformation("Subscribing to {Subject}", subject);
await foreach (var msg in _connection.SubscribeAsync<ExampleMessage>(
subject,
cancellationToken: cancelToken))
{
_logger.LogInformation("Req/Rep - Request [{Message}]", msg.Data);
var message = Random.Shared.NextReplyMessage();
await msg.ReplyAsync(message, cancellationToken: cancelToken);
}
}
Dependency Injection
Now we’ll wire up our DI, to include our options config, the NatsConnection and NatsService
public static class AppBuilding
{
public const string Publisher = "publisher";
public const string Subscriber = "subscriber";
public const string Requestor = "requestor";
public const string Responder = "responder";
public static IServiceCollection AddNatsServices(
this IServiceCollection services,
IConfiguration config)
{
services.Configure<NatsConfig>(opts =>
{
opts.AppType = config["app-type"] ?? "publisher";
opts.NatsUrl = config["nats:url"] ?? "nats://localhost:4222";
opts.Subject = config["nats:subject"] ?? "some.subject";
opts.QueueGroup = config["nats:group"];
});
services.TryAddSingleton<INatsConnection>(sp =>
{
var config = sp.GetRequiredService<IOptions<NatsConfig>>().Value;
return new NatsConnection(new()
{
Url = config.NatsUrl,
SerializerRegistry = NatsJsonSerializerRegistry.Default
});
});
return services.AddHostedService<NatsService>();
}
}
Program: Main
And finally, of course, our generic main to kick off the Host
using Microsoft.Extensions.Hosting;
using NATSUnleashed.CSharpIntro;
var builder = Host.CreateApplicationBuilder(args);
builder.Services.AddNatsServices(builder.Configuration);
var app = builder.Build();
await app.RunAsync();
Excellent, at this stage our app is ready to build and then demo. Let’s kickoff the build quick
dotnet build -c Release
and we should see success similar to this
Restore complete (0.3s)
NATSUnleashed.CSharpIntro succeeded (0.3s) → bin\Release\net9.0\NATSUnleashed.CSharpIntro.dll
Build succeeded in 1.0s
Running NATS Server with Docker
Before we demo, we need a NATS server running. Just as we did in Getting Started with NATS CLI we’ll start up NATS server using Docker. Docker provides a convenient way to do this. If you don't have Docker installed, you'll need to install it from the official Docker website.
From a new terminal execute the following command
docker run --rm --name my-nats -p "4222:4222" -p "8222:8222" nats -n nats1 -m 8222
This command is the same as before and breaks down into a few key pieces
We expose the default NATS port 4222 and the default monitoring port 8222
Name the NATS server instance with -n nats1
Enable NATS monitoring via -m 8222
You should then see an output similar to this
[1] [INF] Starting nats-server
[1] [INF] Version: 2.11.1
[1] [INF] Git: [d78523b]
[1] [INF] Name: nats1
[1] [INF] ID: NBZG74EPNUPKYPVB7YFHOHABDAZ7EN3FFPHQYJUEAQODJYTZVNDMEH25
[1] [INF] Starting http monitor on *******:8222
[1] [INF] Listening for client connections on *******:4222
[1] [INF] Server is ready
Demo: Pub/Sub
Demo time! Open three new consoles to ./bin/Release/net9.0 and kickoff our app
Subscribers 2x
dotnet NATSUnleashed.CSharpIntro.dll --app-type subscriber
Publisher
dotnet NATSUnleashed.CSharpIntro.dll --app-type publisher
With those running, we should see messages published and delivered to each subscriber
Demo: Pub/Sub Queue Groups
Next, we’ll see Queue Groups in action. These allow you distribute messages randomly among subscribers. Stop the Publisher and both subscribers, then restart the subscribers with this new command
dotnet NATSUnleashed.CSharpIntro.dll --app-type subscriber --nats:group my-queue-group
And start the publisher again
dotnet NATSUnleashed.CSharpIntro.dll --app-type publisher
And observe the output, note each message is only delivered to a single subscriber
Demo: Request/Reply
Finally, for Request/Reply, shutdown the subscribers and publisher. Then kickoff one or more responder apps
dotnet NATSUnleashed.CSharpIntro.dll --app-type responder
Then kickoff the requestor app
dotnet NATSUnleashed.CSharpIntro.dll --app-type requestor
And the output should be similar to this, note that only one responder is received by the requestor per request
Wrap Up
Today, we built a C# dotnet console app that integrated with a NATS server via the NATS.Net client. Our app could run in either Pub/Sub or Request/Reply modes, and we demonstrated that functionality. This really lays the foundation for building some really cool stuff. With just some simple configuration and the clean API design of the client library we can get up and running quickly.
Until next time! All code is available at ConcurrentFlows GitHub
Have a specific question about NATS? Want a specific topic covered? Drop it in the comments!
