cloops.nats 1.4.13

CLOOPS NATS SDK

A modern, annotation-based SDK for building reliable distributed systems with NATS. Define your message consumers with simple attributes and let the framework handle the complexity.

Why NATS?

NATS is a powerful messaging system that enables you to build sophisticated, fault-tolerant distributed systems that are location-transparent and globally distributed.

• Faster than HTTP - Binary protocol with lower overhead
• Fewer hops - Direct communication without load balancers or API gateways
• Globally distributed - Deploy applications worldwide without DNS or complex load balancing. Learn more about NATS super clusters

💡 Learn more: Check out this podcast from nats.fm for insights into NATS architecture.

Why This SDK?

Modern developers expect annotation-based definitions and dependency injection—the same developer experience you get with REST frameworks like ASP.NET Core. This SDK brings that same simplicity to NATS, so you can focus on your business logic instead of boilerplate.

What You Get

• ✨ Annotation-based consumers - Define subscribers with simple attributes
• 🛡️ Built-in backpressure - Automatic handling of traffic spikes
• ⚖️ Flexible load balancing - Choose between broadcasting or load balancing strategies
• 🚀 JetStream support - Build temporally decoupled systems with persistent messaging
• 🔧 Dependency injection - Seamless integration with .NET's DI container
• ✅ Automatic message validation - Messages with a Validate() method are automatically validated before processing. Invalid messages are never sent to your handlers
• 🪝 Consumer interceptors - Inspect, replace, or reject messages before handlers run
• 🧯 Consumer exception handlers - Map handler exceptions to a NatsAck for consistent JetStream ack/nak and Core request/reply responses, while still logging the original exception and recording the invocation as a failure

🎯 Building microservices? Check out our microservices-focused SDK built on top of cloops.nats and makes building microservices a breeze!

Quick Start

Installation

Add the cloops.nats package to your .csproj file:

<PackageReference Include="cloops.nats" Version="*" />

Run dotnet restore to install the package.

💡 Tip: For Connection Loops standard messages and subjects, you may also need cloops.nats.schema.

Examples

Broadcast Pattern (Kubernetes/Docker)

Ensure each pod/instance receives all messages by using runtime placeholders in the queue group name:

/// <summary>
/// Broadcast: Each pod gets a unique queue group, so all pods receive all messages
/// Supported placeholders: {POD_NAME}, {HOSTNAME}, {MACHINE_NAME}, {ENV:VAR_NAME}
/// </summary>
[NatsConsumer("test.broadcast", QueueGroupName = "pod-{POD_NAME}")]
public Task<NatsAck> BroadcastHandler(NatsMsg<string> msg, CancellationToken ct = default)
{
    Console.WriteLine($"[Pod {Environment.GetEnvironmentVariable("POD_NAME")}] Received: {msg.Data}");
    return Task.FromResult(new NatsAck(true));
}

Load Balancing Pattern

Distribute messages across multiple instances using the same queue group:

[NatsConsumer("test.lb", QueueGroupName = "workers")]
public async Task<NatsAck> HandleMessage(NatsMsg<string> msg, CancellationToken ct = default)
{
    Console.WriteLine($"Instance received: {msg.Data}");
    await Task.Delay(100, ct).ConfigureAwait(false); // Simulate work
    return new NatsAck(true);
}

Runtime Placeholders

The SDK resolves placeholders dynamically:

• {POD_NAME} → POD_NAME env var, falls back to HOSTNAME or machine name
• {HOSTNAME} → HOSTNAME env var, falls back to machine name
• {MACHINE_NAME} → Machine name
• {ENV:VAR_NAME} → Any environment variable (e.g., {ENV:MY_CUSTOM_VAR})

📝 Note: QueueGroupName is optional. If omitted, an empty string is used, which still enables load balancing. JetStream subscriptions are always load-balanced (no broadcast support).

Timeouts and logging

CloopsNatsClient runs with NATS.Net defaults unless you pass a CloopsNatsClientOptions. Two of those defaults bite production services:

• RequestTimeout is 5 seconds and also bounds the wait for a JetStream publish acknowledgement. A service that publishes to JetStream from a request path and is briefly starved (thread-pool, GC) can miss that window even though the server replied, and gets NatsJSPublishNoResponseException.
• LoggerFactory is a NullLoggerFactory, so NATS.Net's own connect, reconnect, ping and retry diagnostics are discarded.

Always pass a logger factory, and raise RequestTimeout to 15-30 seconds for services that publish to JetStream from request paths:

var cnc = new CloopsNatsClient(
    url: "nats://dev.nats.cloops.in:4222",
    name: "my-service",
    creds: natsCreds,
    options: new CloopsNatsClientOptions
    {
        RequestTimeout = TimeSpan.FromSeconds(30),   // JetStream publish ack wait, request/reply
        LoggerFactory = loggerFactory,               // surfaces NATS.Net's internal diagnostics
        // CommandTimeout, ConnectTimeout are available too; anything else via the escape hatch:
        Configure = o => o with { MaxReconnectRetry = -1 },
    });

logger.LogInformation("NATS RequestTimeout={Timeout}", cnc.Connection.Opts.RequestTimeout);

Every property is optional and an unset property keeps the NATS.Net default, so existing callers using the three argument constructor are unchanged. With a LoggerFactory set (directly or through Configure) the client also logs the connection lifecycle events (ConnectionOpened, ConnectionDisconnected, ReconnectFailed, MessageDropped, LameDuckModeActivated) under the CLOOPS.NATS.CloopsNatsClient category; set LogConnectionEvents = false if you subscribe to them yourself. The client sets SubPendingChannelFullMode = Wait, so a full subscription channel applies backpressure instead of dropping messages and MessageDropped will rarely if ever fire under normal operation. The effective options are exposed as cnc.Options.

Learn More

• 📖 Examples - See real-world usage patterns
• 📚 Full Documentation - Detailed guides, setup instructions, and API reference

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

This project is licensed under the MIT License - see the LICENSE file for details.