I’ve read plenty of blog posts about Swift’s AsyncStream but never had a reason to use it to implement something myself until now. It was surprising easy.

AsyncStream lets you await a sequence of items, returning them when they are ready.

Apple is generally pretty good at making their APIs async, but there are some holdovers, and I came across one this weekend.

Specifically the way to fetch photos from Apple Photos is to use PHAsset.fetchAssets

import Photos
   ...
let fetchOptions = PHFetchOptions()
fetchOptions.includeHiddenAssets = false
let assets = PHAsset.fetchAssets(with: fetchOptions)

This returns a PHFetchResult<PHAsset> which you can then enumerate, passing a callback:

assets.enumerateObjects { (asset, idx, stop) in 
   ...
}

The asset callback parameter is the PHAsset representing the photo or video, the idx is the index, and the stop is a pointer to a boolean you can use to cancel the enumeration. I kid you not.

I implemented my logic as above, but it left a bad taste in my mouth. The surrounding code was all async.

I decided to re-write it using AsyncStream:

let assets = PHAsset.fetchAssets(with: fetchOptions)

// Use AsyncStream's continuation to asynchronously yield elements
let asyncStream = AsyncStream<PHAsset> { continuation in
    assets.enumerateObjects { (asset, _, _) in
    	continuation.yield(asset)
    }
    continuation.finish()
}

// Now iterate over the asyncStream awaiting each element
for await asset in asyncStream {
   ...
}

The AsyncStream has a closure with a continuation which you use to yield elements when they are ready or finish when you are done.

That was it. I’d converted my code to use AsyncStream and it all worked pretty much out of the box … except for that pesky stop parameter. How could I cancel the enumeratation if the AsyncStream was cancelled?

The naive approach would be:

// BAD - DOES NOT WORK
let asyncStream = AsyncStream<PHAsset> { continuation in
    var cancelled = false
    continuation.onTermination = { _ in
        cancelled = true // Mutation of captured var 'cancelled' in concurrently-executing code
    }
    assets.enumerateObjects { (asset, _, stop) in
        if cancelled {
            stop.pointee = true
        } else {
            continuation.yield(asset)
        }
    }
    continuation.finish()
}

This gives the pesky Mutation of captured var ‘cancelled’ in concurrently-executing code for cancelled = true which I’m sure you’ve never come across before.

To resolve it I imported Atomics and used a ManagedAtomic:

import Photos
import Atomics
   ...
let asyncStream = AsyncStream<PHAsset> { continuation in
    // Use ManagedAtomic to wrap a bool
    let cancelled = ManagedAtomic<Bool>(false)
    continuation.onTermination = { _ in
        cancelled.store(true, ordering: .relaxed)
    }
    assets.enumerateObjects { (asset, _, stop) in
        if cancelled.load(ordering: .relaxed) {
            stop.pointee = true // Sir I think your Objective C is showing
        } else {
            continuation.yield(asset)
        }
    }
    continuation.finish()
}

Now I’m happy! If you know a better/cleaner way please let me know.