Core Publish-Subscribe (C#)

Core Publish-Subscribe in Messaging

This example demonstrates the core NATS publish-subscribe behavior. This is the fundamental pattern that all other NATS patterns and higher-level APIs build upon. There are a few takeaways from this example:

Delivery is an at-most-once. For MQTT users, this is referred to as Quality of Service (QoS) 0.
There are two circumstances when a published message won’t be delivered to a subscriber:
- The subscriber does not have an active connection to the server (i.e. the client is temporarily offline for some reason)
- There is a network interruption where the message is ultimately dropped
Messages are published to subjects which can be one or more concrete tokens, e.g. greet.bob. Subscribers can utilize wildcards to show interest on a set of matching subjects.

CLI Go Python JavaScript Rust C# Java Ruby Elixir Crystal C

Jump to the output or the recording

$ nbe run messaging/pub-sub/csharp

View the source code or learn how to run this example yourself

Code

Install NuGet package NATS.Net

using NATS.Net;

NATS_URL environment variable can be used to pass the locations of the NATS servers.

var url = Environment.GetEnvironmentVariable("NATS_URL") ?? "nats://127.0.0.1:4222";

Connect to NATS server. Since connection is disposable at the end of our scope, we should flush our buffers and close the connection cleanly.

await using var nc = new NatsClient(url);

Subscribe to a subject and start waiting for messages in the background.

Console.WriteLine("Waiting for messages...");
var cts = new CancellationTokenSource();
var subscriptionTask = Task.Run(async () =>
{
    await foreach (var msg in nc.SubscribeAsync<Order>("orders.>", cancellationToken: cts.Token))
    {
        var order = msg.Data;
        Console.WriteLine($"Subscriber received {msg.Subject}: {order}");
    }


    Console.WriteLine("Unsubscribed");
});

Wait a bit before publishing orders, so we know the subscriber is ready.

await Task.Delay(1000);

Let’s publish a few orders.

for (int i = 0; i < 5; i++)
{
    Console.WriteLine($"Publishing order {i}...");
    await nc.PublishAsync($"orders.new.{i}", new Order(OrderId: i));
    await Task.Delay(500);
}

We can unsubscribe now all orders are published. Cancelling the subscription should complete the message loop and exit the background task cleanly.

await cts.CancelAsync();
await subscriptionTask;

That’s it! We saw how we can subscribe to a subject and publish messages that would be seen by the subscribers based on matching subjects.

Console.WriteLine("Bye!");


public record Order(int OrderId);

Output

Waiting for messages...
Publishing order 0...
Subscriber received orders.new.0: Order { OrderId = 0 }
Publishing order 1...
Subscriber received orders.new.1: Order { OrderId = 1 }
Publishing order 2...
Subscriber received orders.new.2: Order { OrderId = 2 }
Publishing order 3...
Subscriber received orders.new.3: Order { OrderId = 3 }
Publishing order 4...
Subscriber received orders.new.4: Order { OrderId = 4 }
Unsubscribed
Bye!

Recording

Note, playback is half speed to make it a bit easier to follow.