Manually closing consumer does not close it

[daborg69]

When you call the consumer.Close() method, it results in the stream temporarily closing the connection, but then within a second it is re-opened and starts consuming messages again. The only way to close the consumer and stop the consumption of messages is to actually dispose of it.
await consumer.Close(); consumer.Dispose();

Is this by design that there is no way to manually pause the consumption of messages without disposing of the consumer?

Using your basic sample application: With the code slightly modified to display the actual message this is what happens when you call Close() only after receiving the 10th message.
`
var consumerCount = 0;

        // Create a consumer
        // tag::sample-consumer[]
        System.Console.WriteLine("Starting consuming...");
        var consumer = await Consumer.Create( // <1>
                                             new ConsumerConfig(streamSystem, StreamName)
                                             {
                                                 OffsetSpec = new OffsetTypeFirst(),                                         // <2>
                                                 MessageHandler = async (sourceStream, consumer, messageContext, message) => // <3>
                                                 {
                                                     string val = Encoding.ASCII.GetString(message.Data.Contents);
                                                     System.Console.WriteLine($"Message: {consumerCount}  |  " + val);

                                                     if (Interlocked.Increment(ref consumerCount) == MessageCount)
                                                     {
                                                         System.Console.WriteLine("*********************************");
                                                         System.Console.WriteLine($"All the {MessageCount} messages are received");
                                                         System.Console.WriteLine("*********************************");
                                                         consumerTaskCompletionSource.SetResult(MessageCount);
                                                     }

                                                     if (consumerCount == 10)
                                                         await consumer.Close().ConfigureAwait(false);
                                                     await Task.CompletedTask.ConfigureAwait(false);
                                                 }
                                             },
                                             consumerLogger // <4>
                                            )
                                     .ConfigureAwait(false);
        consumerTaskCompletionSource.Task.Wait();     // <5>
        await consumer.Close().ConfigureAwait(false); // <6>

`

Produces output like this:

Message: 0 | 0
Message: 1 | 1
Message: 2 | 2
Message: 3 | 3
Message: 4 | 4
Message: 5 | 5
Message: 6 | 6
Message: 7 | 7
Message: 8 | 8
Message: 9 | 9
warn: RabbitMQ.Stream.Client.Reliable.Consumer[0]
OperationCanceledException. The consumer id: 0 reference: (null) has been closed while consuming messages
warn: RabbitMQ.Stream.Client.Reliable.Consumer[0]
Received credit response for unknown subscription 0, ResponseCode SubscriptionIdDoesNotExist
warn: RabbitMQ.Stream.Client.Reliable.Consumer[0]
Received credit response for unknown subscription 0, ResponseCode SubscriptionIdDoesNotExist
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
Consumer reference: , stream: Sample.Z disconnected, check if reconnection needed in 200 ms
info: RabbitMQ.Stream.Client.Reliable.Producer[0]
Producer reference: , stream: Sample.Z is asked to be closed
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
Consumer reference: , stream: Sample.Z is disconnected. Client will try reconnect
info: RabbitMQ.Stream.Client.Reliable.Consumer[0]
Consumer reference: , stream: Sample.Z reconnected successfully
Message: 10 | 10
Message: 11 | 11
Message: 12 | 12
Message: 13 | 13
Message: 14 | 14
Message: 15 | 15

[Gsantomaggio]

The problem happens only if you call the close inside the handler, right?

  if (consumerCount == 10)
        await consumer.Close().ConfigureAwait(false);

[daborg69]

No. I just used that code to get back to a simpler situation. So, my actual code is more similar to this:

This is the consumer method, with a delay to simulate a long running consumption task:
` private async Task ConsumeSlow(Message message)
{
_statsList[1].ConsumedMessages++;
_statsList[1].ConsumeLastBatchReceived = (string)message.ApplicationProperties[AP_BATCH];
System.Console.WriteLine("Slow Consumed:");

    //        _statsList[1].ConsumeLastCheckpoint = _consumerB_slow.CheckpointLastOffset;
    _statsList[1].ConsumeCurrentAwaitingCheckpoint = _consumerSlow.CheckpointOffsetCounter;

    // Simulate slow
    await Task.Delay(1000);
    return true;
}`

And here is the method that is called to Stop the consumption (With the dispose in it to force the stopping of message consumption).
public override async Task StopAsync() { // Must call Dispose. The Client Library will re-open the closed connection and continue sending messages. await _streamSystem.Close(); _rawConsumer.Dispose(); IsConnected = false; }

Without the dispose it just keeps receiving messages.

[daborg69]

And now I see the issue. Looked at this 20x and didn't see it. I was calling the streamSystem close and not the rawconsumer.close(). I just changed my code and it stopped receiving messages.

Thanks for your help!