[CS1998] This async method lacks 'await' operators and will run synchronously (#6513)

* This async method lacks 'await' operators and will run synchronously

* removed `async`/`await`

* [improves] - `AsyncEnumerable.cs`
- `TestPublisher.cs`
- `Utils.cs`
- `EventFilterApplier.cs`
- `BlockingQueue.cs`
- `TestKitBase_AwaitConditions.cs`
- `TestKitBase_Within.cs`

* Fix `CS1998`

* changed it `Task.FromResult` to `Task.Run`

Author: eaba

src/contrib/cluster/Akka.Cluster.Metrics.Tests.MultiNode/ClusterMetricsExtensionSpec.cs

@@ -109,11 +109,11 @@ await AwaitAssertAsync(() =>
                 MetricsView.ClusterMetrics.Count.Should().Be(Roles.Count);
             }, TimeSpan.FromSeconds(30));
 
-            await WithinAsync(10.Seconds(), async () =>
-            {
+            await WithinAsync(10.Seconds(), () => {
                 var collector = new MetricsCollectorBuilder().Build(Cluster.System);
                 collector.Sample().Metrics.Count.Should().BeGreaterThan(3);
                 EnterBarrier("after");
+                return Task.CompletedTask;
             });
         }
 

src/contrib/cluster/Akka.Cluster.Metrics.Tests.MultiNode/ClustetMetricsRoutingSpec.cs

@@ -246,11 +246,11 @@ private async Task Should_prefer_node_with_more_free_memory_capacity()
 
             await RunOnAsync(async () =>
             {
-                await WithinAsync(20.Seconds(), async () =>
-                {
-                   Sys.ActorOf(Props.Create<AdaptiveLoadBalancingRouterConfig.MemoryAllocator>(), "memory-allocator")
-                       .Tell(AdaptiveLoadBalancingRouterConfig.AllocateMemory.Instance);
-                   ExpectMsg("done");
+                await WithinAsync(20.Seconds(), () => {
+                    Sys.ActorOf(Props.Create<AdaptiveLoadBalancingRouterConfig.MemoryAllocator>(), "memory-allocator")
+                                                                             .Tell(AdaptiveLoadBalancingRouterConfig.AllocateMemory.Instance);
+                    ExpectMsg("done");
+                    return Task.CompletedTask;
                 });
             }, _config.Node2);
 

src/contrib/cluster/Akka.Cluster.Metrics.Tests.MultiNode/Sample/StatsSampleSpec.cs

@@ -88,27 +88,27 @@ public async Task Stats_sample_should_illustrate_how_to_startup_cluster()
 
         private async Task Should_startup_cluster()
         {
-            await WithinAsync(15.Seconds(), async () =>
-            {
+            await WithinAsync(15.Seconds(), () => {
                 var cluster = Cluster.Get(Sys);
                 cluster.Subscribe(TestActor, typeof(ClusterEvent.MemberUp));
                 ExpectMsg<ClusterEvent.CurrentClusterState>();
 
                 var firstAddress = Node(_config.First).Address;
                 var secondAddress = Node(_config.Second).Address;
                 var thirdAddress = Node(_config.Third).Address;
-                
+
                 cluster.Join(firstAddress);
 
                 Sys.ActorOf(Props.Create<StatsWorker>(), "statsWorker");
                 Sys.ActorOf(Props.Create<StatsService>(), "statsService");
 
                 ReceiveN(3).Select(m => (m as ClusterEvent.MemberUp).Member.Address).Distinct()
                     .Should().BeEquivalentTo(firstAddress, secondAddress, thirdAddress);
-                
+
                 cluster.Unsubscribe(TestActor);
-                
+
                 EnterBarrier("all-up");
+                return Task.CompletedTask;
             });
         }
 

src/contrib/cluster/Akka.Cluster.Tools.Tests/PublishSubscribe/DistributedPubSubMediatorSpec.cs

@@ -80,7 +80,7 @@ public async Task DistributedPubSubMediator_should_send_messages_to_dead_letter(
 
             // assert
             await EventFilter.DeadLetter<object>().ExpectAsync(1,
-                async () => { mediator.Tell(new Publish("pub-sub", "hit")); });
+                () => { mediator.Tell(new Publish("pub-sub", "hit")); return Task.CompletedTask; });
         }
     }
 

src/contrib/cluster/Akka.Cluster.Tools.Tests/Singleton/ClusterSingletonProxySpec.cs

@@ -53,7 +53,7 @@ public async Task ClusterSingletonProxy_with_zero_buffering_should_work()
 
             // have to wait for cluster singleton to be ready, otherwise message will be rejected
             await AwaitConditionAsync(
-                async () => Cluster.Get(testSystem.Sys).State.Members.Count(m => m.Status == MemberStatus.Up) == 2,
+                () => Task.FromResult(Cluster.Get(testSystem.Sys).State.Members.Count(m => m.Status == MemberStatus.Up) == 2),
                 TimeSpan.FromSeconds(30));
 
             try

src/contrib/cluster/Akka.DistributedData.Tests/ReplicatorSpecs.cs

@@ -173,7 +173,7 @@ public async Task Bugfix_Duplicate_Publish()
             await ReplicatorDuplicatePublish();
         }
 
-        private async Task ReplicatorDuplicatePublish()
+        private Task ReplicatorDuplicatePublish()
         {
             var p1 = CreateTestProbe(_sys1);
             var p2 = CreateTestProbe(_sys2);
@@ -205,6 +205,7 @@ private async Task ReplicatorDuplicatePublish()
 
             // no probe should receive an update
             p2.ExpectNoMsg(TimeSpan.FromSeconds(1));
+            return Task.CompletedTask;
         }
 
         /// <summary>
@@ -217,7 +218,7 @@ public async Task Bugfix_4198_PNCounterDictionary_Merge()
             await PNCounterDictionary_Should_Merge();
         }
 
-        private async Task PNCounterDictionary_Should_Merge()
+        private Task PNCounterDictionary_Should_Merge()
         {
             var p1 = CreateTestProbe(_sys1);
             var p2 = CreateTestProbe(_sys2);
@@ -262,6 +263,7 @@ private async Task PNCounterDictionary_Should_Merge()
             });
 
             Sys.Log.Info("Done");
+            return Task.CompletedTask;
         }
 
         /// <summary>
@@ -366,7 +368,7 @@ public async Task Bugfix_4302_ORMultiValueDictionary_Merge()
             await ORMultiValueDictionary_Should_Merge();
         }
 
-        private async Task ORMultiValueDictionary_Should_Merge()
+        private Task ORMultiValueDictionary_Should_Merge()
         {
             var changedProbe = CreateTestProbe(_sys2);
 
@@ -475,6 +477,7 @@ private async Task ORMultiValueDictionary_Should_Merge()
                         changedProbe.ExpectMsg<Changed>(g => Equals(g.Key, _keyJ)).Get(_keyJ).Entries);
                 });
             });
+            return Task.CompletedTask;
         }
 
         private void VerifyMultiValueDictionaryEntries(

src/core/Akka.Cluster.Tests/ClusterLogSpec.cs

@@ -52,7 +52,7 @@ protected async Task AwaitUpAsync()
         {
             await WithinAsync(TimeSpan.FromSeconds(10), async() =>
             {
-                await AwaitConditionAsync(async () => ClusterView.IsSingletonCluster);
+                await AwaitConditionAsync(() => Task.FromResult(ClusterView.IsSingletonCluster));
                 ClusterView.Self.Address.ShouldBe(_selfAddress);
                 ClusterView.Members.Select(m => m.Address).ShouldBe(new Address[] { _selfAddress });
                 await AwaitAssertAsync(() => ClusterView.Status.ShouldBe(MemberStatus.Up));

src/core/Akka.Cluster.Tests/ClusterSpec.cs

@@ -80,7 +80,7 @@ public async Task A_cluster_must_initially_become_singleton_cluster_when_joining
             ClusterView.Members.Count.Should().Be(0);
             _cluster.Join(_selfAddress);
             LeaderActions(); // Joining -> Up
-            await AwaitConditionAsync(async () => ClusterView.IsSingletonCluster);
+            await AwaitConditionAsync(() => Task.FromResult(ClusterView.IsSingletonCluster));
             ClusterView.Self.Address.Should().Be(_selfAddress);
             ClusterView.Members.Select(m => m.Address).ToImmutableHashSet()
                 .Should().BeEquivalentTo(ImmutableHashSet.Create(_selfAddress));
@@ -259,7 +259,7 @@ public async Task A_cluster_must_cancel_LeaveAsync_task_if_CancellationToken_fir
 
             // Cancelling the first task
             cts.Cancel();
-            await AwaitConditionAsync(async () => task1.IsCanceled, null, "Task should be cancelled");
+            await AwaitConditionAsync(() => Task.FromResult(task1.IsCanceled), null, "Task should be cancelled");
 
             await WithinAsync(TimeSpan.FromSeconds(10), async () =>
             {
@@ -274,12 +274,12 @@ await WithinAsync(TimeSpan.FromSeconds(10), async () =>
                 ExpectMsg<ClusterEvent.MemberRemoved>().Member.Address.Should().Be(_selfAddress);
 
                 // Second task should complete (not cancelled)
-                await AwaitConditionAsync(async () => task2.IsCompleted && !task2.IsCanceled, null, "Task should be completed, but not cancelled.");
+                await AwaitConditionAsync(() => Task.FromResult(task2.IsCompleted && !task2.IsCanceled), null, "Task should be completed, but not cancelled.");
             }, cancellationToken: cts.Token);
 
             // Subsequent LeaveAsync() tasks expected to complete immediately (not cancelled)
             var task3 = _cluster.LeaveAsync();
-            await AwaitConditionAsync(async () => task3.IsCompleted && !task3.IsCanceled, null, "Task should be completed, but not cancelled.");
+            await AwaitConditionAsync(() => Task.FromResult(task3.IsCompleted && !task3.IsCanceled), null, "Task should be completed, but not cancelled.");
         }
 
         [Fact]

src/core/Akka.Docs.Tests/Actors/ReceiveTimeoutSpecs.cs

@@ -47,7 +47,7 @@ protected override void PreStart()
         // </ReceiveTimeoutActor>
 
         [Fact]
-        public async Task ShouldReceiveTimeoutActors()
+        public Task ShouldReceiveTimeoutActors()
         {
             var receiveTimeout = Sys.ActorOf(
                 Props.Create(() => new ReceiveTimeoutActor(TimeSpan.FromMilliseconds(100), TestActor)), 
@@ -58,6 +58,7 @@ public async Task ShouldReceiveTimeoutActors()
 
             // then should receive timeout due to inactivity
             ExpectMsg("timeout", TimeSpan.FromSeconds(30));
+            return Task.CompletedTask;
         }
     }
 }

src/core/Akka.Persistence.Tests/ReceivePersistentActorAsyncAwaitSpec.cs

@@ -533,9 +533,9 @@ public AsyncFailingActor(string persistenceId)
 
                 RecoverAny(o => { });
 
-                CommandAsync<string>(async m =>
-                {
+                CommandAsync<string>(m => {
                     ThrowException();
+                    return Task.CompletedTask;
                 });
             }
 
@@ -599,8 +599,7 @@ public AsyncReentrantActor(string persistenceId)
 
                 RecoverAny(o => { });
 
-                CommandAsync<string>(async msg =>
-                {
+                CommandAsync<string>(msg => {
                     var sender = Sender;
 #pragma warning disable CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
                     Task.Run(() =>
@@ -612,6 +611,7 @@ public AsyncReentrantActor(string persistenceId)
 #pragma warning restore CS4014 // Because this call is not awaited, execution of the current method continues before the call is completed
 
                     Thread.Sleep(3000);
+                    return Task.CompletedTask;
                 });
             }
         }
@@ -635,7 +635,7 @@ public void Actor_PipeTo_should_not_be_delayed_by_async_receive()
         }
 
         [Fact]
-        public async Task Actor_receiveasync_overloads_should_work()
+        public Task Actor_receiveasync_overloads_should_work()
         {
             var actor = Sys.ActorOf(Props.Create(() => new AsyncAwaitActor("pid")));
 
@@ -647,8 +647,7 @@ public async Task Actor_receiveasync_overloads_should_work()
 
             actor.Tell(1.0);
             ExpectMsg<string>(m => "handled".Equals(m), TimeSpan.FromMilliseconds(1000));
-
-
+            return Task.CompletedTask;
         }
     }
 }

src/core/Akka.Persistence/Journal/MemoryJournal.cs

@@ -182,20 +182,20 @@ protected override bool ReceivePluginInternal(object message)
             }
         }
 
-        private async Task<(IEnumerable<string> Ids, int LastOrdering)> SelectAllPersistenceIdsAsync(int offset)
+        private Task<(IEnumerable<string> Ids, int LastOrdering)> SelectAllPersistenceIdsAsync(int offset)
         {
-            return (new HashSet<string>(_allMessages.Skip(offset).Select(p => p.PersistenceId)), _allMessages.Count); 
+            return Task.FromResult<(IEnumerable<string> Ids, int LastOrdering)>((new HashSet<string>(_allMessages.Skip(offset).Select(p => p.PersistenceId)), _allMessages.Count)); 
         }
 
         /// <summary>
         /// Replays all events with given tag withing provided boundaries from memory.
         /// </summary>
         /// <param name="replay">TBD</param>
         /// <returns>TBD</returns>
-        private async Task<int> ReplayTaggedMessagesAsync(ReplayTaggedMessages replay)
+        private Task<int> ReplayTaggedMessagesAsync(ReplayTaggedMessages replay)
         {
             if (!_tagsToMessagesMapping.ContainsKey(replay.Tag))
-                return 0;
+                return Task.FromResult(0);
 
             int index = 0;
             foreach (var persistence in _tagsToMessagesMapping[replay.Tag]
@@ -207,10 +207,10 @@ private async Task<int> ReplayTaggedMessagesAsync(ReplayTaggedMessages replay)
                 index++;
             }
 
-            return _tagsToMessagesMapping[replay.Tag].Count - 1;
+            return Task.FromResult(_tagsToMessagesMapping[replay.Tag].Count - 1);
         }
 
-        private async Task<int> ReplayAllEventsAsync(ReplayAllEvents replay)
+        private Task<int> ReplayAllEventsAsync(ReplayAllEvents replay)
         {
             int index = 0;
             var replayed = _allMessages
@@ -222,8 +222,7 @@ private async Task<int> ReplayAllEventsAsync(ReplayAllEvents replay)
                 replay.ReplyTo.Tell(new ReplayedEvent(message, replay.FromOffset + index), ActorRefs.NoSender);
                 index++;
             }
-
-            return _allMessages.Count - 1;
+            return Task.FromResult(_allMessages.Count - 1);
         }
 
         #region QueryAPI
@@ -439,7 +438,6 @@ public sealed class ReplayedEvent : INoSerializationVerificationNeeded, IDeadLet
             /// TBD
             /// </summary>
             /// <param name="persistent">TBD</param>
-            /// <param name="tag">TBD</param>
             /// <param name="offset">TBD</param>
             public ReplayedEvent(IPersistentRepresentation persistent, int offset)
             {

src/core/Akka.Remote.Tests/ActorsLeakSpec.cs

@@ -234,7 +234,7 @@ await EventFilter.Warning(contains: "Association with remote system").ExpectOneA
              * Wait for the ReliableDeliverySupervisor to receive its "TooLongIdle" message,
              * which will throw a HopelessAssociation wrapped around a TimeoutException.
              */
-            await EventFilter.Exception<TimeoutException>().ExpectOneAsync(async () => { });
+            await EventFilter.Exception<TimeoutException>().ExpectOneAsync(() => { return Task.CompletedTask; });
 
             await AwaitAssertAsync(() =>
             {

src/core/Akka.Remote.Tests/RemotingSpec.cs

@@ -505,7 +505,7 @@ public async Task Stash_inbound_connections_until_UID_is_known_for_pending_outbo
                     (new ActorAssociationEventListener(remoteTransportProbe)));
 
                 // Hijack associations through the test transport
-                await AwaitConditionAsync(async () => registry.TransportsReady(rawLocalAddress, rawRemoteAddress));
+                await AwaitConditionAsync(() => Task.FromResult(registry.TransportsReady(rawLocalAddress, rawRemoteAddress)));
                 var testTransport = registry.TransportFor(rawLocalAddress).Value.Item1;
                 testTransport.WriteBehavior.PushConstant(true);
 
@@ -588,7 +588,7 @@ public async Task Properly_quarantine_stashed_inbound_connections()
                     (new ActorAssociationEventListener(remoteTransportProbe)));
 
                 // Hijack associations through the test transport
-                await AwaitConditionAsync(async () => registry.TransportsReady(rawLocalAddress, rawRemoteAddress));
+                await AwaitConditionAsync(() => Task.FromResult(registry.TransportsReady(rawLocalAddress, rawRemoteAddress)));
                 var testTransport = registry.TransportFor(rawLocalAddress).Value.Item1;
                 testTransport.WriteBehavior.PushConstant(true);