Background Service???

[scottf]

COPIED FROM SLACK:

Our org is looking to move more towards event based service integration, and I’m busy trying to spin up a proof of concept using JetStream & dotnet. The local POC is using docker-compose. Down the line, it makes sense to have dedicated services for the subscribers, but that the moment I’m wanting to simply add a stream subscription to the aspnet minimal API project as a HostedService / BackgroundService, so API calls to one of the services could publish events that other services could consume, but at the same time background workers could listen & consume messages from other subjects.
I’ve looked at the Samples, but haven’t seen anything that’s using a BackgroundService as a long running subscription, so was wondering if anyone could hep me out. I’m also quite new to dotnet and the fundamentals of BackgroundServices so that’s probably making it a bit harder for me to simply re-implement the examples from the samples directory and to get it to work.

I guess the simplest version of what I need to try work out is how can I add a long running JetStream subscription to the Worker class created when running dotnet new worker --name <Project.Name> (some microsoft docs here: https://docs.microsoft.com/en-us/dotnet/core/extensions/queue-service).

[scottf]

ANSWER FROM SLACK:

I just happened to work on a similar POC with dotnet 6. Relevant code snippets
Program.cs

using Service;

CreateHostBuilder(args).Build().Run();

static IHostBuilder CreateHostBuilder(string[] args) =>
    Host.CreateDefaultBuilder(args)
    .ConfigureServices((hostContext, services) =>
    {
        services.AddSingleton<IMessaging, NatsMessaging>();
        services.AddHostedService<Worker>();
    });

Worker.cs

namespace Service;

public class Worker : IHostedService
{
    private readonly ILogger<Worker> _logger;
    private readonly IMessaging _messaging;

    public Worker(ILogger<Worker> logger, IMessaging messaging)
    {
        _logger = logger;
        _messaging = messaging;
    }

    public Task StartAsync(CancellationToken cancellationToken)
    {
        _messaging.Subscribe(MessageHandler);
        return Task.CompletedTask;
    }

    private void MessageHandler(object? sender, string message)
    {
        try
        {
            _logger.LogDebug("Handling message: {Message}", message);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error handling message.");
        }
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        _messaging.Dispose();
        return Task.CompletedTask;
    }
}

NatsMessaging.cs

using NATS.Client;
using NATS.Client.JetStream;
using System.Text;

namespace Service;

public class NatsMessaging : IMessaging
{
    private readonly ILogger<NatsMessaging> _logger;
    private IConnection? _connection;
    private bool _disposed;

    public NatsMessaging(ILogger<NatsMessaging> logger)
    {
        _logger = logger;
    }

    public void Subscribe(EventHandler<string> messageHandler)
    {
        var subject = "my-subject";
        var service = "my-service";
        EnsureConnection();
        EnsureStream("my-stream", subject);

        var consumerConf = ConsumerConfiguration.Builder()
            .WithDurable(service)
            .Build();
        var subscribeOptions = PushSubscribeOptions.Builder()
            .WithConfiguration(consumerConf)
            .Build();

        // This will create the stream consumer in NATS if it does not already exist.
        var subscription = _connection!.CreateJetStreamContext().PushSubscribeAsync(
            subject,
            service,
            (sender, args) =>
            {
                messageHandler(sender, Encoding.UTF8.GetString(args.Message.Data));
            },
            true,
            subscribeOptions);
        _logger.LogDebug("Created NATS subscription {Subscription}", subscription);
    }

    private void EnsureConnection()
    {
        if (_connection is not null) {
            return;
        }

        _connection = new ConnectionFactory().CreateConnection();
        _logger.LogDebug("Created NATS connection: {Connection}", _connection);
    }

    private void EnsureStream(string name, string subject)
    {
        if (_connection is null)
        {
            EnsureConnection();
        }

        var streamConfig = StreamConfiguration.Builder()
            .WithName(name)
            .WithSubjects(subject)
            .WithStorageType(StorageType.File)
            .WithReplicas(3)
            .WithRetentionPolicy(RetentionPolicy.WorkQueue)
            .Build();

        var js = _connection!.CreateJetStreamManagementContext();
        var stream = js.GetStreamNames().Contains(name)
            ? js.UpdateStream(streamConfig)
            : js.AddStream(streamConfig);
        _logger.LogDebug("Using NATS stream: {Stream}", stream);
    }

    public void Dispose()
    {
        if (_disposed)
        {
            return;
        }
        _disposed = true;

        /* Connection.Drain() will:
         * - Drain() all subscriptions
         * - Flush() the connection
         * - Close() all subscriptions
         * - Close() the connection
         */
        _connection?.Drain();
        _connection?.Dispose();
    }
}

public interface IMessaging : IDisposable
{
    void Subscribe(EventHandler<string> messageHandler);
}

[Exagram]

The above example describes a push-based consumer. For pull-based consumer I would recommend replacing Subscribe(EventHandler<string> ...) with IMessage Subscribe() in NatsMessaging.cs and a while-loop in Worker.cs. I'm looking for feedback on NATS best practices.

Worker.cs:

namespace Service;

public class Worker : IHostedService
{
    private readonly ILogger<Worker> _logger;
    private readonly IMessaging _messaging;

    public Worker(ILogger<Worker> logger, IMessaging messaging)
    {
        _logger = logger;
        _messaging = messaging;
    }

    public async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            var msgs = _messaging.Subscribe();
            _logger.LogWarning("Add logic to handle the messages: " + msgs);
        }
    }

    public Task StopAsync(CancellationToken cancellationToken)
    {
        _messaging.Dispose();
        return Task.CompletedTask;
    }
}

NatsMessaging.cs:

public class NatsMessaging : IMessaging
{
    private readonly ILogger<NatsMessaging> _logger;
    private IConnection? _connection;
    private bool _disposed;

    public NatsMessaging(ILogger<NatsMessaging> logger)
    {
        _logger = logger;
    }

    public IMessage Subscribe()
    {
        var subject = "my-subject";
        var service = "my-service";
        EnsureConnection();
        EnsureStream("my-stream", subject);

        var consumerConf = ConsumerConfiguration.Builder()
            .WithDurable(service)
            .Build();
        var subscribeOptions = PullSubscribeOptions.Builder()
            .WithConfiguration(consumerConf)
            .Build();

        // This will create the stream consumer in NATS if it does not already exist.
        var subscription = _connection!.CreateJetStreamContext().PullSubscribe(
            subject,
            subscribeOptions);
        var msgs = subscription.Fetch(1, 1000); // TODO: Sane?
        if (msgs.Count() != 1)
        {
            return null;
        }
        var msg = msgs[0];
        return new NatsMessage { Subject = msg.Subject, Body = msg.Body };
    }
}

public interface IMessaging : IDisposable
{
    IMessage Subscribe();
}

public interface IMessage
{
    string Subject { get; set; }
    string Body { get; set; }
}

public class NatsMessage : IMessage
{
    public string Subject { get; set; }
    public string Body { get; set; }
}