Cleanup actor system tests

Author: carson-katri

Tests/ErlangActorSystemTests/ErlangActorSystemTests.swift

@@ -238,52 +238,6 @@ import Foundation
         }
     }
 
-    @StableNames
-    distributed actor PubSubTest: HasRemoteCallAdapter {
-        typealias ActorSystem = ErlangActorSystem
-        
-        nonisolated var remoteCallAdapter: some RemoteCallAdapter {
-            .primitive
-        }
-        
-        @StableName("hello_world")
-        distributed func helloWorld() {
-            print("Received hello world event")
-        }
-    }
-    
-    @Test func pubSub() async throws {
-        let actorSystem = try await ErlangActorSystem(name: "swift", cookie: "LJTPNYYQIOIRKYDCWCQH")
-        try await actorSystem.connect(to: "iex@DCKYRD-NMXCKatri")
-        
-        let actor = PubSubTest(actorSystem: actorSystem)
-        
-        let pubSub = try await PubSub(name: "test_pubsub", actorSystem: actorSystem)
-        
-        await pubSub.whenLocal {
-            $0.subscribe(actor, to: "topic")
-        }
-        
-        print("joined")
-        
-        try await pubSub.whenLocal { try await $0.broadcast(Term.atom("hello_world"), to: "topic") }
-        
-        try await Task.sleep(for: .seconds(1024))
-    }
-    
-    @Resolvable
-    @StableNames
-    protocol CounterProtocol: DistributedActor, HasStableNames where ActorSystem == ErlangActorSystem {
-        @StableName("count")
-        distributed var count: Int { get }
-        
-        @StableName("increment")
-        distributed func increment()
-        
-        @StableName("decrement")
-        distributed func decrement()
-    }
-    
     @StableNames
     distributed actor Counter: CounterProtocol {
         var _count = 0
@@ -329,181 +283,15 @@ import Foundation
     }
 }
 
-import CErlInterface
-
-/// ```elixir
-/// {:forward_to_local, "<topic>", <message>, <dispatcher>}
-/// ```
-///
-/// 'Elixir.Phoenix.PubSub'
-struct PubSubRemoteCallAdapter: RemoteCallAdapter {
-    let topic: String
-    let primitive = PrimitiveRemoteCallAdapter()
-    
-    func encode(
-        _ invocation: RemoteCallInvocation,
-        for system: ActorSystem
-    ) throws -> EncodedRemoteCall {
-        let message = ErlangTermBuffer()
-        message.newWithVersion()
-        message.encode(tupleHeader: 4)
-        _ = Data(topic.utf8).withUnsafeBytes {
-            message.encode(binary: $0.baseAddress!, len: Int32($0.count))
-        }
-        
-        message.append(try primitive.encode(invocation, for: system).message)
-        
-        message.encode(atom: "Elixir.Phoenix.PubSub")
-        print(message)
-        
-        return EncodedRemoteCall(
-            message: message,
-            continuationAdapter: nil
-        )
-    }
-    
-    func decode(
-        _ message: ErlangTermBuffer,
-        for system: ActorSystem
-    ) throws -> LocalCallInvocation {
-        var index: Int32 = 0
-        
-        var version: Int32 = 0
-        message.decode(version: &version, index: &index)
-        
-        var arity: Int32 = 0
-        message.decode(tupleHeader: &arity, index: &index)
-        
-        message.skipTerm(index: &index) // :forward_to_local
-        
-        var type: UInt32 = 0
-        var size: Int32 = 0
-        message.getType(type: &type, size: &size, index: &index)
-        let binary: UnsafeMutableRawPointer = .allocate(byteCount: Int(size), alignment: 0)
-        var length: Int = 0
-        message.decode(binary: binary, len: &length, index: &index)
-        let topic = String(data: Data(bytes: binary, count: length), encoding: .utf8)!
-        guard topic == self.topic else { throw PubSubError.unhandledTopic(topic) }
-        
-        let messageStart = index
-        message.skipTerm(index: &index)
-        
-        return try primitive.decode(message[messageStart..<index], for: system)
-    }
-    
-    enum PubSubError: Error {
-        case unhandledTopic(String)
-    }
-}
-
-extension RemoteCallAdapter where Self == ProcessGroupRemoteCallAdapter {
-    static func pubSub(topic: String) -> ProcessGroupRemoteCallAdapter {
-        .processGroup(PubSubRemoteCallAdapter(topic: topic))
-    }
-}
-
+@Resolvable
 @StableNames
-public distributed actor PubSub: HasRemoteCallAdapter {
-    public typealias ActorSystem = ErlangActorSystem
-    
-    private let name: String
-    private var subscribers = [String:Set<ActorSystem.ActorID>]()
-    
-    public nonisolated var remoteCallAdapter: ProcessGroupRemoteCallAdapter {
-        .processGroup
-    }
-    
-    public init(name: String, actorSystem: ActorSystem) async throws {
-        self.name = name
-        self.actorSystem = actorSystem
-        
-        try await self.join(
-            scope: "Elixir.Phoenix.PubSub",
-            group: "Elixir.\(name).Adapter"
-        )
-    }
-    
-    @StableName("forward_to_local")
-    distributed func forwardToLocal(topic: String, message: Term, dispatcher: String) async throws {
-        for subscriber in subscribers[topic] ?? [] {
-            guard let actor = actorSystem.resolve(id: subscriber)
-            else { continue }
-            
-            let remoteCallAdapter = (actor as? any HasRemoteCallAdapter)?.remoteCallAdapter ?? actorSystem.remoteCallAdapter
-            let localCall = try remoteCallAdapter.decode(message.makeBuffer(), for: actorSystem)
-            
-            nonisolated(unsafe) let handler = ActorSystem.ResultHandler(
-                sender: localCall.sender,
-                resultHandlerAdapter: localCall.resultHandler,
-                fileDescriptor: 0
-            )
-            
-            let decoder = TermDecoder()
-            nonisolated(unsafe) var invocationDecoder = ActorSystem.InvocationDecoder(
-                buffer: localCall.arguments,
-                decoder: decoder,
-                index: 0
-            )
-            
-            if let stableNamed = actor as? any HasStableNames {
-                try! await stableNamed._executeStableName(
-                    target: RemoteCallTarget(localCall.identifier),
-                    invocationDecoder: &invocationDecoder,
-                    handler: handler
-                )
-            } else {
-                try await actorSystem.executeDistributedTarget(
-                    on: actor,
-                    target: RemoteCallTarget(localCall.identifier),
-                    invocationDecoder: &invocationDecoder,
-                    handler: handler
-                )
-            }
-        }
-    }
+protocol CounterProtocol: DistributedActor, HasStableNames where ActorSystem == ErlangActorSystem {
+    @StableName("count")
+    distributed var count: Int { get }
 
-    public func subscribe<Act>(_ act: Act, to topic: String)
-        where Act: DistributedActor, Act.ID == ActorSystem.ActorID
-    {
-        subscribers[topic, default: []].insert(act.id)
-    }
-    
-    public func unsubscribe<Act>(_ act: Act, to topic: String)
-        where Act: DistributedActor, Act.ID == ActorSystem.ActorID
-    {
-        subscribers[topic]?.remove(act.id)
-    }
+    @StableName("increment")
+    distributed func increment()
 
-    public func localBroadcast<Act: DistributedActor>(
-        to topic: String,
-        as actorType: Act.Type = Act.self,
-        _ body: @Sendable (Act) async throws -> ()
-    ) async throws where Act.ID == ActorSystem.ActorID {
-        for subscriber in subscribers["topic"] ?? [] {
-            guard let actor = try actorSystem.resolve(id: subscriber, as: actorType)
-            else { continue }
-            try await body(actor)
-        }
-    }
-    
-    public func broadcast(
-        _ message: some Encodable,
-        to topic: String
-    ) async throws {
-        let encoder = TermEncoder()
-        encoder.includeVersion = false
-        let message = try Term(from: encoder.encode(message))
-        
-        try await forwardToLocal(topic: topic, message: message, dispatcher: "Elixir.Phoenix.PubSub")
-        
-        for member in try await ProcessGroups.remoteMembers(
-            scope: "Elixir.Phoenix.PubSub",
-            group: "Elixir.\(name).Adapter",
-            using: actorSystem
-        ) {
-            print(member)
-            let remote = try PubSub.resolve(id: member, using: actorSystem)
-            try await remote.forwardToLocal(topic: topic, message: message, dispatcher: "Elixir.Phoenix.Pubsub")
-        }
-    }
-}
+    @StableName("decrement")
+    distributed func decrement()
+}