Class NatsCommandQueue

Namespace

Savvyio.Extensions.NATS.Commands

Assembly

Savvyio.Extensions.NATS.dll

Provides a NATS JetStream implementation of the IPointToPointChannel<TRequest> for command messages.

public class NatsCommandQueue : NatsMessage, IDisposable, IAsyncDisposable, IHealthCheckProvider<INatsConnection>, IPointToPointChannel<ICommand>, ISender<ICommand>, IReceiver<ICommand>

Inheritance

object

Disposable

AsyncDisposable

NatsMessage

NatsCommandQueue

Implements

IDisposable

IAsyncDisposable

IHealthCheckProvider<INatsConnection>

IPointToPointChannel<ICommand>

ISender<ICommand>

IReceiver<ICommand>

Derived

NatsCommandQueue<TMarker>

Inherited Members

NatsMessage.NatsClient

NatsMessage.Marshaller

NatsMessage.OnDisposeManagedResourcesAsync()

NatsMessage.GetHealthCheckTarget()

AsyncDisposable.OnDisposeManagedResources()

AsyncDisposable.DisposeAsync()

Disposable.OnDisposeUnmanagedResources()

Disposable.Dispose()

Disposable.Dispose(bool)

Disposable.Disposed

Examples

NatsCommandQueue is the NATS JetStream command queue that publishes and receives IMessage<ICommand> envelopes. Configure it with NatsCommandQueueOptions and IMarshaller to set up the connection.

using System;
using Savvyio.Commands;
using Savvyio.Extensions.NATS.Commands;
using Savvyio.Messaging;

namespace ExampleApp;

public class NatsCommandQueueConfig
{
    public static NatsCommandQueueOptions CreateOptions()
    {
        return new NatsCommandQueueOptions
        {
            NatsUrl = new Uri("nats://localhost:4222"),
            Subject = "account-commands",
            StreamName = "savvyio-commands",
            ConsumerName = "account-command-consumer",
            AutoAcknowledge = true
        };
    }

    public static IPointToPointChannel<ICommand> AsChannel(NatsCommandQueue queue) => queue;
}

Constructors

NatsCommandQueue(IMarshaller, NatsCommandQueueOptions)

Initializes a new instance of the NatsCommandQueue class.

public NatsCommandQueue(IMarshaller marshaller, NatsCommandQueueOptions options)

Parameters

marshaller IMarshaller

The marshaller used for serializing and deserializing messages.

options NatsCommandQueueOptions

The NatsCommandQueueOptions used to configure this instance.

Exceptions

ArgumentNullException

marshaller cannot be null - or - options cannot be null.

ArgumentException

options are not in a valid state.

Methods

CreateConsumerAsync(StreamConfig, ConsumerConfig, CancellationToken)

Creates or updates the stream and consumer used by receive operations.

protected virtual Task<INatsJSConsumer> CreateConsumerAsync(StreamConfig streamConfig, ConsumerConfig consumerConfig, CancellationToken cancellationToken)

Parameters

streamConfig StreamConfig

The stream configuration.

consumerConfig ConsumerConfig

The consumer configuration.

cancellationToken CancellationToken

The cancellation token of the asynchronous operation.

Returns

Task<INatsJSConsumer>

A consumer that can fetch messages.

CreateJetStreamContext()

Creates a NATS JetStream context used by send and receive operations.

protected virtual INatsJSContext CreateJetStreamContext()

Returns

INatsJSContext

An NATS.Client.JetStream.INatsJSContext for JetStream operations.

FetchMessagesAsync(INatsJSConsumer, NatsJSFetchOpts, CancellationToken)

Fetches messages from the specified NATS JetStream consumer.

protected virtual IAsyncEnumerable<NatsCommandQueue.ReceivedNatsMessage> FetchMessagesAsync(INatsJSConsumer consumer, NatsJSFetchOpts options, CancellationToken cancellationToken)

Parameters

consumer INatsJSConsumer

The consumer to fetch messages from.

options NatsJSFetchOpts

The fetch options.

cancellationToken CancellationToken

The cancellation token of the asynchronous operation.

Returns

IAsyncEnumerable<NatsCommandQueue.ReceivedNatsMessage>

An asynchronous sequence of received NATS messages.

PublishMessageAsync(INatsJSContext, string, string, NatsHeaders)

Publishes a serialized message to NATS JetStream.

protected virtual Task PublishMessageAsync(INatsJSContext context, string subject, string message, NatsHeaders headers)

Parameters

context INatsJSContext

The JetStream context to use for publishing.

subject string

The subject to publish to.

message string

The serialized message payload.

headers NatsHeaders

The message headers.

Returns

Task

A task that represents the asynchronous operation.

ReceiveAsync(Action<AsyncOptions>)

Receives command messages asynchronously from the configured NATS JetStream consumer.

public IAsyncEnumerable<IMessage<ICommand>> ReceiveAsync(Action<AsyncOptions> setup = null)

Parameters

setup Action<AsyncOptions>

The AsyncOptions which may be configured.

Returns

IAsyncEnumerable<IMessage<ICommand>>

An IAsyncEnumerable<T> that yields IMessage<T> instances as they are received.

SendAsync(IEnumerable<IMessage<ICommand>>, Action<AsyncOptions>)

Sends the specified command messages asynchronously to the configured NATS JetStream subject.

public Task SendAsync(IEnumerable<IMessage<ICommand>> messages, Action<AsyncOptions> setup = null)

Parameters

messages IEnumerable<IMessage<ICommand>>

The messages to send.

setup Action<AsyncOptions>

The AsyncOptions which may be configured.

Returns

Task

A Task that represents the asynchronous operation.

See Also

NatsMessage

IPointToPointChannel<TRequest>