Fix the Unsubscribe timeout problem

[Gsantomaggio]

• The Unsubscribe function sync (like the others). The consumers' list is updated in the right way. The connection is correctly closed since there are no pending consumers.

• Fix the handle delivery if the consumer is removed, but there are still chunks on the wire. Add debug logs for this situation.

• Add Degug Asserts to validate the buffer size, which can help understand some parse chunk errors. That can be temporary at some point; we can remove them. Ref https://github.com/rabbitmq/rabbitmq-stream-dotnet-client/issues/310

• Add the cancellation token to the connection class it helps when the reader is blocked and the consumer is closed.

• Given this use case so short-life consumers, the initial credits should be low to avoid keeping the read handler busy.

• Minor change is replace Memory to ReadOnlyMemory it is a bit faster and safe

• Uniform the log debug and error messages with consumer info it adds all the consumer info that can be helpful in case of debugging and errors

[Gsantomaggio]

@jonnepmyra do you have a chance to test? It adds more detailed logs to understand the chunk parse error

[codecov[bot]]

Codecov Report

Attention: 40 lines in your changes are missing coverage. Please review.

Comparison is base (a2e06bf) 92.79% compared to head (355e635) 92.49%.

:exclamation: Current head 355e635 differs from pull request most recent head 2125dcd. Consider uploading reports for the commit 2125dcd to get more accurate results

Additional details and impacted files

```diff @@ Coverage Diff @@ ## main #313 +/- ## ========================================== - Coverage 92.79% 92.49% -0.31% ========================================== Files 112 112 Lines 9804 9884 +80 Branches 785 825 +40 ========================================== + Hits 9098 9142 +44 - Misses 536 564 +28 - Partials 170 178 +8 ``` | [Files](https://app.codecov.io/gh/rabbitmq/rabbitmq-stream-dotnet-client/pull/313?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=rabbitmq) | Coverage Δ | | |---|---|---| | [RabbitMQ.Stream.Client/Connection.cs](https://app.codecov.io/gh/rabbitmq/rabbitmq-stream-dotnet-client/pull/313?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=rabbitmq#diff-UmFiYml0TVEuU3RyZWFtLkNsaWVudC9Db25uZWN0aW9uLmNz) | `91.39% <81.25%> (+2.73%)` | :arrow_up: | | [RabbitMQ.Stream.Client/Deliver.cs](https://app.codecov.io/gh/rabbitmq/rabbitmq-stream-dotnet-client/pull/313?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=rabbitmq#diff-UmFiYml0TVEuU3RyZWFtLkNsaWVudC9EZWxpdmVyLmNz) | `82.71% <0.00%> (-3.19%)` | :arrow_down: | | [RabbitMQ.Stream.Client/Client.cs](https://app.codecov.io/gh/rabbitmq/rabbitmq-stream-dotnet-client/pull/313?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=rabbitmq#diff-UmFiYml0TVEuU3RyZWFtLkNsaWVudC9DbGllbnQuY3M=) | `88.16% <56.00%> (-2.35%)` | :arrow_down: | | [RabbitMQ.Stream.Client/RawConsumer.cs](https://app.codecov.io/gh/rabbitmq/rabbitmq-stream-dotnet-client/pull/313?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=rabbitmq#diff-UmFiYml0TVEuU3RyZWFtLkNsaWVudC9SYXdDb25zdW1lci5jcw==) | `84.33% <72.61%> (-2.10%)` | :arrow_down: | ... and [2 files with indirect coverage changes](https://app.codecov.io/gh/rabbitmq/rabbitmq-stream-dotnet-client/pull/313/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=rabbitmq)

:umbrella: View full report in Codecov by Sentry.
:loudspeaker: Have feedback on the report? Share it here.

[jonnepmyra]

@jonnepmyra do you have a chance to test? It adds more detailed logs to understand the chunk parse error

Thanks! I'm on it!

[jonnepmyra]

I've been thoroughly testing this pull request over the past few days, and it appears to be a clear improvement.

Before this fix, we encountered the following errors during high-throughput scenarios with short-lived consumers. However, after applying this fix, these issues do not seem to occur anymore:

Error while processing chunk: XYZ

System.NullReferenceException: Object reference not set to an instance of an object.
   at RabbitMQ.Stream.Client.RawConsumer.<>c__DisplayClass14_0.<<ParseChunk>g__DispatchMessage|1>d.MoveNext() in C:\git\rabbitmq-streamclient\rabbitmq-stream-dotnet-client\RabbitMQ.Stream.Client\RawConsumer.cs:line 201

An error occurred while calling Dispose

System.ArgumentNullException: Value cannot be null. (Parameter 'array')
   at System.Buffers.TlsOverPerCoreLockedStacksArrayPool`1.Return(T[] array, Boolean clearArray)
   at System.IO.Pipelines.BufferSegment.ResetMemory()
   at System.IO.Pipelines.BufferSegment.Reset()
   at System.IO.Pipelines.StreamPipeReader.CompleteAndGetNeedsDispose()
   at System.IO.Pipelines.StreamPipeReader.Complete(Exception exception)
   at RabbitMQ.Stream.Client.Connection.Dispose() in C:\git\rabbitmq-streamclient\rabbitmq-stream-dotnet-client\RabbitMQ.Stream.Client\Connection.cs:line 223
   at RabbitMQ.Stream.Client.Client.Close(String reason) in C:\git\rabbitmq-streamclient\rabbitmq-stream-dotnet-client\RabbitMQ.Stream.Client\Client.cs:line 583

While we are still experiencing some Close() timeouts with short-lived stream consumers, it's worth noting that this may be related to a non-default InitialCredits value (specifically, 100 instead of the default 2) mentioned by @Gsantomaggio. At present, we can consider this as a non-issue.

Great work and thanks for helping out!

[Gsantomaggio]

Thank you @jonnepmyra for the contribution