Unhandled exception. System.AggregateException: One or more errors occurred.

[Gsantomaggio]

Found this issue working on #23

The active test run was aborted. Reason: Test host process crashed : Unhandled exception. System.AggregateException: One or more errors occurred. (Operation is not valid due to the current state of the object.)
 ---> System.InvalidOperationException: Operation is not valid due to the current state of the object.
   at System.Threading.Tasks.Sources.ManualResetValueTaskSourceCore`1.SignalCompletion()
   at System.Threading.Tasks.Sources.ManualResetValueTaskSourceCore`1.SetException(Exception error)
   at RabbitMQ.Stream.Client.ManualResetValueTaskSource`1.SetException(Exception error) in /source/RabbitMQ.Stream.Client/Client.cs:line 460
   at RabbitMQ.Stream.Client.Client.<>c__33`2.<Request>b__33_0(Object valueTaskSource) in /source/RabbitMQ.Stream.Client/Client.cs:line 225
   at System.Threading.CancellationTokenSource.CallbackNode.<>c.<ExecuteCallback>b__9_0(Object s)
   at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
--- End of stack trace from previous location ---
   at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
   at System.Threading.CancellationTokenSource.CallbackNode.ExecuteCallback()
   at System.Threading.CancellationTokenSource.ExecuteCallbackHandlers(Boolean throwOnFirstException)
   --- End of inner exception stack trace ---
   at System.Threading.CancellationTokenSource.ExecuteCallbackHandlers(Boolean throwOnFirstException)
   at System.Threading.CancellationTokenSource.NotifyCancellation(Boolean throwOnFirstException)
   at System.Threading.CancellationTokenSource.<>c.<.cctor>b__56_0(Object obj)
   at System.Threading.TimerQueueTimer.CallCallback(Boolean isThreadPool)
   at System.Threading.TimerQueueTimer.Fire(Boolean isThreadPool)
   at System.Threading.TimerQueue.FireNextTimers()
   at System.Threading.TimerQueue.AppDomainTimerCallback(Int32 id)

How to reproduce the problem:

docker run -it --rm --name test \
		-p 5552:5552 -p 5672:5672 -p 15672:15672 \
		-e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="-rabbitmq_stream advertised_host localhost" \
		-v $(pwd):/source \
		--cpuset-cpus="0-1" \
		--pull always \
		pivotalrabbitmq/rabbitmq-stream

Then:

docker exec -it test    /bin/bash
cd /source/
dotnet test Tests/Tests.csproj  --logger "console;verbosity=detailed"

Just reduce the timeout:

private async ValueTask<TOut> Request<TIn, TOut>(Func<uint, TIn> request, int timeout = 100)

[Gsantomaggio]

Disabling the parallel tests seems to solve the problem

[Gsantomaggio]

Close via #25 (comment)

[iuribrindeiro]

Hey @Gsantomaggio I'm still seeing this issue on my local environment. My scenario is the following:
I open 1 Stream System connection for a single consumer to be used by the whole application lifecycle (I was wondering if this is correct... Should I have more than 1 consumer? Could you direct me to some docs that answer this?)
I also open 1 Stream System connection to be used by 8 consumers.
When trying to close all of these connections I get the same error:

I tried to reproduce this in a smaller and simpler project but I couldn't :/
I bet it is realted to the fact that I'm trying to close multiple consumers and one producer at the same time.

Also, I'm starting 2 Stream Systems because if I use only 1, I can't instantiate a producer and a consumer at the same time, a deadlock happens. Also tried to reproduce this on another project but no lucky. Any thoughts?

[Gsantomaggio]

Can you give me more context ?
Ex: client versions ?
RabbitMQ versions ?

[iuribrindeiro]

yeah, sure!
RabbitMQ 3.11.7 Erlang 25.2
RabbitMQ.Stream.Client 1.1.0

[iuribrindeiro]

Oh, and this is a super stream, btw.

Consumer config:

 await Consumer.Create(
          new ConsumerConfig(streamSystem, _rabbitMqClientOptions.SuperStreamName)
          {
              ClientProvidedName = QueueName,
              Reference = _rabbitMqClientOptions.SuperStreamName,
              ConsumerUpdateListener = ConsumerUpdateListener,
              ReconnectStrategy = this,
              MessageHandler = onNewMessage,
              IsSingleActiveConsumer = true,
              IsSuperStream = true
          }, _loggerFactory.CreateLogger<Consumer>()).ConfigureAwait(false)

Producer config:

        await Producer.Create(new ProducerConfig(streamSystem, _rabbitMqClientOptions.SuperStreamName)
        {
            SuperStreamConfig = new SuperStreamConfig
            {
                Routing = msg => msg.Properties.MessageId.ToString()
            }
        }).ConfigureAwait(false);

        return _producer;

StreamSystemOptions:

new StreamSystemConfig
        {
            VirtualHost = "/",
            UserName = "iuri",
            Password = "iuri",
            AddressResolver = lbAddressResolver,
            Endpoints = new List<EndPoint> { lbAddressResolver.EndPoint }
        }

[Gsantomaggio]

@iuribrindeiro can you please open a discussion?
Please add a separate repo with the issue so we can easily reproduce it. Thank you

[iuribrindeiro]

@iuribrindeiro can you please open a discussion? Please add a separate repo with the issue so we can easily reproduce it. Thank you

Done 👇
#222