single active consumer is not consuming messages

[blendereru]

Hi there! Got an issue when enabling a single active consumer in a stream consumers.
Publisher:

using System.Text;
using RabbitMQ.Stream.Client;
using RabbitMQ.Stream.Client.Reliable;

var streamSystem = await StreamSystem.Create(new StreamSystemConfig()).ConfigureAwait(false);
await streamSystem.CreateStream(new StreamSpec("my-sac-stream")).ConfigureAwait(false);
var producer = await Producer.Create(new ProducerConfig(streamSystem, "my-sac-stream")).ConfigureAwait(false);
//var messages = new List<Message>(); enable when sending in batch
for (int i = 0; i < 50; i++)
{
    var body = Encoding.UTF8.GetBytes($"Message #{i}");
    var message = new Message(body);
    await producer.Send(message).ConfigureAwait(false);
}
Console.WriteLine("Sending 50 messages to my-sac-stream");
await producer.Close().ConfigureAwait(false);
await streamSystem.Close().ConfigureAwait(false);

Consumers:

using System.Buffers;
using System.Text;
using RabbitMQ.Stream.Client;
using RabbitMQ.Stream.Client.Reliable;

var streamSystem = await StreamSystem.Create(new StreamSystemConfig()).ConfigureAwait(false);
var consumer = await Consumer.Create(new ConsumerConfig(streamSystem, "my-sac-stream")
{
    Reference = "sac_consumer",
    OffsetSpec = new OffsetTypeFirst(),
    IsSingleActiveConsumer = true,
    MessageHandler = async (_, consumer, context, message) =>
    {
        var text = Encoding.UTF8.GetString(message.Data.Contents.ToArray());
        Console.WriteLine($"The message {text} was received");
        await Task.CompletedTask.ConfigureAwait(false);
    },
    ConsumerUpdateListener = async (consumerRef, stream, isActive) =>
    {
        var offset = await streamSystem.QueryOffset(consumerRef, stream).ConfigureAwait(false);
        return new OffsetTypeOffset(offset);
    }
}).ConfigureAwait(false);
Console.WriteLine("Consumer is running. Press [enter] to exit.");
Console.ReadLine();
await consumer.Close().ConfigureAwait(false);
await streamSystem.Close().ConfigureAwait(false);

when i launch my producer, everything is ok:

Sending 50 messages to my-sac-stream

when i launch my consumers, the single active consumer works fine, but the consumption process doesn't:

Consumer is running. Press [enter] to exit.

Console doesn't print eveything else. The message above occurs in every consumer.

Did i forget something in my code ? Thanks in advance

[Gsantomaggio]

@blendereru the API streamSystem.QueryOffset could raise an exception if the offset does not exist.
( maybe it is time to implement #370 )

You need to store and offset before querying it. See await consumer.StoreOffset(context.Offset).ConfigureAwait(false);

Here is a working example:

 var streamSystem = await StreamSystem.Create(new StreamSystemConfig()).ConfigureAwait(false);
        var consumer = await Consumer.Create(new ConsumerConfig(streamSystem, "my-sac-stream")
        {
            Reference = "sac_consumer",
            OffsetSpec = new OffsetTypeFirst(),
            IsSingleActiveConsumer = true,
            MessageHandler = async (_, consumer, context, message) =>
            {
                var text = Encoding.UTF8.GetString(message.Data.Contents.ToArray());
                Console.WriteLine($"The message {text} was received");
                
                // Store the offset of the message.
                // store offset for each message is not a good practice
                //This is only for demo purpose
                await consumer.StoreOffset(context.Offset).ConfigureAwait(false);
                
                await Task.CompletedTask.ConfigureAwait(false);
            },
            ConsumerUpdateListener = async (consumerRef, stream, isActive) =>
            {
                try
                {
                    var offset = await streamSystem.QueryOffset(consumerRef, stream).ConfigureAwait(false);
                    return new OffsetTypeOffset(offset);
                }
                catch (OffsetNotFoundException)
                {
                    Console.WriteLine(
                        $"Offset not found for stream {stream} and consumer {consumerRef}. Will use the first offset");
                    return new OffsetTypeFirst();
                }
            }
        }).ConfigureAwait(false);
        Console.WriteLine("Consumer is running. Press [enter] to exit.");
        Console.ReadLine();
        await consumer.Close().ConfigureAwait(false);
        await streamSystem.Close().ConfigureAwait(false);

[blendereru]

Thanks @Gsantomaggio, it worked! Is this because consumers starts consuming at the very end of a stream by default (next offset) ?

[Gsantomaggio]

@blendereru The problem is that the queryOffset raises an exception, and the comuser is inconsistent.

In this PR I added a log so it is more clear.

I also added a new API, TryQueryOffset, that avoids raising an exception when the offset does not exit. See the example

Thank you for raising the issue.

[blendereru]

Thanks!