Implemented AsyncSequence peerEvents property that can be used to observe peer discovery and connection events with the new Swift Concurrency API

Author: insidegui

Sources/MultipeerKit/Internal API/MockMultipeerConnection.swift

@@ -39,5 +39,31 @@ final class MockMultipeerConnection: MultipeerProtocol {
     func getLocalPeerId() -> String? {
         return localPeer.id
     }
+    
+    @discardableResult func findFakePeer(with id: String) -> Peer {
+        let fakePeer = Peer(
+            underlyingPeer: MCPeerID(displayName: id),
+            id: id,
+            name: id,
+            discoveryInfo: nil,
+            isConnected: false
+        )
+        
+        didFindPeer?(fakePeer)
+        
+        return fakePeer
+    }
+    
+    func loseFakePeer(_ fakePeer: Peer) {
+        didLosePeer?(fakePeer)
+    }
+    
+    func connectFakePeer(_ fakePeer: Peer) {
+        didConnectToPeer?(fakePeer)
+    }
+    
+    func disconnectFakePeer(_ fakePeer: Peer) {
+        didDisconnectFromPeer?(fakePeer)
+    }
 
 }

Sources/MultipeerKit/Public API/MultipeerTransceiver.swift

@@ -163,31 +163,46 @@ public final class MultipeerTransceiver {
     public func invite(_ peer: Peer, with context: Data?, timeout: TimeInterval, completion: InvitationCompletionHandler?) {
         connection.invite(peer, with: context, timeout: timeout, completion: completion)
     }
+    
+    public enum PeerEvent: Hashable {
+        case found(Peer)
+        case lost(Peer)
+        case connected(Peer)
+        case disconnected(Peer)
+    }
 
     private func handlePeerAdded(_ peer: Peer) {
         guard !availablePeers.contains(peer) else { return }
 
         availablePeers.append(peer)
         peerAdded(peer)
+        
+        peerEventOccurred(.found(peer))
     }
 
     private func handlePeerRemoved(_ peer: Peer) {
         guard let idx = availablePeers.firstIndex(where: { $0.underlyingPeer == peer.underlyingPeer }) else { return }
 
         availablePeers.remove(at: idx)
         peerRemoved(peer)
+        
+        peerEventOccurred(.lost(peer))
     }
 
     private func handlePeerConnected(_ peer: Peer) {
         setConnected(true, on: peer)
 
         peerConnected(peer)
+        
+        peerEventOccurred(.connected(peer))
     }
 
     private func handlePeerDisconnected(_ peer: Peer) {
         setConnected(false, on: peer)
 
         peerDisconnected(peer)
+        
+        peerEventOccurred(.disconnected(peer))
     }
 
     private func setConnected(_ connected: Bool, on peer: Peer) {
@@ -197,5 +212,44 @@ public final class MultipeerTransceiver {
         mutablePeer.isConnected = connected
         availablePeers[idx] = mutablePeer
     }
+    
+    // MARK: - Swift Concurrency Support
+    
+    private typealias PeerEventCallback = (PeerEvent) -> Void
+    
+    private var internalPeerEventCallbacks: [UUID: PeerEventCallback] = [:]
+   
+    private func addInternalPeerEventsCallback(with block: @escaping PeerEventCallback) -> UUID {
+        let id = UUID()
+        internalPeerEventCallbacks[id] = block
+        return id
+    }
+    
+    private func peerEventOccurred(_ event: PeerEvent) {
+        DispatchQueue.main.async {
+            self.internalPeerEventCallbacks.values.forEach { $0(event) }
+        }
+    }
 
 }
+
+@available(tvOS 13.0, *)
+@available(iOS 13.0, *)
+@available(macOS 10.15, *)
+public extension MultipeerTransceiver {
+    
+    var peerEvents: AsyncStream<PeerEvent> {
+        AsyncStream { [weak self] continuation in
+            guard let self = self else { return }
+            
+            let id = self.addInternalPeerEventsCallback { event in
+                continuation.yield(event)
+            }
+            
+            continuation.onTermination = { @Sendable _ in
+                self.internalPeerEventCallbacks[id] = nil
+            }
+        }
+    }
+    
+}

Tests/MultipeerKitTests/MultipeerKitTests.swift

@@ -47,10 +47,48 @@ final class MultipeerKitTests: XCTestCase {
 
         wait(for: [expect], timeout: 2)
     }
+    
+    @available(tvOS 13.0, *)
+    @available(iOS 13.0, *)
+    @available(macOS 10.15, *)
+    func testAsyncEventsStreamContinuesWithEachPeerEvent() throws {
+        let transceiver = makeMockTransceiver()
+        transceiver.resume()
+        
+        let exp = expectation(description: "Async Peer")
+        exp.expectedFulfillmentCount = 4
+        
+        Task.detached {
+            // Used to ensure that events are yielded in the right order.
+            var currentEvent = 0
+            
+            for await event in transceiver.peerEvents {
+                switch event {
+                case .found(let peer):
+                    XCTAssertEqual(peer.id, "A")
+                    XCTAssertEqual(currentEvent, 0)
+                case .connected(let peer):
+                    XCTAssertEqual(peer.id, "A")
+                    XCTAssertEqual(currentEvent, 1)
+                case .disconnected(let peer):
+                    XCTAssertEqual(peer.id, "A")
+                    XCTAssertEqual(currentEvent, 2)
+                case .lost(let peer):
+                    XCTAssertEqual(peer.id, "A")
+                    XCTAssertEqual(currentEvent, 3)
+                }
+                
+                currentEvent += 1
+                exp.fulfill()
+            }
+        }
+        
+        let peer = transceiver.mockConnection.findFakePeer(with: "A")
+        transceiver.mockConnection.connectFakePeer(peer)
+        transceiver.mockConnection.disconnectFakePeer(peer)
+        transceiver.mockConnection.loseFakePeer(peer)
+        
+        wait(for: [exp], timeout: 2)
+    }
 
-    static var allTests = [
-        ("testCallingResumeResumesConnection", testCallingResumeResumesConnection),
-        ("testCallingStopStopsConnection", testCallingStopStopsConnection),
-        ("testReceivingCustomPayload", testReceivingCustomPayload),
-    ]
 }