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:

1. Stand up a C# console project with our NuGet dependencies

2. Integrate with the NATS server via the C# NATS client

3. 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

1. We expose the default NATS port 4222 and the default monitoring port 8222

2. Name the NATS server instance with -n nats1

3. 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!