MultipleStreams.swift

import Foundation

final class FooClass: @unchecked Sendable {
    let streamA: AsyncStream<Int>
    let streamB: AsyncStream<Int>
    
    init() {
        // Publishes a new value every 5s
        self.streamA = AsyncStream<Int> { continuation in
            DispatchQueue.main.sync {
                Timer.scheduledTimer(withTimeInterval: 5, repeats: true) { timer in
                    // print("Timer A fired!")
                    continuation.yield(0)
                }
            }
        }
        // Publishes a new value every 5s, with a 2.5s offset to the first one
        self.streamB = AsyncStream<Int> { continuation in
            DispatchQueue.main.asyncAfter(deadline: .now() + 2.5) {
                Timer.scheduledTimer(withTimeInterval: 5, repeats: true) { timer in
                    // print("Timer B fired!")
                    continuation.yield(1)
                }
            }
        }
    }
    
    deinit {
        print("[!] FooClass deinit called, memory deallocated ✅")
        exit(1)
    }
    
    // Let's pretend this method does some work that takes around 3 seconds
    // to complete.
    // Since this is an instance method of FooClass, calling this function
    // always captures self strongly.
    func doSomething(with value: Int, from stream: String) async {
        print("    doSomething() from \(stream) strongly captured self")
        try! await Task.sleep(for: .seconds(3))
        print("    doSomething() from \(stream) released self")
    }
    
    func subscribeToA() {
        Task { [weak self] in
            print("[!] start Task A")
            guard let streamA = self?.streamA else {
                return
            }
            for await value in streamA {
                await self?.doSomething(with: value, from: "A")
            }
        }
    }
    func subscribeToB() {
        Task { [weak self] in
            print("[!] start Task B")
            guard let streamB = self?.streamB else {
                return
            }
            for await value in streamB {
                await self?.doSomething(with: value, from: "B")
            }
        }
    }
}

func test() async {
    print("[!] External reference to FooClass instance created")
    let foo = FooClass()
    print("[!] subscribe to A")
    foo.subscribeToA()
    print("[!] subscribe to B")
    foo.subscribeToB()
    // Pretend FooClass() needs to be alive for 5 seconds
    try! await Task.sleep(for: .seconds(10))
    print("[!] Last external reference to FooClass instance goes out of scope")
}

// Test whether referencing and dereferencing a FooClass instance
// correctly deallocates memory.
Task {
    await test()
    
    // Fallback test: if FooClass's deinit hasn't been called after
    // 60 seconds (ending the program), raise a fatal error.
    try await Task.sleep(for: .seconds(60))
    print("[!] FooClass instance still in memory after 60 seconds ❌")
    fatalError()
}