[SUB-90011] Subject does not match consumer configuration filter.

[Hulkstance]

I know this has already been asked but I'm misunderstanding something.

I'm publishing messages to PERSISTENCE.binance, PERSISTENCE.kraken, etc. and subscribing to PERSISTENCE.*, but for some reason I'm getting the following error:

[SUB-90011] Subject does not match consumer configuration filter.

I saw this example but it uses the pull-based consumer.

Snippet

private const string Stream = "mkt-data";

await _natsPublisher.PublishAsync(stream, "PERSISTENCE.binance", request);

private const string Stream = "mkt-data";
private const string Subject = "PERSISTENCE.*";
private const string Durable = "bind-durable";
private const string DeliverSubject = "bind-deliver";

_ = _natsSubscriber.Subscribe(stream, Subject, Durable, DeliverSubject, msg =>
{
     var json = Encoding.UTF8.GetString(msg.Data);
    _logger.LogInformation("Message arrived: {Message}", json);
});

public sealed class NatsPublisher : INatsPublisher
{
    private readonly IConnection _connection;
    private readonly Lazy<IJetStream> _jetStreamFactory;

    public NatsPublisher(IConnection connection)
    {
        _connection = connection;
        _jetStreamFactory = new Lazy<IJetStream>(() => connection.CreateJetStreamContext());
    }

    private IJetStream JetStream => _jetStreamFactory.Value;

    public async ValueTask<PublishAck> PublishAsync<T>(
        string stream,
        string subject,
        T message)
        where T : class
    {
        JetStreamUtils.CreateStreamOrUpdateSubjects(_connection, stream, subject);

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

    public async ValueTask<PublishAck> PublishAsync<T>(
        string stream,
        string subject,
        T message,
        IEnumerable<(string, string)> headers)
        where T : class
    {
        JetStreamUtils.CreateStreamOrUpdateSubjects(_connection, stream, subject);

        var payload = JsonSerializer.SerializeToUtf8Bytes(message);
        var msg = new Msg(subject, null, null, payload);

        foreach (var (header, val) in headers)
        {
            msg.Header[header] = val;
        }

        return await JetStream.PublishAsync(msg);
    }

    public async ValueTask<PublishAck> PublishDeduplicationIdAsync<T>(
        string stream,
        string subject,
        T message,
        string id)
        where T : class
    {
        JetStreamUtils.CreateStreamOrUpdateSubjects(_connection, stream, subject);

        var payload = JsonSerializer.SerializeToUtf8Bytes(message);
        var msg = new Msg(subject, null, null, payload)
        {
            Header = { ["Nats-Msg-Id"] = id }
        };

        return await JetStream.PublishAsync(msg);
    }
}

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

    public NatsSubscriber(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 IJetStreamSubscription Subscribe(
        string stream,
        string subject,
        string durable,
        string deliverSubject,
        Action<Msg> onMessageArrived)
    {
        JetStreamUtils.CreateStreamWhenDoesNotExist(_connection, stream, subject);

        var consumerConfiguration = ConsumerConfiguration.Builder()
            .WithDurable(durable)
            .WithDeliverSubject(deliverSubject)
            .Build();
        JetStreamManagement.AddOrUpdateConsumer(stream, consumerConfiguration);

        var so = PushSubscribeOptions.BindTo(stream, durable);
        var subscription = JetStream.PushSubscribeAsync(
            subject,
            (_, e) => onMessageArrived(e.Message),
            true,
            so);

        return subscription;
    }
}

public static class JetStreamUtils
{
    // ----------------------------------------------------------------------------------------------------
    // STREAM INFO / CREATE / UPDATE
    // ----------------------------------------------------------------------------------------------------
    public static StreamInfo? GetStreamInfoOrNullWhenNotExist(IJetStreamManagement jsm, string streamName)
    {
        try
        {
            return jsm.GetStreamInfo(streamName);
        }
        catch (NATSJetStreamException e)
        {
            if (e.ErrorCode == 404)
            {
                return null;
            }

            throw;
        }
    }

    public static bool StreamExists(IConnection c, string streamName)
    {
        return GetStreamInfoOrNullWhenNotExist(c.CreateJetStreamManagementContext(), streamName) != null;
    }

    public static bool StreamExists(IJetStreamManagement jsm, string streamName)
    {
        return GetStreamInfoOrNullWhenNotExist(jsm, streamName) != null;
    }

    public static void ExitIfStreamExists(IJetStreamManagement jsm, string streamName)
    {
        if (StreamExists(jsm, streamName))
        {
            Environment.Exit(-1);
        }
    }

    public static void ExitIfStreamNotExists(IConnection c, string streamName)
    {
        if (!StreamExists(c, streamName))
        {
            Environment.Exit(-1);
        }
    }

    public static StreamInfo CreateStream(IJetStreamManagement jsm, string streamName, StorageType storageType,
        params string[] subjects)
    {
        var sc = StreamConfiguration.Builder()
            .WithName(streamName)
            .WithStorageType(storageType)
            .WithSubjects(subjects)
            .Build();

        var si = jsm.AddStream(sc);

        return si;
    }

    public static StreamInfo CreateStream(IJetStreamManagement jsm, string stream, params string[] subjects)
    {
        return CreateStream(jsm, stream, StorageType.Memory, subjects);
    }

    public static StreamInfo CreateStream(IConnection c, string stream, params string[] subjects)
    {
        return CreateStream(c.CreateJetStreamManagementContext(), stream, StorageType.Memory, subjects);
    }

    public static StreamInfo CreateStreamExitWhenExists(IConnection c, string streamName, params string[] subjects)
    {
        return CreateStreamExitWhenExists(c.CreateJetStreamManagementContext(), streamName, subjects);
    }

    public static StreamInfo CreateStreamExitWhenExists(IJetStreamManagement jsm, string streamName,
        params string[] subjects)
    {
        ExitIfStreamExists(jsm, streamName);
        return CreateStream(jsm, streamName, StorageType.Memory, subjects);
    }

    public static void CreateStreamWhenDoesNotExist(IJetStreamManagement jsm, string stream, params string[] subjects)
    {
        try
        {
            jsm.GetStreamInfo(stream);
            return;
        }
        catch (NATSJetStreamException)
        {
        }

        var sc = StreamConfiguration.Builder()
            .WithName(stream)
            .WithStorageType(StorageType.Memory)
            .WithSubjects(subjects)
            .Build();
        jsm.AddStream(sc);
    }

    public static void CreateStreamWhenDoesNotExist(IConnection c, string stream, params string[] subjects)
    {
        CreateStreamWhenDoesNotExist(c.CreateJetStreamManagementContext(), stream, subjects);
    }

    public static StreamInfo CreateStreamOrUpdateSubjects(IJetStreamManagement jsm, string streamName,
        StorageType storageType, params string[] subjects)
    {
        var si = GetStreamInfoOrNullWhenNotExist(jsm, streamName);
        if (si == null)
        {
            return CreateStream(jsm, streamName, storageType, subjects);
        }

        var sc = si.Config;
        var needToUpdate = false;
        foreach (var sub in subjects)
        {
            if (!sc.Subjects.Contains(sub))
            {
                needToUpdate = true;
                sc.Subjects.Add(sub);
            }
        }

        if (needToUpdate)
        {
            si = jsm.UpdateStream(sc);
        }

        return si;
    }

    public static StreamInfo CreateStreamOrUpdateSubjects(IJetStreamManagement jsm, string streamName,
        params string[] subjects)
    {
        return CreateStreamOrUpdateSubjects(jsm, streamName, StorageType.Memory, subjects);
    }

    public static StreamInfo CreateStreamOrUpdateSubjects(IConnection c, string stream, params string[] subjects)
    {
        return CreateStreamOrUpdateSubjects(c.CreateJetStreamManagementContext(), stream, StorageType.Memory,
            subjects);
    }

    // ----------------------------------------------------------------------------------------------------
    // PUBLISH
    // ----------------------------------------------------------------------------------------------------
    public static void Publish(IConnection c, string subject, int count)
    {
        Publish(c.CreateJetStreamContext(), subject, "data", count);
    }

    public static void Publish(IJetStream js, string subject, int count)
    {
        Publish(js, subject, "data", count);
    }

    public static void Publish(IJetStream js, string subject, string prefix, int count)
    {
        for (var x = 1; x <= count; x++)
        {
            var data = prefix + x;
            js.Publish(subject, Encoding.UTF8.GetBytes(data));
        }
    }

    public static void PublishInBackground(IJetStream js, string subject, string prefix, int count)
    {
        new Thread(() =>
        {
            try
            {
                for (var x = 1; x <= count; x++)
                {
                    js.Publish(subject, Encoding.ASCII.GetBytes(prefix + "-" + x));
                }
            }
            catch (Exception)
            {
                Environment.Exit(-1);
            }
        }).Start();
        Thread.Sleep(100); // give the publish thread a little time to get going
    }

    // ----------------------------------------------------------------------------------------------------
    // READ MESSAGES
    // ----------------------------------------------------------------------------------------------------
    public static IList<Msg> ReadMessagesAck(ISyncSubscription sub, int timeout = 1000)
    {
        IList<Msg> messages = new List<Msg>();
        var keepGoing = true;
        while (keepGoing)
        {
            try
            {
                var msg = sub.NextMessage(timeout);
                messages.Add(msg);
                msg.Ack();
            }
            catch (NATSTimeoutException)
            {
                keepGoing = false;
            }
        }

        return messages;
    }
}

[scottf]

1. When you create the consumer here, the consumer filter subject will be blank which is equivalent to > . This does not match the subscribe subject of PERSISTENCE.*. Add WithFilterSubject(subject) to your consumer builder.

        var consumerConfiguration = ConsumerConfiguration.Builder()
            .WithDurable(durable)
            .WithDeliverSubject(deliverSubject)
            .Build();
        JetStreamManagement.AddOrUpdateConsumer(stream, consumerConfiguration);

        var so = PushSubscribeOptions.BindTo(stream, durable);
        var subscription = JetStream.PushSubscribeAsync(
            subject,
            (_, e) => onMessageArrived(e.Message),
            true,
            so);

2. Not sure why you are doing this as part of subscription. Usually the stream is created well head of time and this would never be done here. Maybe before the publish, but usually just ahead of time.

JetStreamUtils.CreateStreamWhenDoesNotExist(_connection, stream, subject);

3. I would get out of the habit of creating or updating your consumers as part of the subscription also. If they are durable, the preferred method would be to create them before hand. If you still want to do it here, I would suggest looking up the consumer to see if it exists, then creating it if only it does not already exist.

[Hulkstance]

Thanks! I actually realized that I needed a pull-based consumer as it's more suitable for my use case.

What do you suggest for 2) and 3)? Moving JetStreamUtils.CreateStreamWhenDoesNotExist and JetStreamManagement.AddOrUpdateConsumer to the constructor feels wrong. Does creating a singleton service specially for that reason sounds better?

Below is the updated code, but I'm still not sure what I should do with the stream creation ahead of time.

public sealed class NatsJetStreamSubscriber : INatsSubscriber
{
    private const int PollBatchSize = 1;
    private const int PollWaitMills = 1000;

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

    public NatsJetStreamSubscriber(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 IJetStreamSubscription Subscribe(
        string stream,
        string subject,
        string consumer,
        Func<byte[], ValueTask> eventProcessor,
        CancellationToken cancellationToken = default)
    {
        JetStreamUtils.CreateStreamWhenDoesNotExist(_connection, stream, $"{subject}.>");

        JetStreamManagement.AddOrUpdateConsumer(stream, ConsumerConfiguration.Builder()
            .WithDurable(consumer)
            .Build()
        );

        var pullOptions = PullSubscribeOptions.BindTo(stream, consumer);
        var sub = JetStream.PullSubscribe($"{subject}.>", pullOptions);

        _ = Task.Run(async () =>
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                foreach (var message in sub.Fetch(PollBatchSize, PollWaitMills))
                {
                    await eventProcessor(message.Data);
                    message.Ack();
                }
            }
        }, cancellationToken);

        return sub;
    }
}

public sealed class NatsJetStreamPublisher : INatsPublisher
{
    private readonly IConnection _connection;
    private readonly Lazy<IJetStream> _jetStreamFactory;

    public NatsJetStreamPublisher(IConnection connection)
    {
        _connection = connection;
        _jetStreamFactory = new Lazy<IJetStream>(() => connection.CreateJetStreamContext());
    }

    private IJetStream JetStream => _jetStreamFactory.Value;

    public async ValueTask<PublishAck> PublishAsync<T>(string stream, string subject, string exchange, T message)
        where T : class
    {
        JetStreamUtils.CreateStreamOrUpdateSubjects(_connection, stream, $"{subject}.>");

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

    public async ValueTask<PublishAck> PublishAsync<T>(string stream, string subject, string exchange, T message,
        IEnumerable<(string, string)> headers)
        where T : class
    {
        JetStreamUtils.CreateStreamOrUpdateSubjects(_connection, stream, $"{subject}.>");

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

        foreach (var (header, val) in headers)
        {
            msg.Header[header] = val;
        }

        return await JetStream.PublishAsync(msg);
    }

    public async ValueTask<PublishAck> PublishDeduplicationIdAsync<T>(string stream, string subject, string exchange,
        T message, string id)
        where T : class
    {
        JetStreamUtils.CreateStreamOrUpdateSubjects(_connection, stream, $"{subject}.>");

        var payload = JsonSerializer.SerializeToUtf8Bytes(message);
        var msg = new Msg($"{subject}.{exchange}", null, null, payload)
        {
            Header = { ["Nats-Msg-Id"] = id }
        };

        return await JetStream.PublishAsync(msg);
    }
}