Help with AsyncStream termination task termination

[sillygoose]

Trying to find a way to terminate an AsyncStream when it is no longer needed. I have a TCA application that uses enum-drven navigation that operates as a big state machine pulling real-time data from a vehicle using an AsyncStream in a dependancy. Think of the download example in the Case Studies providing progress updates forever so there is no call to continuation.finish() and the download needs to ended externally.

       obdResponseStream: {
            .init { continuation in
                continuation.onTermination = { @Sendable status in
                    print("Stream terminated with status \(status)")
                }
                let task = Task {
                    while true {
                        let item = try await ObdManager.client.getObdResponse()
                        let result = parseObdResponse(obdLinkCommand: item.command, obdResponse: item.response)
                        switch result {
                        case .success(let obdLinkParse):
                            print("obdResponseStream.yield(\(obdLinkParse.command))")
                            continuation.yield(.response(obdLinkParse))
                        case .failure(let obdError):
                            continuation.yield(.error(obdError))
                        }
                        try? Task.checkCancellation()
                    }
                }
                task.cancel()
            }
        },

I found this code strange since task.cancel() is called right away but the task is never cancelled even though the termination closure is called when the scope changes. While debugging I ended up creating a second AsyncStream which shows the first stream is still alive and well:

connected-state: 2 commands sent: [...]
obdResponseStream.yield(pluggedIn)
obdResponseStream.yield(ignition)
Stream terminated with status cancelled (onTermination closure executed)
IgnitionOn-state (state change, now using obdResponseStream1)

ignitionOn-state: 5 commands sent: [...]
obdResponseStream1.yield(interiorTemperature)
obdResponseStream1.yield(exteriorTemperature)
obdResponseStream.yield(gear)  <===  old stream
obdResponseStream.yield(hvbEnergyToEmpty)  <=== old stream
obdResponseStream1.yield(hvbStateOfCharge)

You can see that the original (obdResponseStream) and new (obdResponseStream1) streams are both consuming data but I need the first stream gone to make this all work.

Any suggestions on programmatically killing off this stream from the TCA reducer?

[autoreleasefool]

I believe the problem is you're checking for your task cancellation incorrectly, and so the while loop runs indefinitely. Task.checkCancellation() is meant to throw an error, and you're swallowing it with try? so its unable to halt the execution.

You could instead check Task.isCancelled and break if it's true. Replace your line try? Task.checkCancellation() with the following:

if Task.isCancelled {
    break
}

You can reproduce this with the following playground:

import PlaygroundSupport

let task = Task {
	while true {
		print("task is running")

		// Try commenting out the next line and uncommenting the 3 after to see the output is only printed ~1 time
		 try? Task.checkCancellation()
//		if Task.isCancelled {
//			break
//		}
	}
}

task.cancel()

PlaygroundPage.current.needsIndefiniteExecution = true

[sillygoose]

Thanks for the great insight. Since I wanted to make some progress I created a command that is used to signal the continuation that it is finished. But it didn't work, that task kept consuming input that went in the closed stream.

I then included your idea about the while loop an it now works, the updated code is

        obdResponseStream: {
            .init { continuation in
                Task {
                    var taskCompleted = false
                    while !taskCompleted {
                        let result = try await ObdManager.client.getObdResponse()
                        switch result {
                        case .success(let obdLinkParse):
                            continuation.yield(.response(obdLinkParse))
                            if obdLinkParse.command == .lastCommand {
                                continuation.finish()
                                taskCompleted = true
                            }
                        case .failure(let obdError):
                            continuation.yield(.error(obdError))
                        }
                    }
                }
            }
        }

Thanks for your help, moving on to the next struggle.