Address remaining feedback in #1012 (#1015)

yim-lee · web-flow · commit 208152735792 · 2022-07-20T19:39:11.000+09:00
diff --git a/IntegrationTests/tests_01_cluster/it_Clustered_swim_suspension_reachability/main.swift b/IntegrationTests/tests_01_cluster/it_Clustered_swim_suspension_reachability/main.swift
@@ -41,15 +41,6 @@ let system = await ClusterSystem("System") { settings in
     settings.downingStrategy = .none
 }
 
-let ref = try system._spawn(
-    "streamWatcher",
-    of: Cluster.Event.self,
-    .receive { context, event in
-        context.log.info("Event: \(event)")
-        return .same
-    }
-)
-
 Task {
     for await event in system.cluster.events {
         system.log.info("Event: \(event)")
diff --git a/Sources/DistributedActors/Cluster/ClusterEventStream.swift b/Sources/DistributedActors/Cluster/ClusterEventStream.swift
@@ -41,15 +41,27 @@ public struct ClusterEventStream: AsyncSequence {
         self.actor = nil
     }
 
-    func subscribe(_ ref: _ActorRef<Cluster.Event>, file: String = #filePath, line: UInt = #line) async {
+    nonisolated func subscribe(_ ref: _ActorRef<Cluster.Event>, file: String = #filePath, line: UInt = #line) {
+        Task {
+            await self._subscribe(ref, file: file, line: line)
+        }
+    }
+
+    func _subscribe(_ ref: _ActorRef<Cluster.Event>, file: String = #filePath, line: UInt = #line) async {
         guard let actor = self.actor else { return }
 
         await actor.whenLocal { __secretlyKnownToBeLocal in // TODO(distributed): this is annoying, we must track "known to be local" in typesystem instead
             __secretlyKnownToBeLocal.subscribe(ref)
         }
     }
 
-    func unsubscribe(_ ref: _ActorRef<Cluster.Event>, file: String = #filePath, line: UInt = #line) async {
+    nonisolated func unsubscribe(_ ref: _ActorRef<Cluster.Event>, file: String = #filePath, line: UInt = #line) {
+        Task {
+            await self._unsubscribe(ref, file: file, line: line)
+        }
+    }
+
+    func _unsubscribe(_ ref: _ActorRef<Cluster.Event>, file: String = #filePath, line: UInt = #line) async {
         guard let actor = self.actor else { return }
 
         await actor.whenLocal { __secretlyKnownToBeLocal in // TODO(distributed): this is annoying, we must track "known to be local" in typesystem instead
diff --git a/Sources/DistributedActors/Cluster/Leadership.swift b/Sources/DistributedActors/Cluster/Leadership.swift
@@ -112,9 +112,7 @@ extension Leadership {
         var behavior: _Behavior<Cluster.Event> {
             .setup { context in
                 context.log.trace("Configured with \(self.election)")
-                Task {
-                    await context.system.cluster.events.subscribe(context.myself)
-                }
+                context.system.cluster.events.subscribe(context.myself)
 
                 // FIXME: we have to add "own node" since we're not getting the .snapshot... so we have to manually act as if..
                 _ = self.membership.applyMembershipChange(Cluster.MembershipChange(node: context.system.cluster.uniqueNode, previousStatus: nil, toStatus: .joining))
diff --git a/Sources/DistributedActors/Cluster/NodeDeathWatcher.swift b/Sources/DistributedActors/Cluster/NodeDeathWatcher.swift
@@ -214,9 +214,7 @@ enum NodeDeathWatcherShell {
                     () // ignore other changes, we only need to react on nodes becoming DOWN
                 }
             }
-            Task {
-                await context.system.cluster.events.subscribe(onClusterEventRef)
-            }
+            context.system.cluster.events.subscribe(onClusterEventRef)
 
             return NodeDeathWatcherShell.behavior(instance)
         }
diff --git a/Sources/DistributedActors/Cluster/Reception/_OperationLogClusterReceptionistBehavior.swift b/Sources/DistributedActors/Cluster/Reception/_OperationLogClusterReceptionistBehavior.swift
@@ -95,13 +95,10 @@ public final class _OperationLogClusterReceptionist {
             context.log.debug("Initialized receptionist")
 
             // === listen to cluster events ------------------
-            Task {
-                await context.system.cluster.events.subscribe(
-                    context.subReceive(Cluster.Event.self) { event in
-                        self.onClusterEvent(context, event: event)
-                    }
-                )
+            let onClusterEventRef = context.subReceive(Cluster.Event.self) { event in
+                self.onClusterEvent(context, event: event)
             }
+            context.system.cluster.events.subscribe(onClusterEventRef)
 
             // === timers ------------------
             // periodically gossip to other receptionists with the last seqNr we've seen,
diff --git a/Sources/DistributedActors/Gossip/Gossiper+Shell.swift b/Sources/DistributedActors/Gossip/Gossiper+Shell.swift
@@ -330,9 +330,7 @@ extension GossipShell {
                     () // ignore
                 }
             }
-            Task {
-                await context.system.cluster.events.subscribe(onClusterEventRef)
-            }
+            context.system.cluster.events.subscribe(onClusterEventRef)
 
         case .fromReceptionistListing(let id):
             let key = _Reception.Key(_ActorRef<Message>.self, id: id)
diff --git a/Sources/DistributedActorsTestKit/ActorTestKit.swift b/Sources/DistributedActorsTestKit/ActorTestKit.swift
@@ -106,7 +106,7 @@ extension ActorTestKit {
     ) async -> ActorTestProbe<Cluster.Event> {
         let eventStream = self.system.cluster.events
         let p = self.makeTestProbe(naming ?? _ActorNaming.prefixed(with: "\(ClusterEventStream.self)-subscriberProbe"), expecting: Cluster.Event.self)
-        await eventStream.subscribe(p.ref)
+        await eventStream._subscribe(p.ref)
         return p
     }
 }
diff --git a/Tests/DistributedActorsTests/Cluster/AssociationClusteredTests.swift b/Tests/DistributedActorsTests/Cluster/AssociationClusteredTests.swift
@@ -97,8 +97,8 @@ final class ClusterAssociationTests: ClusteredActorSystemsXCTestCase {
 
         let firstEventsProbe = self.testKit(first).makeTestProbe(expecting: Cluster.Event.self)
         let secondEventsProbe = self.testKit(second).makeTestProbe(expecting: Cluster.Event.self)
-        await first.cluster.events.subscribe(firstEventsProbe.ref)
-        await second.cluster.events.subscribe(secondEventsProbe.ref)
+        await first.cluster.events._subscribe(firstEventsProbe.ref)
+        await second.cluster.events._subscribe(secondEventsProbe.ref)
 
         first.cluster.join(node: second.cluster.uniqueNode.node)
 
@@ -113,8 +113,8 @@ final class ClusterAssociationTests: ClusteredActorSystemsXCTestCase {
             settings.bindPort = secondPort
         }
         let secondReplacementEventsProbe = self.testKit(secondReplacement).makeTestProbe(expecting: Cluster.Event.self)
-        await secondReplacement.cluster.events.subscribe(secondReplacementEventsProbe.ref)
-        await second.cluster.events.subscribe(secondReplacementEventsProbe.ref)
+        await secondReplacement.cluster.events._subscribe(secondReplacementEventsProbe.ref)
+        await second.cluster.events._subscribe(secondReplacementEventsProbe.ref)
 
         // the new replacement node is now going to initiate a handshake with 'first' which knew about the previous
         // instance (oldSecond) on the same node; It should accept this new handshake, and ban the previous node.
diff --git a/Tests/DistributedActorsTests/Cluster/ClusterEventStreamTests.swift b/Tests/DistributedActorsTests/Cluster/ClusterEventStreamTests.swift
@@ -37,9 +37,9 @@ final class ClusterEventStreamTests: ClusterSystemXCTestCase, @unchecked Sendabl
 
         let eventStream = ClusterEventStream(system, customName: "testClusterEvents")
 
-        await eventStream.subscribe(p1.ref) // sub before first -> up was published
+        await eventStream._subscribe(p1.ref) // sub before first -> up was published
         await eventStream.publish(.membershipChange(.init(member: self.memberA, toStatus: .up)))
-        await eventStream.subscribe(p2.ref)
+        await eventStream._subscribe(p2.ref)
         await eventStream.publish(.membershipChange(.init(member: self.memberB, toStatus: .up)))
 
         // ==== p1 ---------------------
@@ -89,7 +89,7 @@ final class ClusterEventStreamTests: ClusterSystemXCTestCase, @unchecked Sendabl
         await eventStream.publish(.membershipChange(.init(member: self.memberA, toStatus: .up)))
         await eventStream.publish(.membershipChange(.init(member: self.memberB, toStatus: .joining)))
         await eventStream.publish(.membershipChange(.init(member: self.memberB, toStatus: .up)))
-        await eventStream.subscribe(p1.ref)
+        await eventStream._subscribe(p1.ref)
 
         // ==== p1 ---------------------
 
diff --git a/Tests/DistributedActorsTests/Cluster/ClusterLeaderActionsClusteredTests.swift b/Tests/DistributedActorsTests/Cluster/ClusterLeaderActionsClusteredTests.swift
@@ -35,9 +35,7 @@ final class ClusterLeaderActionsClusteredTests: ClusteredActorSystemsXCTestCase
         _ = try first._spawn(
             "selfishSingleLeader",
             _Behavior<Cluster.Event>.setup { context in
-                Task {
-                    await context.system.cluster.events.subscribe(context.myself)
-                }
+                context.system.cluster.events.subscribe(context.myself)
 
                 return .receiveMessage { event in
                     switch event {
@@ -154,9 +152,9 @@ final class ClusterLeaderActionsClusteredTests: ClusteredActorSystemsXCTestCase
         }
 
         let p1 = self.testKit(first).makeTestProbe(expecting: Cluster.Event.self)
-        await first.cluster.events.subscribe(p1.ref)
+        await first.cluster.events._subscribe(p1.ref)
         let p2 = self.testKit(second).makeTestProbe(expecting: Cluster.Event.self)
-        await second.cluster.events.subscribe(p2.ref)
+        await second.cluster.events._subscribe(p2.ref)
 
         first.cluster.join(node: second.cluster.uniqueNode.node)
 
@@ -221,21 +219,21 @@ final class ClusterLeaderActionsClusteredTests: ClusteredActorSystemsXCTestCase
             settings.downingStrategy = .timeout(.init(downUnreachableMembersAfter: .milliseconds(300)))
         }
         let p1 = testKit(first).makeTestProbe(expecting: Cluster.Event.self)
-        await first.cluster.events.subscribe(p1.ref)
+        await first.cluster.events._subscribe(p1.ref)
 
         let second = await setUpNode("second") { settings in
             settings.autoLeaderElection = .lowestReachable(minNumberOfMembers: 2)
             settings.downingStrategy = .timeout(.init(downUnreachableMembersAfter: .milliseconds(300)))
         }
         let p2 = testKit(second).makeTestProbe(expecting: Cluster.Event.self)
-        await second.cluster.events.subscribe(p2.ref)
+        await second.cluster.events._subscribe(p2.ref)
 
         let third = await setUpNode("third") { settings in
             settings.autoLeaderElection = .lowestReachable(minNumberOfMembers: 2)
             settings.downingStrategy = .timeout(.init(downUnreachableMembersAfter: .milliseconds(300)))
         }
         let p3 = self.testKit(third).makeTestProbe(expecting: Cluster.Event.self)
-        await third.cluster.events.subscribe(p3.ref)
+        await third.cluster.events._subscribe(p3.ref)
 
         try await self.joinNodes(node: first, with: second)
         try await self.joinNodes(node: second, with: third)
@@ -293,7 +291,7 @@ final class ClusterLeaderActionsClusteredTests: ClusteredActorSystemsXCTestCase
             settings.downingStrategy = .timeout(.init(downUnreachableMembersAfter: .milliseconds(200)))
         }
         let p1 = self.testKit(first).makeTestProbe(expecting: Cluster.Event.self)
-        await first.cluster.events.subscribe(p1.ref)
+        await first.cluster.events._subscribe(p1.ref)
 
         let second = await setUpNode("second") { settings in
             settings.swim.probeInterval = .milliseconds(300)
@@ -302,7 +300,7 @@ final class ClusterLeaderActionsClusteredTests: ClusteredActorSystemsXCTestCase
             settings.downingStrategy = .timeout(.init(downUnreachableMembersAfter: .milliseconds(200)))
         }
         let p2 = self.testKit(second).makeTestProbe(expecting: Cluster.Event.self)
-        await second.cluster.events.subscribe(p2.ref)
+        await second.cluster.events._subscribe(p2.ref)
 
         let third = await setUpNode("third") { settings in
             settings.swim.probeInterval = .milliseconds(300)
@@ -311,7 +309,7 @@ final class ClusterLeaderActionsClusteredTests: ClusteredActorSystemsXCTestCase
             settings.downingStrategy = .timeout(.init(downUnreachableMembersAfter: .milliseconds(200)))
         }
         let p3 = self.testKit(third).makeTestProbe(expecting: Cluster.Event.self)
-        await third.cluster.events.subscribe(p3.ref)
+        await third.cluster.events._subscribe(p3.ref)
 
         try await self.joinNodes(node: first, with: second)
         try await self.joinNodes(node: second, with: third)

Original file line number	Diff line number	Diff line change
`@@ -214,9 +214,7 @@ enum NodeDeathWatcherShell {`
`214`	`214`	`() // ignore other changes, we only need to react on nodes becoming DOWN`
`215`	`215`	`}`
`216`	`216`	`}`
`217`		`- Task {`
`218`		`- await context.system.cluster.events.subscribe(onClusterEventRef)`
`219`		`- }`
	`217`	`+ context.system.cluster.events.subscribe(onClusterEventRef)`
`220`	`218`
`221`	`219`	`return NodeDeathWatcherShell.behavior(instance)`
`222`	`220`	`}`
Original file line number	Diff line number	Diff line change
`@@ -330,9 +330,7 @@ extension GossipShell {`
`330`	`330`	`() // ignore`
`331`	`331`	`}`
`332`	`332`	`}`
`333`		`- Task {`
`334`		`- await context.system.cluster.events.subscribe(onClusterEventRef)`
`335`		`- }`
	`333`	`+ context.system.cluster.events.subscribe(onClusterEventRef)`
`336`	`334`
`337`	`335`	`case .fromReceptionistListing(let id):`
`338`	`336`	`let key = _Reception.Key(_ActorRef<Message>.self, id: id)`
Original file line number	Diff line number	Diff line change
`@@ -106,7 +106,7 @@ extension ActorTestKit {`
`106`	`106`	`) async -> ActorTestProbe<Cluster.Event> {`
`107`	`107`	`let eventStream = self.system.cluster.events`
`108`	`108`	`let p = self.makeTestProbe(naming ?? _ActorNaming.prefixed(with: "\(ClusterEventStream.self)-subscriberProbe"), expecting: Cluster.Event.self)`
`109`		`- await eventStream.subscribe(p.ref)`
	`109`	`+ await eventStream._subscribe(p.ref)`
`110`	`110`	`return p`
`111`	`111`	`}`
`112`	`112`	`}`