Consumer is not receiving messages

[Warrolen]

I'm building a message bus library but unfortunately I cannot get the consumer to consume the produced messages. Nothing really happens for some reason and I'm not even sure how to debug what's happening. It doesn't log any messages or so.

What's wrong with it? Why is the consumer not receiving the messages?

Producer/Program.cs

using Contracts;
using NATS.Client;
using Sts.MessageBus.Streaming.JetStreams;

var options = ConnectionFactory.GetDefaultOptions();
var connectionFactory = new ConnectionFactory();
var connection = connectionFactory.CreateConnection(options);

var producer = new JetStreamProducer(connection);
await producer.PublishAsync("SUBJECT", "A", new Test { Title = "Hello from title", Message = "Hello from message" });

Consumer/Program.cs

using NATS.Client;
using Sts.MessageBus.Streaming.JetStreams;

var options = ConnectionFactory.GetDefaultOptions();
var connectionFactory = new ConnectionFactory();
var connection = connectionFactory.CreateConnection(options);

var consumer = new JetStreamConsumer(connection);

var reader = consumer.Subscribe("SUBJECT.A");

await foreach (var item in reader.ReadAllAsync())
{
    Console.WriteLine($"Message received: {item.Message.Value}");
}

Console.ReadLine();

Code

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

namespace Sts.MessageBus.Streaming.JetStreams;

public sealed class JetStreamProducer
{
    private readonly IConnection _connection;
    private readonly Lazy<IJetStream> _jetStreamFactory;
    private readonly Lazy<IJetStreamManagement> _jetStreamManagementFactory;

    public JetStreamProducer(IConnection connection)
    {
        _connection = connection;
        _jetStreamFactory = new Lazy<IJetStream>(() => connection.CreateJetStreamContext());
        _jetStreamManagementFactory = new Lazy<IJetStreamManagement>(() => connection.CreateJetStreamManagementContext());
    }

    private IJetStream JetStream => _jetStreamFactory.Value;

    private IJetStreamManagement JetStreamManagement => _jetStreamManagementFactory.Value;

    public async Task PublishAsync<T>(string subject, string exchange, T message, CancellationToken cancellationToken = default)
    {
        JetStreamUtils.CreateStreamOrUpdateSubjects(_connection, $"{subject}-stream", $"{subject}.>");

        var payload = JsonSerializer.SerializeToUtf8Bytes(message);
        var msg = new Msg($"{subject}.{exchange}", null, null, payload);
        await JetStream.PublishAsync(msg);
    }
}

using System.Text;
using System.Text.Json;
using System.Threading.Channels;
using NATS.Client;
using NATS.Client.JetStream;
using Sts.MessageBus.Streaming.Abstractions;

namespace Sts.MessageBus.Streaming.JetStreams;

public sealed class JetStreamConsumer
{
    private const int PollBatchSize = 1;
    private const int PollWaitInterval = 1000;

    private readonly Lazy<IJetStream> _jetStreamFactory;
    private readonly Lazy<IJetStreamManagement> _jetStreamManagementFactory;

    public JetStreamConsumer(IConnection connection)
    {
        _jetStreamFactory = new Lazy<IJetStream>(() => connection.CreateJetStreamContext());
        _jetStreamManagementFactory = new Lazy<IJetStreamManagement>(() => connection.CreateJetStreamManagementContext());
    }

    private IJetStream JetStream => _jetStreamFactory.Value;

    private IJetStreamManagement JetStreamManagement => _jetStreamManagementFactory.Value;

    public ChannelReader<ConsumeResult> Subscribe(string subject, CancellationToken cancellationToken = default)
    {
        var channel = Channel.CreateUnbounded<ConsumeResult>();

        _ = StreamReaderAsync(channel, subject, cancellationToken);

        return channel.Reader;
    }

    private async Task StreamReaderAsync(ChannelWriter<ConsumeResult> writer, string subject, CancellationToken cancellationToken)
    {
        CreateQueueIfNotExists(subject);

        var bindTo = PullSubscribeOptions.BindTo($"{subject}-stream", $"{subject}-durable");
        var subscription = JetStream.PullSubscribe($"{subject}.>", bindTo);


        ConsumerInfo afterInfo = subscription.GetConsumerInformation();
        Console.WriteLine(
            "\nConsumerInfo{" +
            "Name='" + afterInfo.Name + '\'' +
            ", NumPending=" + afterInfo.NumPending +
            ", NumWaiting=" + afterInfo.NumWaiting +
            ", Delivered=" + afterInfo.Delivered +
            ", AckFloor=" + afterInfo.AckFloor +
            "}\n\n");

        while (!cancellationToken.IsCancellationRequested)
        {
            foreach (var message in subscription.Fetch(PollBatchSize, PollWaitInterval))
            {
                var msg = message.Data.ToString();

                await writer.WriteAsync(new ConsumeResult(subject, message.Subject, msg), cancellationToken).ConfigureAwait(false);
                message.Ack();
            }
        }
    }

    private void CreateQueueIfNotExists(string subject)
    {
        var streamName = $"{subject}-stream";

        var streams = JetStreamManagement.GetStreamNames();
        if (streams.Contains(streamName))
        {
            return;
        }

        var streamConfiguration = StreamConfiguration
            .Builder()
            .WithName(streamName)
            .WithStorageType(StorageType.Memory)
            .WithSubjects($"{subject}-subject")
            .Build();
        JetStreamManagement.AddStream(streamConfiguration);

        var consumerConfiguration = ConsumerConfiguration.Builder()
            .WithDurable($"{subject}-durable")
            .WithDeliverSubject($"{subject}-deliver")
            .WithDeliverGroup($"{subject}-queue")
            .Build();
        JetStreamManagement.AddOrUpdateConsumer(streamName, consumerConfiguration);
    }
}