How to exit from `RunLoop`

Question

I'm subclassing InputStream from iOS Foundation SDK for my needs. I need to implement functionality that worker thread can sleep until data appear in the stream. The test I'm using to cover the functionality is below:

    func testStreamWithRunLoop() {
        let inputStream = BLEInputStream() // custom input stream subclass
        inputStream.delegate = self
        let len = Int.random(in: 0..<100)
        let randomData = randData(length: len) // random data generation
        
        let tenSeconds = Double(10)
        let oneSecond = TimeInterval(1)
        runOnBackgroundQueueAfter(oneSecond) {
            inputStream.accept(randomData) // input stream receives the data
        }
        let dateInFuture = Date(timeIntervalSinceNow: tenSeconds) // time in 10 sec
        inputStream.schedule(in: .current, forMode: RunLoop.Mode.default) // 
        RunLoop.current.run(until: dateInFuture) // wait for data appear in input stream
        XCTAssertTrue(dateInFuture.timeIntervalSinceNow > 0, "Timeout. RunLoop didn't exit in 1 sec. ")
    }

Here the overriden methods of InputStream

    public override func schedule(in aRunLoop: RunLoop, forMode mode: RunLoop.Mode) {
        self.runLoop = aRunLoop // save RunLoop object
        var context = CFRunLoopSourceContext() // make context
        self.runLoopSource = CFRunLoopSourceCreate(nil, 0, &context) // make source
        let cfloopMode: CFRunLoopMode = CFRunLoopMode(mode as CFString) 
        CFRunLoopAddSource(aRunLoop.getCFRunLoop(), self.runLoopSource!, cfloopMode)
    }

public func accept(_ data: Data) { 
        guard data.count > 0 else { return }
        
        self.data += data
        delegate?.stream?(self, handle: .hasBytesAvailable)
        
        if let runLoopSource {
            CFRunLoopSourceSignal(runLoopSource)
        }
        if let runLoop {
            CFRunLoopWakeUp(runLoop.getCFRunLoop())
        }
    }

But calling CFRunLoopSourceSignal(runLoopSource) and CFRunLoopWakeUp(runLoop.getCFRunLoop()) not get exit from runLoop.

Does anybody know where I'm mistaking ?

Thanks all!

PS: Here the Xcode project on GitHub

Answer 1

1. In Apple's runtime, NSMainThread and MainRunLoop is very special object. MainRunLoop does shallow sleep rather than deep sleep like normal RunLoop. Apple's frameworks likely to dispatch events to MainRunLoop.

which means....

• MainRunLoop awakes automatically with CFRunLoopPerform and CFRunLoopSource0 signal
• OS and Frameworks likely to attach external sources and observer

2. Subclassing OutputStream and InputStream is not recommended. They are toll-free-bridged to CFWriteStream and CFReadStream, and Stream API are heavily depends on CFStream API

CFReadStream, CFWriteStream

Consider using CFReadStreamSetDispatchQueue(_:_:) and CFWriteStreamSetDispatchQueue(_:_:) rather then RunLoop which is much easier.

extension InputStream {
    
    final var targetQueue:DispatchQueue? {
        get { CFReadStreamCopyDispatchQueue(self) }
        set { CFReadStreamSetDispatchQueue(self, newValue) }
    }
    
}

extension OutputStream {
    
    final var targetQueue:DispatchQueue? {
        get { CFWriteStreamCopyDispatchQueue(self) }
        set { CFWriteStreamSetDispatchQueue(self, newValue) }
    }
    
}

Answer 2

Finally I figured out some issues with my code.

First of all I need to remove CFRunLoopSource object from run loop CFRunLoopRemoveSource(). In according with documentation if RunLoop has no input sources then it exits immediately.

    public func accept(_ data: Data) {
        guard data.count > 0 else { return }
        
        self.data += data
        delegate?.stream?(self, handle: .hasBytesAvailable)
        
        if let runLoopSource, let runLoop, let runLoopMode {
            CFRunLoopRemoveSource(runLoop.getCFRunLoop(), runLoopSource, runLoopMode)
        }
        if let runLoop {
            CFRunLoopWakeUp(runLoop.getCFRunLoop())
        }
    }

Second issue is related that I used XCTest environment and it's RunLoop didn't exit for some reasons (Ask the community for help).

I used real application environment and created Thread subclass to check my implementation. The thread by default has run loop without any input sources attached to it. I added input stream to it. And using main thread emulated that stream received data.

Here the Custom Thread implement that runs and sleep until it receive signal from BLEInputStream

class StreamThread: Thread, StreamDelegate {
    let stream: BLEInputStream
    
    init(stream: BLEInputStream) {
        self.stream = stream
    }
    
    override func main() {
        stream.delegate = self
        stream.schedule(in: .current, forMode: RunLoop.Mode.default)
        print("start()")
        let tenSeconds = Double(10)
        let dateInFuture = Date(timeIntervalSinceNow: tenSeconds)
        RunLoop.current.run(until: dateInFuture)
        print("after 10 seconds")
    }
    
    override func start() {
        super.start()
    }
    
    func stream(_ aStream: Stream, handle eventCode: Stream.Event) {
        if eventCode == .errorOccurred {
            print("eventCode == .errorOccurred")
        }
        else if eventCode == .hasBytesAvailable {
            print("eventCode == .hasBytesAvailable")
        }
    }
}

Here the some UIViewController methods which runs from main thread

    override func viewDidAppear(_ animated: Bool) {
        super.viewDidAppear(animated)
        let baseDate = Date.now
        let thread = StreamThread(stream: stream, baseDate: baseDate)
        thread.start()
        print("main thread pauses at \(Date.now.timeIntervalSince(baseDate))")
        Thread.sleep(forTimeInterval: 2) 
        print("stream accepts Data \(Date.now.timeIntervalSince(baseDate))")
        stream.accept(Data([1,2,3]))
    }

Here the result:

Everything works as expected - the thread sleeps until input stream receive data. No processor resources consuming.

Although it's allowed to subclass InputStream, there is no good explanation in the documentation how to correctly implement custom InputStream