* More improvements to TestBasicRoundtripConcurrentManyMessages

lukebakken · lukebakken · commit a6cfa912c111 · 2024-05-09T10:55:55.000-07:00
diff --git a/projects/Test/Integration/TestAsyncConsumer.cs b/projects/Test/Integration/TestAsyncConsumer.cs
@@ -159,8 +159,8 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                     {
                         if (args.Initiator == ShutdownInitiator.Peer)
                         {
-                            publish1SyncSource.TrySetResult(false);
-                            publish2SyncSource.TrySetResult(false);
+                            publish1SyncSource.TrySetException(ea.Exception);
+                            publish2SyncSource.TrySetException(ea.Exception);
                         }
                     });
                 };
@@ -171,8 +171,8 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                     {
                         if (args.Initiator == ShutdownInitiator.Peer)
                         {
-                            publish1SyncSource.TrySetResult(false);
-                            publish2SyncSource.TrySetResult(false);
+                            publish1SyncSource.TrySetException(ea.Exception);
+                            publish2SyncSource.TrySetException(ea.Exception);
                         }
                     });
                 };
@@ -209,25 +209,30 @@ public async Task TestBasicRoundtripConcurrentManyMessages()
                                 int publish1_count = 0;
                                 int publish2_count = 0;
 
-                                consumer.Received += async (o, a) =>
+                                consumer.Received += (o, a) =>
                                 {
                                     string decoded = _encoding.GetString(a.Body.ToArray());
                                     if (decoded == publish1)
                                     {
                                         if (Interlocked.Increment(ref publish1_count) >= publish_total)
                                         {
                                             publish1SyncSource.TrySetResult(true);
-                                            await publish2SyncSource.Task;
                                         }
                                     }
                                     else if (decoded == publish2)
                                     {
                                         if (Interlocked.Increment(ref publish2_count) >= publish_total)
                                         {
                                             publish2SyncSource.TrySetResult(true);
-                                            await publish1SyncSource.Task;
                                         }
                                     }
+                                    else
+                                    {
+                                        var ex = new InvalidOperationException("incorrect message - should never happen!");
+                                        publish1SyncSource.TrySetException(ex);
+                                        publish2SyncSource.TrySetException(ex);
+                                    }
+                                    return Task.CompletedTask;
                                 };
 
                                 await consumeChannel.BasicConsumeAsync(queueName, true, string.Empty, false, false, null, consumer);

Original file line number	Diff line number	Diff line change
`@@ -159,8 +159,8 @@ public async Task TestBasicRoundtripConcurrentManyMessages()`
`159`	`159`	`{`
`160`	`160`	`if (args.Initiator == ShutdownInitiator.Peer)`
`161`	`161`	`{`
`162`		`- publish1SyncSource.TrySetResult(false);`
`163`		`- publish2SyncSource.TrySetResult(false);`
	`162`	`+ publish1SyncSource.TrySetException(ea.Exception);`
	`163`	`+ publish2SyncSource.TrySetException(ea.Exception);`
`164`	`164`	`}`
`165`	`165`	`});`
`166`	`166`	`};`
`@@ -171,8 +171,8 @@ public async Task TestBasicRoundtripConcurrentManyMessages()`
`171`	`171`	`{`
`172`	`172`	`if (args.Initiator == ShutdownInitiator.Peer)`
`173`	`173`	`{`
`174`		`- publish1SyncSource.TrySetResult(false);`
`175`		`- publish2SyncSource.TrySetResult(false);`
	`174`	`+ publish1SyncSource.TrySetException(ea.Exception);`
	`175`	`+ publish2SyncSource.TrySetException(ea.Exception);`
`176`	`176`	`}`
`177`	`177`	`});`
`178`	`178`	`};`
`@@ -209,25 +209,30 @@ public async Task TestBasicRoundtripConcurrentManyMessages()`
`209`	`209`	`int publish1_count = 0;`
`210`	`210`	`int publish2_count = 0;`
`211`	`211`
`212`		`- consumer.Received += async (o, a) =>`
	`212`	`+ consumer.Received += (o, a) =>`
`213`	`213`	`{`
`214`	`214`	`string decoded = _encoding.GetString(a.Body.ToArray());`
`215`	`215`	`if (decoded == publish1)`
`216`	`216`	`{`
`217`	`217`	`if (Interlocked.Increment(ref publish1_count) >= publish_total)`
`218`	`218`	`{`
`219`	`219`	`publish1SyncSource.TrySetResult(true);`
`220`		`- await publish2SyncSource.Task;`
`221`	`220`	`}`
`222`	`221`	`}`
`223`	`222`	`else if (decoded == publish2)`
`224`	`223`	`{`
`225`	`224`	`if (Interlocked.Increment(ref publish2_count) >= publish_total)`
`226`	`225`	`{`
`227`	`226`	`publish2SyncSource.TrySetResult(true);`
`228`		`- await publish1SyncSource.Task;`
`229`	`227`	`}`
`230`	`228`	`}`
	`229`	`+ else`
	`230`	`+ {`
	`231`	`+ var ex = new InvalidOperationException("incorrect message - should never happen!");`
	`232`	`+ publish1SyncSource.TrySetException(ex);`
	`233`	`+ publish2SyncSource.TrySetException(ex);`
	`234`	`+ }`
	`235`	`+ return Task.CompletedTask;`
`231`	`236`	`};`
`232`	`237`
`233`	`238`	`await consumeChannel.BasicConsumeAsync(queueName, true, string.Empty, false, false, null, consumer);`