From 609ab9cf25c3ba89856854a34fc5675ced8bce7c Mon Sep 17 00:00:00 2001
From: Gregorius Soedharmo <arkatufus@yahoo.com>
Date: Tue, 10 May 2022 00:18:38 +0700
Subject: [PATCH 1/2] Add async fluent builder feature to TestPublisher

---
 .../PublisherFluentBuilder.cs                 | 314 ++++++++++++++++++
 .../SubscriberFluentBuilder.cs                | 173 +++++++---
 .../Akka.Streams.TestKit/TestPublisher.cs     | 185 ++++-------
 .../TestPublisher_Fluent.cs                   | 113 +++++++
 .../TestPublisher_Shared.cs                   | 141 ++++++++
 .../Akka.Streams.TestKit/TestSubscriber.cs    |  53 ++-
 .../TestSubscriber_Fluent.cs                  |  30 ++
 .../TestSubscriber_Shared.cs                  |   4 +-
 src/core/Akka.Streams.TestKit/Utils.cs        |   4 +-
 9 files changed, 837 insertions(+), 180 deletions(-)
 create mode 100644 src/core/Akka.Streams.TestKit/PublisherFluentBuilder.cs
 create mode 100644 src/core/Akka.Streams.TestKit/TestPublisher_Fluent.cs
 create mode 100644 src/core/Akka.Streams.TestKit/TestPublisher_Shared.cs

diff --git a/src/core/Akka.Streams.TestKit/PublisherFluentBuilder.cs b/src/core/Akka.Streams.TestKit/PublisherFluentBuilder.cs
new file mode 100644
index 00000000000..5b8f5e9b9fd
--- /dev/null
+++ b/src/core/Akka.Streams.TestKit/PublisherFluentBuilder.cs
@@ -0,0 +1,314 @@
+// //-----------------------------------------------------------------------
+// // <copyright file="PublisherFluentBuilder.cs" company="Akka.NET Project">
+// //     Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
+// //     Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// // </copyright>
+// //-----------------------------------------------------------------------
+
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Akka.TestKit;
+using Reactive.Streams;
+using static Akka.Streams.TestKit.TestPublisher;
+
+namespace Akka.Streams.TestKit
+{
+    public class PublisherFluentBuilder<T>
+    {
+        private readonly List<Func<CancellationToken, Task>> _tasks = new List<Func<CancellationToken, Task>>();
+        private bool _executed;
+
+        internal PublisherFluentBuilder(ManualProbe<T> probe)
+        {
+            Probe = probe;
+        }
+
+        public ManualProbe<T> Probe { get; }
+
+        /// <summary>
+        /// Execute the async chain.
+        /// </summary>
+        /// <param name="asyncAction"></param>
+        /// <param name="cancellationToken"></param>
+        /// <exception cref="InvalidOperationException"></exception>
+        public async Task ExecuteAsync(Func<Task>? asyncAction = null, CancellationToken cancellationToken = default)
+        {
+            if (_executed)
+                throw new InvalidOperationException("Fluent async builder has already been executed.");
+            _executed = true;
+            
+            foreach (var func in _tasks)
+            {
+                await func(cancellationToken)
+                    .ConfigureAwait(false);
+            }
+
+            if (asyncAction != null)
+                await asyncAction()
+                    .ConfigureAwait(false);
+        }
+
+        #region ManualProbe<T> wrapper
+
+        /// <summary>
+        /// Execute the async chain and then receive messages for a given duration or until one does not match a given partial function.
+        /// NOTE: This method will execute the async chain
+        /// </summary>
+        public async IAsyncEnumerable<TOther> ReceiveWhileAsync<TOther>(
+            TimeSpan? max = null,
+            TimeSpan? idle = null,
+            Func<object, TOther>? filter = null,
+            int msgCount = int.MaxValue,
+            [EnumeratorCancellation] CancellationToken cancellationToken = default) where TOther : class
+        {
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            await foreach (var item in ManualProbe<T>.ReceiveWhileTask(Probe.Probe, max, idle, filter, msgCount, cancellationToken))
+            {
+                yield return item;
+            }
+        }
+        
+        /// <summary>
+        /// Execute the async chain and then expect a publisher event from the stream.
+        /// NOTE: This method will execute the async chain
+        /// </summary>
+        public async Task<IPublisherEvent> ExpectEventAsync(CancellationToken cancellationToken = default)
+        {
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await ManualProbe<T>.ExpectEventTask(Probe.Probe, cancellationToken)
+                .ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Execute the async chain and then execute the code block while bounding its execution time between <paramref name="min"/> and
+        /// <paramref name="max"/>. <see cref="WithinAsync{TOther}(TimeSpan,TimeSpan,Func{TOther},CancellationToken)"/> blocks may be nested. 
+        /// All methods in this class which take maximum wait times are available in a version which implicitly uses
+        /// the remaining time governed by the innermost enclosing <see cref="WithinAsync{TOther}(TimeSpan,TimeSpan,Func{TOther},CancellationToken)"/> block.
+        /// 
+        /// <para />
+        /// 
+        /// Note that the timeout is scaled using <see cref="TestKitBase.Dilated"/>, which uses the
+        /// configuration entry "akka.test.timefactor", while the min Duration is not.
+        /// 
+        /// <![CDATA[
+        /// var ret = await probe.AsyncBuilder().Within(Timespan.FromMilliseconds(50), Timespan.FromSeconds(3), () =>
+        /// {
+        ///     test.Tell("ping");
+        ///     return ExpectMsg<string>();
+        /// });
+        /// ]]>
+        /// 
+        /// <![CDATA[
+        /// await probe.AsyncBuilder().Within(Timespan.FromMilliseconds(50), Timespan.FromSeconds(3), async () =>
+        /// {
+        ///     test.Tell("ping");
+        ///     await ExpectMsgAsync<string>("expected");
+        /// });
+        /// 
+        /// NOTE: This method will execute the async chain
+        /// ]]>
+        /// </summary>
+        public async Task<TOther> WithinAsync<TOther>(
+            TimeSpan min,
+            TimeSpan max,
+            Func<TOther> function,
+            CancellationToken cancellationToken = default)
+        {
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await Probe.WithinAsync(min, max, function, cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Execute the async chain and then execute the code block while bounding its execution time between <paramref name="min"/> and
+        /// <paramref name="max"/>. <see cref="WithinAsync{TOther}(TimeSpan,TimeSpan,Func{TOther},CancellationToken)"/> blocks may be nested. 
+        /// All methods in this class which take maximum wait times are available in a version which implicitly uses
+        /// the remaining time governed by the innermost enclosing <see cref="WithinAsync{TOther}(TimeSpan,TimeSpan,Func{TOther},CancellationToken)"/> block.
+        /// 
+        /// <para />
+        /// 
+        /// Note that the timeout is scaled using <see cref="TestKitBase.Dilated"/>, which uses the
+        /// configuration entry "akka.test.timefactor", while the min Duration is not.
+        /// 
+        /// <![CDATA[
+        /// var ret = await probe.AsyncBuilder().Within(Timespan.FromMilliseconds(50), Timespan.FromSeconds(3), async () =>
+        /// {
+        ///     test.Tell("ping");
+        ///     return await ExpectMsgAsync<string>();
+        /// });
+        /// ]]>
+        /// 
+        /// NOTE: This method will execute the async chain
+        /// </summary>
+        public async Task<TOther> WithinAsync<TOther>(
+            TimeSpan min,
+            TimeSpan max,
+            Func<Task<TOther>> function,
+            CancellationToken cancellationToken = default)
+        {
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await Probe.WithinAsync(min, max, function, cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Sane as calling WithinAsync(TimeSpan.Zero, max, function, cancellationToken).
+        /// 
+        /// NOTE: This method will execute the async chain
+        /// </summary>
+        public async Task<TOther> WithinAsync<TOther>(TimeSpan max, Func<TOther> execute, CancellationToken cancellationToken = default)
+        {
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await Probe.WithinAsync(max, execute, cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Sane as calling WithinAsync(TimeSpan.Zero, max, function, cancellationToken).
+        /// 
+        /// NOTE: This method will execute the async chain
+        /// </summary>
+        public async Task<TOther> WithinAsync<TOther>(TimeSpan max, Func<Task<TOther>> execute, CancellationToken cancellationToken = default)
+        {
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await Probe.WithinAsync(max, execute, cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+        }
+        
+        #endregion
+        
+        #region ManualProbe<T> fluent wrapper
+
+        /// <summary>
+        /// Fluent async DSL
+        /// Expect demand from the given subscription.
+        /// </summary>
+        public PublisherFluentBuilder<T> ExpectRequest(ISubscription subscription, int nrOfElements)
+        {
+            _tasks.Add(ct => ManualProbe<T>.ExpectRequestTask(Probe.Probe, subscription, nrOfElements, ct));
+            return this;
+        }
+
+        /// <summary>
+        /// Fluent async DSL
+        /// Expect no messages.
+        /// </summary>
+        public PublisherFluentBuilder<T> ExpectNoMsg()
+        {
+            _tasks.Add(ct => ManualProbe<T>.ExpectNoMsgTask(Probe.Probe, ct));
+            return this;
+        }
+
+        /// <summary>
+        /// Fluent async DSL
+        /// Expect no messages for given duration.
+        /// </summary>
+        public PublisherFluentBuilder<T> ExpectNoMsg(TimeSpan duration, CancellationToken cancellationToken = default)
+        {
+            _tasks.Add(ct => ManualProbe<T>.ExpectNoMsgTask(Probe.Probe, duration, ct));
+            return this;
+        }
+
+        #endregion
+        
+        #region Probe<T> fluent wrapper
+        
+        /// <summary>
+        /// Asserts that a subscription has been received or will be received
+        /// </summary>
+        public PublisherFluentBuilder<T> EnsureSubscription()
+        {
+            if (!(Probe is Probe<T> probe))
+            {
+                throw new InvalidOperationException($"{nameof(EnsureSubscription)} can only be used on a {nameof(Probe<T>)} instance");
+            }
+            _tasks.Add(ct => Probe<T>.EnsureSubscriptionTask(probe, ct));
+            return this;
+        }
+
+        public PublisherFluentBuilder<T> SendNext(T element)
+        {
+            if (!(Probe is Probe<T> probe))
+            {
+                throw new InvalidOperationException($"{nameof(SendNext)} can only be used on a {nameof(Probe<T>)} instance");
+            }
+            _tasks.Add(ct => Probe<T>.SendNextTask(probe, element, ct));
+            return this;
+        }
+
+        public PublisherFluentBuilder<T> SendNext(IEnumerable<T> elements)
+        {
+            if (!(Probe is Probe<T> probe))
+            {
+                throw new InvalidOperationException($"{nameof(SendNext)} can only be used on a {nameof(Probe<T>)} instance");
+            }
+            _tasks.Add(ct => Probe<T>.SendNextTask(probe, elements, ct));
+            return this;
+        }
+
+        public PublisherFluentBuilder<T> UnsafeSendNext(T element)
+        {
+            if (!(Probe is Probe<T> probe))
+            {
+                throw new InvalidOperationException($"{nameof(UnsafeSendNext)} can only be used on a {nameof(Probe<T>)} instance");
+            }
+            _tasks.Add(ct => Probe<T>.UnsafeSendNextTask(probe, element, ct));
+            return this;
+        }
+
+        public PublisherFluentBuilder<T> SendComplete()
+        {
+            if (!(Probe is Probe<T> probe))
+            {
+                throw new InvalidOperationException($"{nameof(SendComplete)} can only be used on a {nameof(Probe<T>)} instance");
+            }
+            _tasks.Add(ct => Probe<T>.SendCompleteTask(probe, ct));
+            return this;
+        }
+
+        public PublisherFluentBuilder<T> SendError(Exception e)
+        {
+            if (!(Probe is Probe<T> probe))
+            {
+                throw new InvalidOperationException($"{nameof(SendError)} can only be used on a {nameof(Probe<T>)} instance");
+            }
+            _tasks.Add(ct => Probe<T>.SendErrorTask(probe, e, ct));
+            return this;
+        }
+
+        public PublisherFluentBuilder<T> ExpectCancellation()
+        {
+            if (!(Probe is Probe<T> probe))
+            {
+                throw new InvalidOperationException($"{nameof(ExpectCancellation)} can only be used on a {nameof(Probe<T>)} instance");
+            }
+            _tasks.Add(ct => Probe<T>.ExpectCancellationTask(probe, ct));
+            return this;
+        }
+        
+        #endregion
+        
+        public async Task<long> ExpectRequestAsync(CancellationToken cancellationToken = default)
+        {
+            if (!(Probe is Probe<T> probe))
+            {
+                throw new InvalidOperationException($"{nameof(ExpectCancellation)} can only be used on a {nameof(Probe<T>)} instance");
+            }
+
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await Probe<T>.ExpectRequestTask(probe, cancellationToken)
+                .ConfigureAwait(false);
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/core/Akka.Streams.TestKit/SubscriberFluentBuilder.cs b/src/core/Akka.Streams.TestKit/SubscriberFluentBuilder.cs
index 822463c6dc8..56513a8132c 100644
--- a/src/core/Akka.Streams.TestKit/SubscriberFluentBuilder.cs
+++ b/src/core/Akka.Streams.TestKit/SubscriberFluentBuilder.cs
@@ -27,7 +27,8 @@ public class SubscriberFluentBuilder<T>
         /// </summary>
         public async Task<ISubscription> ExpectSubscriptionAsync(CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectSubscriptionTask(Probe, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -39,7 +40,8 @@ public async Task<ISubscription> ExpectSubscriptionAsync(CancellationToken cance
         /// </summary>
         public async Task<ISubscriberEvent> ExpectEventAsync(CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectEventTask(Probe.TestProbe, (TimeSpan?)null, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -53,7 +55,8 @@ public async Task<ISubscriberEvent> ExpectEventAsync(
             TimeSpan? max,
             CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectEventTask(Probe.TestProbe, max, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -64,7 +67,8 @@ public async Task<ISubscriberEvent> ExpectEventAsync(
         /// </summary>
         public async Task<T> ExpectNextAsync(CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectNextTask(Probe.TestProbe, null, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -75,7 +79,8 @@ public async Task<T> ExpectNextAsync(CancellationToken cancellationToken = defau
         /// </summary>
         public async Task<T> ExpectNextAsync(TimeSpan? timeout, CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectNextTask(Probe.TestProbe, timeout, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -89,7 +94,8 @@ public async IAsyncEnumerable<T> ExpectNextNAsync(
             TimeSpan? timeout = null,
             [EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             await foreach (var item in ManualProbe<T>.ExpectNextNTask(Probe.TestProbe, n, timeout, cancellationToken))
             {
                 yield return item;
@@ -102,7 +108,8 @@ public async IAsyncEnumerable<T> ExpectNextNAsync(
         /// </summary>
         public async Task<Exception> ExpectErrorAsync(CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectErrorTask(Probe.TestProbe, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -115,7 +122,8 @@ public async Task<Exception> ExpectErrorAsync(CancellationToken cancellationToke
         /// </summary>
         public async Task<Exception> ExpectSubscriptionAndErrorAsync(CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectSubscriptionAndErrorTask(Probe, true, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -132,7 +140,8 @@ public async Task<Exception> ExpectSubscriptionAndErrorAsync(
             bool signalDemand, 
             CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectSubscriptionAndErrorTask(Probe, signalDemand, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -143,7 +152,8 @@ public async Task<Exception> ExpectSubscriptionAndErrorAsync(
         /// </summary>
         public async Task<object> ExpectNextOrErrorAsync(CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectNextOrErrorTask(Probe.TestProbe, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -154,7 +164,8 @@ public async Task<object> ExpectNextOrErrorAsync(CancellationToken cancellationT
         /// </summary>
         public async Task<object> ExpectNextOrCompleteAsync(CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectNextOrCompleteTask(Probe.TestProbe, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -171,7 +182,8 @@ public async Task<TOther> ExpectNextAsync<TOther>(
             Predicate<TOther> predicate,
             CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectNextTask(Probe.TestProbe, predicate, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -180,7 +192,8 @@ public async Task<TOther> ExpectEventAsync<TOther>(
             Func<ISubscriberEvent, TOther> func,
             CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             return await ManualProbe<T>.ExpectEventTask(Probe.TestProbe, func, cancellationToken)
                 .ConfigureAwait(false);
         }
@@ -196,7 +209,8 @@ public async IAsyncEnumerable<TOther> ReceiveWhileAsync<TOther>(
             int msgs = int.MaxValue,
             [EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             await foreach (var item in Probe.TestProbe.ReceiveWhileAsync(max, idle, filter, msgs, cancellationToken))
             {
                 yield return item;
@@ -212,7 +226,8 @@ public async IAsyncEnumerable<TOther> ReceiveWithinAsync<TOther>(
             int messages = int.MaxValue,
             [EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             await foreach (var item in ManualProbe<T>.ReceiveWithinTask<TOther>(Probe.TestProbe, max, messages, cancellationToken))
             {
                 yield return item;
@@ -220,36 +235,93 @@ public async IAsyncEnumerable<TOther> ReceiveWithinAsync<TOther>(
         }
         
         /// <summary>
-        /// Execute code block while bounding its execution time between <paramref name="min"/> and <paramref name="max"/>.
+        /// Execute the async chain and then execute the code block while bounding its execution time between <paramref name="min"/> and <paramref name="max"/>.
         /// <para />
         /// Note that the timeout is scaled using <see cref="TestKitBase.Dilated"/>, which uses the
         /// configuration entry "akka.test.timefactor", while the min Duration is not.
         /// 
         /// <![CDATA[
-        /// var ret = probe.AsyncBuilder().WithinAsync(Timespan.FromMilliseconds(50), Timespan.FromSeconds(3), () =>
+        /// var ret = await probe.AsyncBuilder().WithinAsync(Timespan.FromMilliseconds(50), Timespan.FromSeconds(3), () =>
         /// {
         ///     test.Tell("ping");
         ///     return ExpectMsg<string>();
         /// });
         /// ]]>
+        ///
+        /// <![CDATA[
+        /// await probe.AsyncBuilder().WithinAsync(Timespan.FromMilliseconds(50), Timespan.FromSeconds(3), async () =>
+        /// {
+        ///     test.Tell("ping");
+        ///     await ExpectMsgAsync<string>("expected");
+        /// });
+        /// ]]>
+        /// 
+        /// NOTE: This method will execute the async chain
         /// </summary>
-        public async Task<TOther> WithinAsync<TOther>(TimeSpan min, TimeSpan max, Func<TOther> execute, CancellationToken cancellationToken = default)
+        public async Task<TOther> WithinAsync<TOther>(
+            TimeSpan min,
+            TimeSpan max,
+            Func<TOther> function,
+            string? hint = null,
+            TimeSpan? epsilonValue = null, 
+            CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
-            return await Probe.TestProbe.WithinAsync(min, max, execute, cancellationToken: cancellationToken)
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await Probe.TestProbe.WithinAsync(min, max, function, hint, epsilonValue, cancellationToken)
+                .ConfigureAwait(false);
+        }
+        
+        /// <summary>
+        /// Execute the async chain and then execute the code block while bounding its execution time between <paramref name="min"/> and <paramref name="max"/>.
+        /// <para />
+        /// Note that the timeout is scaled using <see cref="TestKitBase.Dilated"/>, which uses the
+        /// configuration entry "akka.test.timefactor", while the min Duration is not.
+        /// 
+        /// <![CDATA[
+        /// var ret = await probe.AsyncBuilder().WithinAsync(Timespan.FromMilliseconds(50), Timespan.FromSeconds(3), async () =>
+        /// {
+        ///     test.Tell("ping");
+        ///     await ExpectMsgAsync<string>("expected");
+        /// });
+        /// ]]>
+        /// 
+        /// NOTE: This method will execute the async chain
+        /// </summary>
+        public async Task<TOther> WithinAsync<TOther>(
+            TimeSpan min,
+            TimeSpan max,
+            Func<Task<TOther>> asyncFunction,
+            string? hint = null,
+            TimeSpan? epsilonValue = null, 
+            CancellationToken cancellationToken = default)
+        {
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await Probe.TestProbe.WithinAsync(min, max, asyncFunction, hint, epsilonValue, cancellationToken)
                 .ConfigureAwait(false);
         }
         
         /// <summary>
-        /// Execute code block while bounding its execution time with a <paramref name="max"/> timeout.
+        /// Execute the async chain and then execute code block while bounding its execution time with a <paramref name="max"/> timeout.
         /// 
         /// <![CDATA[
-        /// var ret = probe.AsyncBuilder().WithinAsync(Timespan.FromSeconds(3), () =>
+        /// var ret = await probe.AsyncBuilder().WithinAsync(Timespan.FromSeconds(3), () =>
         /// {
         ///     test.Tell("ping");
         ///     return ExpectMsg<string>();
         /// });
         /// ]]>
+        ///
+        /// <![CDATA[
+        /// await probe.AsyncBuilder().WithinAsync(Timespan.FromSeconds(3), async () =>
+        /// {
+        ///     test.Tell("ping");
+        ///     await ExpectMsgAsync<string>("expected");
+        /// });
+        /// ]]>
+        /// 
+        /// NOTE: This method will execute the async chain
         /// </summary>
         public async Task<TOther> WithinAsync<TOther>(TimeSpan max, Func<TOther> execute, CancellationToken cancellationToken = default)
         {
@@ -258,31 +330,58 @@ public async Task<TOther> WithinAsync<TOther>(TimeSpan max, Func<TOther> execute
                 .ConfigureAwait(false);
         }
         
-        public async Task WithinAsync(
+        /// <summary>
+        /// Execute the async chain and then execute code block while bounding its execution time with a <paramref name="max"/> timeout.
+        /// 
+        /// <![CDATA[
+        /// var ret = await probe.AsyncBuilder().WithinAsync(Timespan.FromSeconds(3), async () =>
+        /// {
+        ///     test.Tell("ping");
+        ///     return await ExpectMsgAsync<string>();
+        /// });
+        /// ]]>
+        /// NOTE: This method will execute the async chain
+        /// </summary>
+        public async Task<TOther> WithinAsync<TOther>(
             TimeSpan max,
-            Func<Task> actionAsync,
+            Func<Task<TOther>> actionAsync,
             TimeSpan? epsilonValue = null,
             CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
-            await Probe.TestProbe.WithinAsync(max, actionAsync, epsilonValue, cancellationToken)
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
+            return await Probe.TestProbe.WithinAsync(max, actionAsync, epsilonValue, cancellationToken)
                 .ConfigureAwait(false);
         }
         
         /// <summary>
-        /// Attempt to drain the stream into an IAsyncEnumerable (by requesting long.MaxValue elements).
+        /// Execute the async chain and then attempt to drain the stream into an IAsyncEnumerable (by requesting long.MaxValue elements).
+        /// NOTE: This method will execute the async chain
         /// </summary>
         public async IAsyncEnumerable<T> ToStrictAsync(
             TimeSpan atMost,
             [EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
-            await ExecuteAsync(cancellationToken: cancellationToken).ConfigureAwait(false);
+            await ExecuteAsync(cancellationToken: cancellationToken)
+                .ConfigureAwait(false);
             await foreach (var item in ManualProbe<T>.ToStrictTask(Probe, atMost, cancellationToken))
             {
                 yield return item;
             }
         }
 
+        #endregion
+
+        private readonly List<Func<CancellationToken, Task>> _tasks = new List<Func<CancellationToken, Task>>();
+        private bool _executed;
+
+        internal SubscriberFluentBuilder(ManualProbe<T> probe)
+        {
+            Probe = probe;
+        }
+        
+        public ManualProbe<T> Probe { get; }
+
         /// <summary>
         /// Execute the async chain.
         /// </summary>
@@ -297,25 +396,15 @@ public async Task ExecuteAsync(Func<Task>? asyncAction = null, CancellationToken
             
             foreach (var func in _tasks)
             {
-                await func(cancellationToken);
+                await func(cancellationToken)
+                    .ConfigureAwait(false);
             }
 
             if (asyncAction != null)
-                await asyncAction();
+                await asyncAction()
+                    .ConfigureAwait(false);
         }
         
-        #endregion
-
-        private readonly List<Func<CancellationToken, Task>> _tasks = new List<Func<CancellationToken, Task>>();
-        private bool _executed;
-
-        internal SubscriberFluentBuilder(ManualProbe<T> probe)
-        {
-            Probe = probe;
-        }
-        
-        public ManualProbe<T> Probe { get; }
-
         #region Probe<T> wrapper
 
         /// <summary>
diff --git a/src/core/Akka.Streams.TestKit/TestPublisher.cs b/src/core/Akka.Streams.TestKit/TestPublisher.cs
index 2bd4ee258bd..1a2e542e9f3 100644
--- a/src/core/Akka.Streams.TestKit/TestPublisher.cs
+++ b/src/core/Akka.Streams.TestKit/TestPublisher.cs
@@ -21,7 +21,7 @@ namespace Akka.Streams.TestKit
     /// <summary>
     /// Provides factory methods for various Publishers.
     /// </summary>
-    public static class TestPublisher
+    public static partial class TestPublisher
     {
         #region messages
 
@@ -66,13 +66,14 @@ public RequestMore(ISubscription subscription, long nrOfElements)
         /// This probe does not track demand.Therefore you need to expect demand before sending
         ///  elements downstream.
         /// </summary>
-        public class ManualProbe<T> : IPublisher<T>
+        public partial class ManualProbe<T> : IPublisher<T>
         {
-            private readonly TestProbe _probe;
+            private volatile StreamTestKit.PublisherProbeSubscription<T> _subscription_DoNotUseDirectly;
+            public TestProbe Probe { get; }
 
             internal ManualProbe(TestKitBase system, bool autoOnSubscribe = true)
             {
-                _probe = system.CreateTestProbe();
+                Probe = system.CreateTestProbe();
                 AutoOnSubscribe = autoOnSubscribe;
             }
 
@@ -80,13 +81,21 @@ internal ManualProbe(TestKitBase system, bool autoOnSubscribe = true)
 
             public IPublisher<T> Publisher => this;
 
+            public StreamTestKit.PublisherProbeSubscription<T> Subscription
+            {
+#pragma warning disable CS0420
+                get => Volatile.Read(ref _subscription_DoNotUseDirectly);
+                protected set => Volatile.Write(ref _subscription_DoNotUseDirectly, value);
+#pragma warning restore CS0420
+            }
+            
             /// <summary>
             /// Subscribes a given <paramref name="subscriber"/> to this probe.
             /// </summary>
             public void Subscribe(ISubscriber<T> subscriber)
             {
-                var subscription = new StreamTestKit.PublisherProbeSubscription<T>(subscriber, _probe);
-                _probe.Ref.Tell(new Subscribe(subscription));
+                var subscription = new StreamTestKit.PublisherProbeSubscription<T>(subscriber, Probe);
+                Probe.Ref.Tell(new Subscribe(subscription));
                 if (AutoOnSubscribe) subscriber.OnSubscribe(subscription);
             }
 
@@ -95,7 +104,7 @@ public void Subscribe(ISubscriber<T> subscriber)
             /// </summary>
             public StreamTestKit.PublisherProbeSubscription<T> ExpectSubscription(
                 CancellationToken cancellationToken = default)
-                => ExpectSubscriptionAsync(cancellationToken)
+                => ExpectSubscriptionTask(Probe, cancellationToken)
                     .ConfigureAwait(false).GetAwaiter().GetResult();
 
             /// <summary>
@@ -103,64 +112,32 @@ public StreamTestKit.PublisherProbeSubscription<T> ExpectSubscription(
             /// </summary>
             public async Task<StreamTestKit.PublisherProbeSubscription<T>> ExpectSubscriptionAsync(
                 CancellationToken cancellationToken = default)
-            {
-                var msg = await _probe.ExpectMsgAsync<Subscribe>(cancellationToken: cancellationToken)
-                    .ConfigureAwait(false); 
-                return (StreamTestKit.PublisherProbeSubscription<T>) msg.Subscription;
-            }
+                => await ExpectSubscriptionTask(Probe, cancellationToken)
+                    .ConfigureAwait(false);
 
-            /// <summary>
-            /// Expect demand from the given subscription.
-            /// </summary>
-            public void ExpectRequest(
-                ISubscription subscription,
-                int n,
-                CancellationToken cancellationToken = default)
-                => ExpectRequestAsync(subscription, n, cancellationToken)
-                    .ConfigureAwait(false).GetAwaiter().GetResult();
-            
             /// <summary>
             /// Expect demand from the given subscription.
             /// </summary>
             public async Task ExpectRequestAsync(
                 ISubscription subscription, 
-                int n,
+                int nrOfElements,
                 CancellationToken cancellationToken = default)
-            {
-                await _probe.ExpectMsgAsync<RequestMore>(
-                    isMessage: x => x.NrOfElements == n && x.Subscription == subscription, 
-                    cancellationToken: cancellationToken);
-            }
-
-            /// <summary>
-            /// Expect no messages.
-            /// </summary>
-            public void ExpectNoMsg(CancellationToken cancellationToken = default)
-                => ExpectNoMsgAsync(cancellationToken)
-                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                => await ExpectRequestTask(Probe, subscription, nrOfElements, cancellationToken)
+                    .ConfigureAwait(false);
 
             /// <summary>
             /// Expect no messages.
             /// </summary>
             public async Task ExpectNoMsgAsync(CancellationToken cancellationToken = default)
-            {
-                await _probe.ExpectNoMsgAsync(cancellationToken);
-            }
-
-            /// <summary>
-            /// Expect no messages for given duration.
-            /// </summary>
-            public void ExpectNoMsg(TimeSpan duration, CancellationToken cancellationToken = default)
-                => ExpectNoMsgAsync(duration, cancellationToken)
-                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                => await ExpectNoMsgTask(Probe, cancellationToken)
+                    .ConfigureAwait(false);
 
             /// <summary>
             /// Expect no messages for given duration.
             /// </summary>
             public async Task ExpectNoMsgAsync(TimeSpan duration, CancellationToken cancellationToken = default)
-            {
-                await _probe.ExpectNoMsgAsync(duration, cancellationToken);
-            }
+                => await ExpectNoMsgTask(Probe, duration, cancellationToken)
+                    .ConfigureAwait(false);
 
             /// <summary>
             /// Receive messages for a given duration or until one does not match a given partial function.
@@ -169,10 +146,10 @@ public IEnumerable<TOther> ReceiveWhile<TOther>(
                 TimeSpan? max = null,
                 TimeSpan? idle = null,
                 Func<object, TOther> filter = null,
-                int msgs = int.MaxValue,
+                int msgCount = int.MaxValue,
                 CancellationToken cancellationToken = default) where TOther : class
-                => ReceiveWhileAsync(max, idle, filter, msgs, cancellationToken)
-                    .ToListAsync(cancellationToken).ConfigureAwait(false).GetAwaiter().GetResult();
+                => ReceiveWhileTask(Probe, max, idle, filter, msgCount, cancellationToken).ToListAsync(cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
 
             /// <summary>
             /// Receive messages for a given duration or until one does not match a given partial function.
@@ -181,16 +158,22 @@ public IAsyncEnumerable<TOther> ReceiveWhileAsync<TOther>(
                 TimeSpan? max = null,
                 TimeSpan? idle = null,
                 Func<object, TOther> filter = null, 
-                int msgs = int.MaxValue,
+                int msgCount = int.MaxValue,
                 CancellationToken cancellationToken = default) where TOther : class
-                => _probe.ReceiveWhileAsync(max, idle, filter, msgs, cancellationToken);
+                => ReceiveWhileTask(Probe, max, idle, filter, msgCount, cancellationToken);
 
+            /// <summary>
+            /// Expect a publisher event from the stream.
+            /// </summary>
             public IPublisherEvent ExpectEvent(CancellationToken cancellationToken = default)
-                => ExpectEventAsync(cancellationToken)
+                => ExpectEventTask(Probe, cancellationToken)
                     .ConfigureAwait(false).GetAwaiter().GetResult();
 
+            /// <summary>
+            /// Expect a publisher event from the stream.
+            /// </summary>
             public async Task<IPublisherEvent> ExpectEventAsync(CancellationToken cancellationToken = default)
-                => await _probe.ExpectMsgAsync<IPublisherEvent>(cancellationToken: cancellationToken)
+                => await ExpectEventTask(Probe, cancellationToken)
                     .ConfigureAwait(false);
 
             /// <summary>
@@ -254,7 +237,7 @@ public async Task<TOther> WithinAsync<TOther>(
                 TimeSpan max,
                 Func<TOther> execute,
                 CancellationToken cancellationToken = default)
-                => await _probe.WithinAsync(min, max, execute, cancellationToken: cancellationToken)
+                => await Probe.WithinAsync(min, max, execute, cancellationToken: cancellationToken)
                     .ConfigureAwait(false);
             
             /// <summary>
@@ -278,7 +261,7 @@ public async Task<TOther> WithinAsync<TOther>(
             /// </summary>
             /// <param name="min"></param>
             /// <param name="max"></param>
-            /// <param name="execute"></param>
+            /// <param name="actionAsync"></param>
             /// <param name="cancellationToken"></param>
             /// <returns></returns>
             public async Task<TOther> WithinAsync<TOther>(
@@ -286,28 +269,28 @@ public async Task<TOther> WithinAsync<TOther>(
                 TimeSpan max,
                 Func<Task<TOther>> actionAsync,
                 CancellationToken cancellationToken = default)
-                => await _probe.WithinAsync(min, max, actionAsync, cancellationToken: cancellationToken)
+                => await Probe.WithinAsync(min, max, actionAsync, cancellationToken: cancellationToken)
                     .ConfigureAwait(false);
 
             /// <summary>
             /// Sane as calling Within(TimeSpan.Zero, max, function, cancellationToken).
             /// </summary>
-            public TOther Within<TOther>(TimeSpan max, Func<TOther> execute)
-                => WithinAsync(max, execute)
+            public TOther Within<TOther>(TimeSpan max, Func<TOther> execute, CancellationToken cancellationToken = default)
+                => WithinAsync(max, execute, cancellationToken)
                     .ConfigureAwait(false).GetAwaiter().GetResult();
             
             /// <summary>
             /// Sane as calling WithinAsync(TimeSpan.Zero, max, function, cancellationToken).
             /// </summary>
             public async Task<TOther> WithinAsync<TOther>(TimeSpan max, Func<TOther> execute, CancellationToken cancellationToken = default) 
-                => await _probe.WithinAsync(max, execute, cancellationToken: cancellationToken)
+                => await Probe.WithinAsync(max, execute, cancellationToken: cancellationToken)
                     .ConfigureAwait(false);
             
             /// <summary>
             /// Sane as calling WithinAsync(TimeSpan.Zero, max, function, cancellationToken).
             /// </summary>
             public async Task<TOther> WithinAsync<TOther>(TimeSpan max, Func<Task<TOther>> execute, CancellationToken cancellationToken = default) 
-                => await _probe.WithinAsync(max, execute, cancellationToken: cancellationToken)
+                => await Probe.WithinAsync(max, execute, cancellationToken: cancellationToken)
                     .ConfigureAwait(false);
         }
 
@@ -315,11 +298,10 @@ public async Task<TOther> WithinAsync<TOther>(TimeSpan max, Func<Task<TOther>> e
         /// Single subscription and demand tracking for <see cref="ManualProbe{T}"/>.
         /// </summary>
         /// <typeparam name="T"></typeparam>
-        public class Probe<T> : ManualProbe<T>
+        public partial class Probe<T> : ManualProbe<T>
         {
             private readonly long _initialPendingRequests;
-            private StreamTestKit.PublisherProbeSubscription<T> _subscription = null;
-
+            
             internal Probe(TestKitBase system, long initialPendingRequests) : base(system)
             {
                 _initialPendingRequests = Pending = initialPendingRequests;
@@ -334,71 +316,40 @@ internal Probe(TestKitBase system, long initialPendingRequests) : base(system)
             /// Asserts that a subscription has been received or will be received
             /// </summary>
             public void EnsureSubscription(CancellationToken cancellationToken = default)
-            {
-                if(_subscription == null)
-                    _subscription = ExpectSubscription(cancellationToken);
-            }
+                => EnsureSubscriptionTask(this, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
 
             public async Task EnsureSubscriptionAsync(CancellationToken cancellationToken = default)
-            {
-                if(_subscription == null)
-                    _subscription = await ExpectSubscriptionAsync(cancellationToken)
-                        .ConfigureAwait(false);
-            }
-            public Probe<T> SendNext(T element)
-            {
-                EnsureSubscription();
-                var sub = _subscription;
-                if (Pending == 0)
-                    Pending = sub.ExpectRequest();
-                Pending--;
-                sub.SendNext(element);
-                return this;
-            }
+                => await EnsureSubscriptionTask(this, cancellationToken)
+                    .ConfigureAwait(false);
 
-            public Probe<T> UnsafeSendNext(T element)
-            {
-                EnsureSubscription();
-                _subscription.SendNext(element);
-                return this;
-            }
+            public async Task SendNextAsync(T element, CancellationToken cancellationToken = default)
+                => await SendNextTask(this, element, cancellationToken)
+                    .ConfigureAwait(false);
+            
+            public async Task UnsafeSendNextAsync(T element, CancellationToken cancellationToken = default)
+                => await UnsafeSendNextTask(this, element, cancellationToken)
+                    .ConfigureAwait(false);
 
-            public Probe<T> SendComplete()
-            {
-                EnsureSubscription();
-                _subscription.SendComplete();
-                return this;
-            }
+            public async Task SendCompleteAsync(CancellationToken cancellationToken = default)
+                => await SendCompleteTask(this, cancellationToken)
+                    .ConfigureAwait(false);
 
-            public Probe<T> SendError(Exception e)
-            {
-                EnsureSubscription();
-                _subscription.SendError(e);
-                return this;
-            }
+            public async Task SendErrorAsync(Exception e, CancellationToken cancellationToken = default)
+                => await SendErrorTask(this, e, cancellationToken)
+                    .ConfigureAwait(false);
 
             public long ExpectRequest(CancellationToken cancellationToken = default)
-                => ExpectRequestAsync(cancellationToken)
+                => ExpectRequestTask(this, cancellationToken)
                     .ConfigureAwait(false).GetAwaiter().GetResult();
 
             public async Task<long> ExpectRequestAsync(CancellationToken cancellationToken = default)
-            {
-                await EnsureSubscriptionAsync(cancellationToken);
-                var requests = await _subscription.ExpectRequestAsync(cancellationToken)
+                => await ExpectRequestTask(this, cancellationToken)
                     .ConfigureAwait(false);
-                Pending += requests;
-                return requests;
-            }
 
-            public void ExpectCancellation(CancellationToken cancellationToken = default)
-                => ExpectCancellationAsync(cancellationToken)
-                    .ConfigureAwait(false).GetAwaiter().GetResult();
-            
             public async Task ExpectCancellationAsync(CancellationToken cancellationToken = default)
-            {
-                await EnsureSubscriptionAsync(cancellationToken);
-                await _subscription.ExpectCancellationAsync(cancellationToken);
-            }
+                => await ExpectCancellationTask(this, cancellationToken)
+                    .ConfigureAwait(false);
         }
 
         internal sealed class LazyEmptyPublisher<T> : IPublisher<T>
diff --git a/src/core/Akka.Streams.TestKit/TestPublisher_Fluent.cs b/src/core/Akka.Streams.TestKit/TestPublisher_Fluent.cs
new file mode 100644
index 00000000000..1542a6bd06f
--- /dev/null
+++ b/src/core/Akka.Streams.TestKit/TestPublisher_Fluent.cs
@@ -0,0 +1,113 @@
+// //-----------------------------------------------------------------------
+// // <copyright file="TestPublisher_Fluent.cs" company="Akka.NET Project">
+// //     Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
+// //     Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// // </copyright>
+// //-----------------------------------------------------------------------
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Reactive.Streams;
+
+namespace Akka.Streams.TestKit
+{
+    // Fluent implementation
+    public static partial class TestPublisher
+    {
+        public partial class ManualProbe<T>
+        {
+            /// <summary>
+            /// Fluent async DSL.
+            /// This will return an instance of <see cref="PublisherFluentBuilder{T}"/> that will compose and run
+            /// all of its method call asynchronously.
+            /// Note that <see cref="PublisherFluentBuilder{T}"/> contains two types of methods:
+            /// * Methods that returns <see cref="PublisherFluentBuilder{T}"/> are used to chain test methods together
+            ///   using a fluent builder pattern. 
+            /// * Methods with names that ends with the postfix "Async" and returns either a <see cref="Task"/> or
+            ///   a <see cref="Task{TResult}"/>. These methods invokes the previously chained methods asynchronously one
+            ///   after another before executing its own code.
+            /// </summary>
+            /// <returns></returns>
+            public PublisherFluentBuilder<T> AsyncBuilder()
+                => new PublisherFluentBuilder<T>(this);
+
+            /// <summary>
+            /// Fluent DSL
+            /// Expect demand from the given subscription.
+            /// </summary>
+            public ManualProbe<T> ExpectRequest(
+                ISubscription subscription,
+                int nrOfElements,
+                CancellationToken cancellationToken = default)
+            {
+                ExpectRequestTask(Probe, subscription, nrOfElements, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                return this;
+            }
+
+            /// <summary>
+            /// Fluent DSL
+            /// Expect no messages.
+            /// </summary>
+            public ManualProbe<T> ExpectNoMsg(CancellationToken cancellationToken = default)
+            {
+                ExpectNoMsgTask(Probe, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                return this;
+            }
+
+            /// <summary>
+            /// Fluent DSL
+            /// Expect no messages for given duration.
+            /// </summary>
+            public ManualProbe<T> ExpectNoMsg(TimeSpan duration, CancellationToken cancellationToken = default)
+            {
+                ExpectNoMsgTask(Probe, duration, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                return this;
+            }
+
+            
+        }
+        
+        public partial class Probe<T>
+        {
+            public Probe<T> SendNext(T element, CancellationToken cancellationToken = default)
+            {
+                SendNextTask(this, element, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                return this;
+            }
+
+            public Probe<T> UnsafeSendNext(T element, CancellationToken cancellationToken = default)
+            {
+                UnsafeSendNextTask(this, element, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                return this;
+            }
+
+            public Probe<T> SendComplete(CancellationToken cancellationToken = default)
+            {
+                SendCompleteTask(this, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                return this;
+            }
+
+            public Probe<T> SendError(Exception e, CancellationToken cancellationToken = default)
+            {
+                SendErrorTask(this, e, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                return this;
+            }
+            
+            public Probe<T> ExpectCancellation(CancellationToken cancellationToken = default)
+            {
+                ExpectCancellationTask(this, cancellationToken)
+                    .ConfigureAwait(false).GetAwaiter().GetResult();
+                return this;
+            }
+            
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/core/Akka.Streams.TestKit/TestPublisher_Shared.cs b/src/core/Akka.Streams.TestKit/TestPublisher_Shared.cs
new file mode 100644
index 00000000000..cedcb11c8a6
--- /dev/null
+++ b/src/core/Akka.Streams.TestKit/TestPublisher_Shared.cs
@@ -0,0 +1,141 @@
+// //-----------------------------------------------------------------------
+// // <copyright file="TestPublisher_Shared.cs" company="Akka.NET Project">
+// //     Copyright (C) 2009-2022 Lightbend Inc. <http://www.lightbend.com>
+// //     Copyright (C) 2013-2022 .NET Foundation <https://github.com/akkadotnet/akka.net>
+// // </copyright>
+// //-----------------------------------------------------------------------
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Akka.TestKit;
+using Reactive.Streams;
+
+namespace Akka.Streams.TestKit
+{
+    public static partial class TestPublisher
+    {
+        public partial class ManualProbe<T>
+        {
+            protected static async Task<StreamTestKit.PublisherProbeSubscription<T>> ExpectSubscriptionTask(
+                TestProbe probe,
+                CancellationToken cancellationToken)
+            {
+                var msg = await probe.ExpectMsgAsync<Subscribe>(cancellationToken: cancellationToken); 
+                return (StreamTestKit.PublisherProbeSubscription<T>) msg.Subscription;
+            }
+
+            #region void methods
+
+            internal static async Task ExpectRequestTask(
+                TestProbe probe,
+                ISubscription subscription, 
+                int nrOfElements,
+                CancellationToken cancellationToken)
+                => await probe.ExpectMsgAsync<RequestMore>(
+                    isMessage: x => x.NrOfElements == nrOfElements && x.Subscription == subscription, 
+                    cancellationToken: cancellationToken);
+            
+            internal static async Task ExpectNoMsgTask(TestProbe probe, CancellationToken cancellationToken)
+                => await probe.ExpectNoMsgAsync(cancellationToken);
+
+            internal static async Task ExpectNoMsgTask(
+                TestProbe probe,
+                TimeSpan duration,
+                CancellationToken cancellationToken)
+                => await probe.ExpectNoMsgAsync(duration, cancellationToken);
+
+            #endregion
+
+            #region Return type methods
+
+            internal static IAsyncEnumerable<TOther> ReceiveWhileTask<TOther>(
+                TestProbe probe,
+                TimeSpan? max,
+                TimeSpan? idle,
+                Func<object, TOther> filter, 
+                int msgCount,
+                CancellationToken cancellationToken) where TOther : class
+                => probe.ReceiveWhileAsync(max, idle, filter, msgCount, cancellationToken);
+
+            internal static ValueTask<IPublisherEvent> ExpectEventTask(
+                TestProbe probe,
+                CancellationToken cancellationToken)
+                => probe.ExpectMsgAsync<IPublisherEvent>(cancellationToken: cancellationToken);
+
+            #endregion
+        }
+        
+        public partial class Probe<T>
+        {
+            #region void methods
+
+            internal static async Task EnsureSubscriptionTask(Probe<T> probe, CancellationToken cancellationToken)
+            {
+                probe.Subscription ??= await ExpectSubscriptionTask(probe.Probe, cancellationToken);
+            }
+            
+            internal static async Task SendNextTask(Probe<T> probe, T element, CancellationToken cancellationToken)
+            {
+                await EnsureSubscriptionTask(probe, cancellationToken);
+                var sub = probe.Subscription;
+                if (probe.Pending == 0)
+                    probe.Pending = await sub.ExpectRequestAsync(cancellationToken);
+                probe.Pending--;
+                sub.SendNext(element);
+            }
+
+            internal static async Task SendNextTask(Probe<T> probe, IEnumerable<T> elements, CancellationToken cancellationToken)
+            {
+                await EnsureSubscriptionTask(probe, cancellationToken);
+                var sub = probe.Subscription;
+                foreach (var element in elements)
+                {
+                    if (probe.Pending == 0)
+                        probe.Pending = await sub.ExpectRequestAsync(cancellationToken);
+                    probe.Pending--;
+                    sub.SendNext(element);
+                }
+            }
+            
+            internal static async Task UnsafeSendNextTask(Probe<T> probe, T element, CancellationToken cancellationToken)
+            {
+                await EnsureSubscriptionTask(probe, cancellationToken);
+                probe.Subscription.SendNext(element);
+            }
+
+            internal static async Task SendCompleteTask(Probe<T> probe, CancellationToken cancellationToken)
+            {
+                await EnsureSubscriptionTask(probe, cancellationToken);
+                probe.Subscription.SendComplete();
+            }
+            
+            internal static async Task SendErrorTask(Probe<T> probe, Exception e, CancellationToken cancellationToken)
+            {
+                await EnsureSubscriptionTask(probe, cancellationToken);
+                probe.Subscription.SendError(e);
+            }
+            
+            #endregion
+
+            #region Return type methods
+
+            internal static async Task<long> ExpectRequestTask(Probe<T> probe, CancellationToken cancellationToken)
+            {
+                await EnsureSubscriptionTask(probe, cancellationToken);
+                var requests = await probe.Subscription.ExpectRequestAsync(cancellationToken);
+                probe.Pending += requests;
+                return requests;
+            }
+
+            internal static async Task ExpectCancellationTask(Probe<T> probe, CancellationToken cancellationToken = default)
+            {
+                await EnsureSubscriptionTask(probe, cancellationToken);
+                await probe.Subscription.ExpectCancellationAsync(cancellationToken);
+            }
+
+            #endregion
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/core/Akka.Streams.TestKit/TestSubscriber.cs b/src/core/Akka.Streams.TestKit/TestSubscriber.cs
index 29ccb15c938..7968b79aa46 100644
--- a/src/core/Akka.Streams.TestKit/TestSubscriber.cs
+++ b/src/core/Akka.Streams.TestKit/TestSubscriber.cs
@@ -92,7 +92,7 @@ public partial class ManualProbe<T> : ISubscriber<T>
         {
             private readonly TestKitBase _testKit;
             internal readonly TestProbe TestProbe;
-            private volatile ISubscription _subscription;
+            private volatile ISubscription _subscription_DoNotUseDirectly;
 
             internal ManualProbe(TestKitBase testKit)
             {
@@ -103,8 +103,8 @@ internal ManualProbe(TestKitBase testKit)
             public ISubscription Subscription
             {
 #pragma warning disable CS0420
-                get => Volatile.Read(ref _subscription);
-                internal set => Volatile.Write(ref _subscription, value);
+                get => Volatile.Read(ref _subscription_DoNotUseDirectly);
+                protected set => Volatile.Write(ref _subscription_DoNotUseDirectly, value);
 #pragma warning restore CS0420
             }
 
@@ -188,6 +188,10 @@ public async Task<T> ExpectNextAsync(TimeSpan? timeout, CancellationToken cancel
                 => await ExpectNextTask(TestProbe, timeout, cancellationToken)
                     .ConfigureAwait(false);
 
+            public async Task ExpectNextAsync(T element, CancellationToken cancellationToken = default)
+                => await ExpectNextTask(probe: TestProbe, element: element, timeout: null, cancellationToken: cancellationToken)
+                    .ConfigureAwait(false);
+            
             /// <summary>
             /// Expect and return the next <paramref name="n"/> stream elements.
             /// </summary>
@@ -459,32 +463,27 @@ public Probe<T> EnsureSubscription(CancellationToken cancellationToken = default
             /// <summary>
             /// Ensure that the probe has received or will receive a subscription
             /// </summary>
-            public async Task<Probe<T>> EnsureSubscriptionAsync(CancellationToken cancellationToken = default)
-            {
-                await EnsureSubscriptionTask(this, cancellationToken);
-                return this;
-            }
+            public async Task EnsureSubscriptionAsync(CancellationToken cancellationToken = default)
+                => await EnsureSubscriptionTask(this, cancellationToken)
+                    .ConfigureAwait(false);
 
-            public Probe<T> Request(long n)
+            public async Task RequestAsync(long n)
             {
-                EnsureSubscription();
+                await EnsureSubscriptionAsync();
                 Subscription.Request(n);
-                return this;
             }
 
-            public Probe<T> RequestNext(T element)
+            public async Task RequestNextAsync(T element)
             {
-                EnsureSubscription();
+                await EnsureSubscriptionAsync();
                 Subscription.Request(1);
-                ExpectNext(element);
-                return this;
+                await ExpectNextAsync(element);
             }
 
-            public Probe<T> Cancel()
+            public async Task CancelAsync()
             {
-                EnsureSubscription();
+                await EnsureSubscriptionAsync();
                 Subscription.Cancel();
-                return this;
             }
 
             /// <summary>
@@ -497,6 +496,16 @@ public T RequestNext()
                 return ExpectNext();
             }
 
+            /// <summary>
+            /// Request and expect a stream element.
+            /// </summary>
+            public async Task<T> RequestNextAsync()
+            {
+                await EnsureSubscriptionAsync();
+                Subscription.Request(1);
+                return await ExpectNextAsync();
+            }
+
             /// <summary>
             /// Request and expect a stream element during the specified time or timeout.
             /// </summary>
@@ -506,6 +515,14 @@ public T RequestNext(TimeSpan timeout)
                 Subscription.Request(1);
                 return ExpectNext(timeout);
             }
+            
+            public async Task<T> RequestNextAsync(TimeSpan timeout)
+            {
+                await EnsureSubscriptionAsync();
+                Subscription.Request(1);
+                return await ExpectNextAsync(timeout);
+            }
+            
         }
 
         public static ManualProbe<T> CreateManualSubscriberProbe<T>(this TestKitBase testKit)
diff --git a/src/core/Akka.Streams.TestKit/TestSubscriber_Fluent.cs b/src/core/Akka.Streams.TestKit/TestSubscriber_Fluent.cs
index 446b3227265..8af11c806e3 100644
--- a/src/core/Akka.Streams.TestKit/TestSubscriber_Fluent.cs
+++ b/src/core/Akka.Streams.TestKit/TestSubscriber_Fluent.cs
@@ -188,6 +188,9 @@ public ManualProbe<T> ExpectComplete(CancellationToken cancellationToken = defau
                 return this;
             }
             
+            public async Task ExpectCompleteAsync(CancellationToken cancellationToken = default)
+                => await ExpectCompleteTask(TestProbe, null, cancellationToken);
+            
             /// <summary>
             /// Fluent DSL. Expect completion with a timeout.
             /// </summary>
@@ -276,5 +279,32 @@ public ManualProbe<T> MatchNext<TOther>(Predicate<TOther> predicate, Cancellatio
                 return this;                
             }
         }
+
+        public partial class Probe<T>
+        {
+            public Probe<T> Request(long n)
+            {
+                EnsureSubscription();
+                Subscription.Request(n);
+                return this;
+            }
+
+            public Probe<T> Cancel()
+            {
+                EnsureSubscription();
+                Subscription.Cancel();
+                return this;
+            }
+
+            public Probe<T> RequestNext(T element)
+            {
+                EnsureSubscription();
+                Subscription.Request(1);
+                ExpectNext(element);
+                return this;
+            }
+
+
+        }
     }
 }
\ No newline at end of file
diff --git a/src/core/Akka.Streams.TestKit/TestSubscriber_Shared.cs b/src/core/Akka.Streams.TestKit/TestSubscriber_Shared.cs
index 59a5f6f23a0..c707eb289b5 100644
--- a/src/core/Akka.Streams.TestKit/TestSubscriber_Shared.cs
+++ b/src/core/Akka.Streams.TestKit/TestSubscriber_Shared.cs
@@ -305,11 +305,11 @@ internal static async IAsyncEnumerable<T> ToStrictTask(
                 TimeSpan atMost,
                 [EnumeratorCancellation] CancellationToken cancellationToken = default)
             {
-                var deadline = DateTime.UtcNow + atMost;
                 // if no subscription was obtained yet, we expect it
                 await EnsureSubscriptionTask(probe, cancellationToken);
                 probe.Subscription.Request(long.MaxValue);
 
+                var deadline = DateTime.UtcNow + atMost;
                 var result = new List<T>();
                 while (true)
                 {
@@ -330,6 +330,8 @@ internal static async IAsyncEnumerable<T> ToStrictTask(
                             result.Add(next.Element);
                             yield return next.Element;
                             break;
+                        default:
+                            throw new InvalidOperationException($"Invalid response, expected {nameof(OnError)}, {nameof(OnComplete)}, or {nameof(OnNext<T>)}, received [{e.GetType()}]");
                     }
                 }
             }
diff --git a/src/core/Akka.Streams.TestKit/Utils.cs b/src/core/Akka.Streams.TestKit/Utils.cs
index c4942138bb7..5a2e457f34a 100644
--- a/src/core/Akka.Streams.TestKit/Utils.cs
+++ b/src/core/Akka.Streams.TestKit/Utils.cs
@@ -41,13 +41,13 @@ public static async Task<T> AssertAllStagesStoppedAsync<T>(this AkkaSpec spec, F
         
         public static async Task<T> AssertAllStagesStoppedAsync<T>(this AkkaSpec spec, Func<Task<T>> block, IMaterializer materializer)
         {
+            var result = await block();
             if (!(materializer is ActorMaterializerImpl impl))
-                return await block();
+                return result;
 
             var probe = spec.CreateTestProbe(impl.System);
             probe.Send(impl.Supervisor, StreamSupervisor.StopChildren.Instance);
             await probe.ExpectMsgAsync<StreamSupervisor.StoppedChildren>();
-            var result = await block();
 
             await probe.WithinAsync(TimeSpan.FromSeconds(5), async () =>
             {

From 489af63dd0eb9158e246817dee0fb333089012c9 Mon Sep 17 00:00:00 2001
From: Gregorius Soedharmo <arkatufus@yahoo.com>
Date: Tue, 10 May 2022 21:17:22 +0700
Subject: [PATCH 2/2] Revert build.fsx changes

---
 build.fsx | 67 +++++++++++++++++++------------------------------------
 1 file changed, 23 insertions(+), 44 deletions(-)

diff --git a/build.fsx b/build.fsx
index f78ef249ef7..83350b7cabb 100644
--- a/build.fsx
+++ b/build.fsx
@@ -241,34 +241,27 @@ Target "RunTests" (fun _ ->
                             | true -> !! "./src/**/*.Tests.*sproj"
                                       ++ "./src/**/Akka.Streams.Tests.TCK.csproj"
                                       -- "./src/**/*.Tests.MultiNode.csproj"
-                                      -- "./src/**/*.Tests.Performance.csproj"
                                       -- "./src/examples/**"
                             | _ -> !! "./src/**/*.Tests.*sproj" // if you need to filter specs for Linux vs. Windows, do it here
                                    -- "./src/**/*.Tests.MultiNode.csproj"
-                                   -- "./src/**/*.Tests.Performance.csproj"
                                    -- "./src/examples/**"
         rawProjects |> Seq.choose filterProjects
 
-    let projectDlls = projects |> Seq.map ( fun project ->
-            let assemblyName = fileNameWithoutExt project
-            (directory project) @@ "bin" @@ "Release" @@ testNetFrameworkVersion @@ assemblyName + ".dll" 
-        )
-
-    let runSingleProject projectDll =
+    let runSingleProject project =
         let arguments =
             match (hasTeamCity) with
-            | true -> (sprintf "test \"%s\" --blame-crash --blame-hang-timeout 25m -l:trx -l:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -teamcity" projectDll testNetFrameworkVersion outputTests)
-            | false -> (sprintf "test \"%s\" --blame-crash --blame-hang-timeout 25m -l:trx -l:\"console;verbosity=normal\" --framework %s --results-directory \"%s\"" projectDll testNetFrameworkVersion outputTests)
-            
+            | true -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetFrameworkVersion outputTests)
+            | false -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetFrameworkVersion outputTests)
+
         let result = ExecProcess(fun info ->
             info.FileName <- "dotnet"
-            info.WorkingDirectory <- outputTests
+            info.WorkingDirectory <- (Directory.GetParent project).FullName
             info.Arguments <- arguments) (TimeSpan.FromMinutes 30.0)
 
         ResultHandling.failBuildIfXUnitReportedError TestRunnerErrorLevel.Error result
-        
+
     CreateDir outputTests
-    projectDlls |> Seq.iter (runSingleProject)
+    projects |> Seq.iter (runSingleProject)
 )
 
 Target "RunTestsNetCore" (fun _ ->
@@ -278,34 +271,27 @@ Target "RunTestsNetCore" (fun _ ->
                                 | true -> !! "./src/**/*.Tests.*sproj"
                                           ++ "./src/**/Akka.Streams.Tests.TCK.csproj"
                                           -- "./src/**/*.Tests.MultiNode.csproj"
-                                          -- "./src/**/*.Tests.Performance.csproj"
                                           -- "./src/examples/**"
                                 | _ -> !! "./src/**/*.Tests.*sproj" // if you need to filter specs for Linux vs. Windows, do it here
                                        -- "./src/**/*.Tests.MultiNode.csproj"
-                                       -- "./src/**/*.Tests.Performance.csproj"
                                        -- "./src/examples/**"
             rawProjects |> Seq.choose filterProjects
 
-        let projectDlls = projects |> Seq.map ( fun project ->
-                let assemblyName = fileNameWithoutExt project
-                (directory project) @@ "bin" @@ "Release" @@ testNetCoreVersion @@ assemblyName + ".dll" 
-            )
-
-        let runSingleProject projectDll =
+        let runSingleProject project =
             let arguments =
                 match (hasTeamCity) with
-                | true -> (sprintf "test \"%s\" --blame-crash --blame-hang-timeout 25m -l:trx -l:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -teamcity" projectDll testNetCoreVersion outputTests)
-                | false -> (sprintf "test \"%s\" --blame-crash --blame-hang-timeout 25m -l:trx -l:\"console;verbosity=normal\" --framework %s --results-directory \"%s\"" projectDll testNetCoreVersion outputTests)
+                | true -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetCoreVersion outputTests)
+                | false -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetCoreVersion outputTests)
 
             let result = ExecProcess(fun info ->
                 info.FileName <- "dotnet"
-                info.WorkingDirectory <- outputTests
+                info.WorkingDirectory <- (Directory.GetParent project).FullName
                 info.Arguments <- arguments) (TimeSpan.FromMinutes 30.0)
 
             ResultHandling.failBuildIfXUnitReportedError TestRunnerErrorLevel.Error result
 
         CreateDir outputTests
-        projectDlls |> Seq.iter (runSingleProject)
+        projects |> Seq.iter (runSingleProject)
 )
 
 Target "RunTestsNet" (fun _ ->
@@ -315,34 +301,27 @@ Target "RunTestsNet" (fun _ ->
                                 | true -> !! "./src/**/*.Tests.*sproj"
                                           ++ "./src/**/Akka.Streams.Tests.TCK.csproj"
                                           -- "./src/**/*.Tests.MultiNode.csproj"
-                                          -- "./src/**/*.Tests.Performance.csproj"
                                           -- "./src/examples/**"
                                 | _ -> !! "./src/**/*.Tests.*sproj" // if you need to filter specs for Linux vs. Windows, do it here
                                        -- "./src/**/*.Tests.MultiNode.csproj"
-                                       -- "./src/**/*.Tests.Performance.csproj"
                                        -- "./src/examples/**"
             rawProjects |> Seq.choose filterProjects
 
-        let projectDlls = projects |> Seq.map ( fun project ->
-                let assemblyName = fileNameWithoutExt project
-                (directory project) @@ "bin" @@ "Release" @@ testNetVersion @@ assemblyName + ".dll" 
-            )
-        
-        let runSingleProject projectDll =
+        let runSingleProject project =
             let arguments =
                 match (hasTeamCity) with
-                | true -> (sprintf "test \"%s\" --blame-crash --blame-hang-timeout 25m -l:trx -l:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -teamcity" projectDll testNetVersion outputTests)
-                | false -> (sprintf "test \"%s\" --blame-crash --blame-hang-timeout 25m -l:trx -l:\"console;verbosity=normal\" --framework %s --results-directory \"%s\"" projectDll testNetVersion outputTests)
+                | true -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none -teamcity" testNetVersion outputTests)
+                | false -> (sprintf "test -c Release --no-build --logger:trx --logger:\"console;verbosity=normal\" --framework %s --results-directory \"%s\" -- -parallel none" testNetVersion outputTests)
 
             let result = ExecProcess(fun info ->
                 info.FileName <- "dotnet"
-                info.WorkingDirectory <- outputTests
+                info.WorkingDirectory <- (Directory.GetParent project).FullName
                 info.Arguments <- arguments) (TimeSpan.FromMinutes 30.0)
 
             ResultHandling.failBuildIfXUnitReportedError TestRunnerErrorLevel.Error result
 
         CreateDir outputTests
-        projectDlls |> Seq.iter (runSingleProject)
+        projects |> Seq.iter (runSingleProject)
 )
 
 Target "MultiNodeTestsNetCore" (fun _ ->
@@ -363,13 +342,13 @@ Target "MultiNodeTestsNetCore" (fun _ ->
         let runSingleProject projectDll =
             let arguments =
                 match (hasTeamCity) with
-                | true -> (sprintf "test \"%s\" -l:trx -l:\"console;verbosity=detailed\" --framework %s --results-directory \"%s\" -- -teamcity" projectDll testNetCoreVersion outputMultiNode)
-                | false -> (sprintf "test \"%s\" -l:trx -l:\"console;verbosity=detailed\" --framework %s --results-directory \"%s\"" projectDll testNetCoreVersion outputMultiNode)
+                | true -> (sprintf "test \"%s\" -l:\"console;verbosity=detailed\" --framework %s --results-directory \"%s\" -- -teamcity" projectDll testNetCoreVersion outputMultiNode)
+                | false -> (sprintf "test \"%s\" -l:\"console;verbosity=detailed\" --framework %s --results-directory \"%s\"" projectDll testNetCoreVersion outputMultiNode)
 
             let resultPath = (directory projectDll)
             File.WriteAllText(
                 (resultPath @@ "xunit.multinode.runner.json"),
-                (sprintf "{\"outputDirectory\":\"%s\"}" outputMultiNode).Replace("\\", "\\\\"))
+                (sprintf "{\"outputDirectory\":\"%s\", \"useBuiltInTrxReporter\":true}" outputMultiNode).Replace("\\", "\\\\"))
             
             let result = ExecProcess(fun info ->
                 info.FileName <- "dotnet"
@@ -402,13 +381,13 @@ Target "MultiNodeTestsNet" (fun _ ->
         let runSingleProject projectDll =
             let arguments =
                 match (hasTeamCity) with
-                | true -> (sprintf "test \"%s\" -l:trx -l:\"console;verbosity=detailed\" --framework %s --results-directory \"%s\" -- -teamcity" projectDll testNetVersion outputMultiNode)
-                | false -> (sprintf "test \"%s\" -l:trx -l:\"console;verbosity=detailed\" --framework %s --results-directory \"%s\"" projectDll testNetVersion outputMultiNode)
+                | true -> (sprintf "test \"%s\" -l:\"console;verbosity=detailed\" --framework %s --results-directory \"%s\" -- -teamcity" projectDll testNetVersion outputMultiNode)
+                | false -> (sprintf "test \"%s\" -l:\"console;verbosity=detailed\" --framework %s --results-directory \"%s\"" projectDll testNetVersion outputMultiNode)
 
             let resultPath = (directory projectDll)
             File.WriteAllText(
                 (resultPath @@ "xunit.multinode.runner.json"),
-                (sprintf "{\"outputDirectory\":\"%s\"}" outputMultiNode).Replace("\\", "\\\\"))
+                (sprintf "{\"outputDirectory\":\"%s\", \"useBuiltInTrxReporter\":true}" outputMultiNode).Replace("\\", "\\\\"))
             
             let result = ExecProcess(fun info ->
                 info.FileName <- "dotnet"