fix MergeHub_must_work_with_long_streams_when_buffer_size_is_1 (#6564)

* fix MergeHub_must_work_with_long_streams_when_buffer_size_is_1

99% sure that trying to ReSharper this into a method that doesn't use tail recursion actually created a bug here that can occur for really small partitions - making the `firstAttempt` value mutable across all possible instances results in behavior that actually isn't comparable to using isolated function invocations in all instances. Going to run this through CI and see if that adds up.

* Revert "fix MergeHub_must_work_with_long_streams_when_buffer_size_is_1"

This reverts commit 3bc0b46.

* tremendously simplified `MergeHub_must_work_with_long_streams_when_buffer_size_is_1`

Author: Aaronontheweb

src/core/Akka.Streams.Tests/Dsl/HubSpec.cs

@@ -185,25 +185,17 @@ public async Task MergeHub_must_work_with_long_streams_when_buffer_size_is_1()
         {
             await this.AssertAllStagesStoppedAsync(async () =>
             {
-                var (sink, probe) = MergeHub.Source<int>(1)
-                    .ToMaterialized(this.SinkProbe<int>(), Keep.Both)
+                var (sink, result) = MergeHub.Source<int>(1)
+                    .Take(20000)
+                    .ToMaterialized(Sink.Seq<int>(), Keep.Both)
                     .Run(Materializer);
-
+                
+                
                 Source.From(Enumerable.Range(1, 10000)).RunWith(sink, Materializer);
                 Source.From(Enumerable.Range(10001, 10000)).RunWith(sink, Materializer);
-
-                await probe.RequestAsync(int.MaxValue);
-                var result = new List<int>();
-                foreach (var i in Enumerable.Range(1, 20000))
-                {
-                    var evt = await probe.ExpectEventAsync();
-                    if (evt is TestSubscriber.OnNext<int> next)
-                        result.Add(next.Element);
-                    else
-                        throw new Exception($"For element [{i}]: Expected OnNext<int> but received {evt.GetType()}");
-                }
-                result.OrderBy(x => x).Should().BeEquivalentTo(Enumerable.Range(1, 20000));
-            }, Materializer, 300.Seconds());
+                
+                (await result).OrderBy(x => x).Should().BeEquivalentTo(Enumerable.Range(1, 20000));
+            }, Materializer, 3.Seconds());
         }
 
         [Fact]

src/core/Akka.Streams/Dsl/Hub.cs

@@ -86,7 +86,7 @@ public ProducerFailed(string message, Exception cause) : base(message, cause)
     /// <summary>
     /// INTERNAL API
     /// </summary>
-    /// <typeparam name="T">TBD</typeparam>
+    /// <typeparam name="T">The type of element emitted by the MergeHub</typeparam>
     internal class MergeHub<T> : GraphStageWithMaterializedValue<SourceShape<T>, Sink<T, NotUsed>>
     {
         #region Internal classes
@@ -426,14 +426,14 @@ public HubSink(MergeHub<T> hub, AtomicCounterLong idCounter, HubLogic logic)
         /// TBD
         /// </summary>
         /// <param name="perProducerBufferSize">TBD</param>
-        /// <exception cref="ArgumentException">
-        /// This exception is thrown when the specified <paramref name="perProducerBufferSize"/> is less than or equal to zero.
+        /// <exception cref="ArgumentOutOfRangeException">
+        /// This exception is thrown when the specified <paramref name="perProducerBufferSize"/>is less than or equal to zero.
         /// </exception>
         public MergeHub(int perProducerBufferSize)
         {
             if (perProducerBufferSize <= 0)
-                throw new ArgumentException("Buffer size must be positive", nameof(perProducerBufferSize));
-
+                throw new ArgumentOutOfRangeException(nameof(perProducerBufferSize), perProducerBufferSize, "Buffer size must be positive");
+            
             _perProducerBufferSize = perProducerBufferSize;
             DemandThreshold = perProducerBufferSize / 2 + perProducerBufferSize % 2;
             Shape = new SourceShape<T>(Out);