Understanding Swift Concurrency’s AsyncStream – Donny Wals

In an earlier publish, I wrote about alternative ways that you may bridge your present asynchronous code over to Swift’s new Concurrency system that leverages async / await. The mechanisms proven there work nice for code the place your code produces a single consequence that may be modeled as a single worth.

Nevertheless in some instances this isn’t potential as a result of your present code will present a number of values over time. That is the case for issues like obtain progress, the consumer’s present location, and different comparable conditions.

Typically talking, these sorts of patterns can be modeled as AsyncSequence objects that you may iterate over utilizing an asynchronous for loop. A primary instance of this may be the strains property on URL:

let url = URL(string: "https://donnywals.com")!

for attempt await line in url.strains {
    // use line
}

However what’s one of the best ways to construct your personal async sequences? Implementing the AsyncSequence protocol and constructing your on AsyncIterator sounds tedious and error-prone. Fortunately, there’s no purpose so that you can be doing any of that.

On this publish, I’ll present you how one can leverage Swift’s AsyncStream to construct customized async sequences that produce values everytime you want them to.

Producing a easy async stream

An async stream might be produced in numerous methods. The simplest approach to create an async stream is to make use of the AsyncStream(unfolding:) initializer. Its utilization seems to be a bit as follows:

let stream = AsyncStream(unfolding: {
    return Int.random(in: 0..<Int.max)
})

After all, this instance isn’t significantly helpful by itself nevertheless it does present how easy the idea of AsyncStream(unfolding:) is. We use this model of AsyncStream every time we are able to produce and return return values for our async stream. The closure that’s handed to unfolding is async so which means we are able to await asynchronous operations from inside our unfolding closure. Your unfolding closure will probably be referred to as each time you’re anticipated to start producing a worth to your stream. In apply which means your closure will probably be referred to as, you carry out some work, you come back a worth after which your closure is known as. This repeats till the for loop is cancelled, the duty that accommodates your async for loop is cancelled, or till you come back nil out of your unfolding closure.

The AsyncStream(unfolding:) approach to produce a stream of values is sort of handy nevertheless it’s significantly helpful in conditions the place:

• You need to carry out async work that must be awaited to supply components
• You’ve gotten a must deal with again stress when bridging an API you personal

Whenever you’re bridging an present API that’s based mostly on delegates or for APIs that leverage callbacks to speak outcomes, you in all probability gained’t be capable to use AsyncStream(unfolding:). Whereas it’s the only and least error-prone approach to construct an async stream, it’s additionally the best way that I’ve discovered to be most limiting and it doesn’t usually match nicely with bridging present code over to Swift Concurrency.

Extra flexibility might be discovered within the continuation based mostly API for AsyncStream.

Producing an async stream with a continuation

When an asynchronous closure doesn’t fairly suit your use case for creating your personal async stream, a continuation based mostly strategy may be a a lot better answer for you. With a continuation you have got the flexibility to assemble an async stream object and ship values over the async stream every time values turn into out there.

We will do that by creating an AsyncStream utilizing the AsyncStream(construct:) initializer:

let stream2 = AsyncStream { cont in
    cont.yield(Int.random(in: 0..<Int.max))
}

The instance above creates an AsyncStream that produces a single integer worth. This worth is produced by calling yield on the continuation. Each time we now have a worth to ship, we must always name yield on the continuation with the worth that we need to ship.

If we’re constructing an AsyncStream that wraps a delegate based mostly API, we are able to maintain on to our continuation within the delegate object and name yield every time a related delegate methodology is known as.

For instance, we may name continuation.yield from inside a CLLocationManagerDelegate every time a brand new consumer location is made out there to us:

class AsyncLocationStream: NSObject, CLLocationManagerDelegate {
    lazy var stream: AsyncStream<CLLocation> = {
        AsyncStream { (continuation: AsyncStream<CLLocation>.Continuation) -> Void in
            self.continuation = continuation
        }
    }()
    var continuation: AsyncStream<CLLocation>.Continuation?

    func locationManager(_ supervisor: CLLocationManager, didUpdateLocations places: [CLLocation]) {

        for location in places {
            continuation?.yield(location)
        }
    }
}

The instance above is a really naive place to begin for creating an async stream of consumer places. There are a few issues we don’t absolutely take into consideration equivalent to cancelling and beginning location statement or asking for location permissions.

At its core although, this instance is a superb place to begin for experimenting with async streams.

Be aware that this strategy is not going to await customers of your async stream to devour a worth absolutely earlier than you’ll be able to ship your subsequent worth down the stream. As an alternative, all values that you just ship will probably be buffered in your async stream by default which can or will not be what you need.

In sensible phrases which means if you ship values down your stream quicker than the consuming for loop can course of these values, you’ll find yourself with a buffer stuffed with values that will probably be delivered to the consuming for loop with a delay. This may be precisely what you want, but when the values you ship are considerably time delicate and ephemeral it could probably make sense to drop values if the consuming for loop isn’t able to obtain values.

We may determine that we by no means need to maintain on to greater than 1 location and that we solely need to buffer the final identified location to keep away from processing stale information. We will do that by setting a buffering coverage on our async stream:

lazy var stream: AsyncStream<CLLocation> = {
    AsyncStream(bufferingPolicy: .bufferingNewest(1)) { (continuation: AsyncStream<CLLocation>.Continuation) -> Void in
        self.continuation = continuation
    }
}()

This code passes a bufferingPolicy of .bufferingNewest(1) to our AsyncStream. Because of this we are going to solely buffer a single worth if the consuming for loop isn’t processing gadgets quick sufficient, and we are going to discard older values in favor of retaining solely the newest location.

If our stream involves a pure shut, you’ll be able to name end() in your continuation to finish the stream of values.

In case your stream may fail with an error, you can too select to create an AsyncThrowingStream as an alternative of an AsyncStream. The important thing distinction is that buyers of a throwing stream should await new values utilizing attempt await as an alternative simply await. To make your stream throw an error you’ll be able to both name end(throwing:) in your continuation or you’ll be able to name yield(with:) utilizing a Consequence object that represents a failure.

Whereas the fundamentals of constructing an AsyncStream aren’t significantly advanced, we do want to consider how we handle the lifecycles of the issues we create rigorously. Particularly as a result of we’re not speculated to make our continuations outlive our streams which is a very simple mistake to make if you’re bridging present delegate based mostly code.

Managing your stream’s lifecycle

There are basically two methods for an async stream to finish. First, the stream may naturally finish producing values as a result of no additional values might be produced. You’ll name end in your continuation and you’ll present any cleanup that you’ll want to do on the similar time. For instance, you can set the continuation that you just’re holding on to to nil to ensure you can’t by accident use it anymore.

Alternatively, your stream can finish as a result of the duty that’s used to run your async stream is cancelled. Think about the next:

let places = AsyncLocationStream()

let process = Process {
    for await location in places.stream {
        print(location)
    }
}

process.cancel()

When one thing just like the above occurs, we are going to need to be sure that we don’t name yield on our continuation anymore except we begin a brand new stream with a brand new, lively, continuation.

We will detect and reply to the top of our stream by setting an onTermination handler on our continuation:

self.continuation?.onTermination = { end in
    print(consequence)
    self.continuation = nil
}

Ideally we set this handler instantly after we first create our async stream.

Along with the stream being cancelled or in any other case going out of scope, we may break out of our loop which is able to finally trigger our process to complete. That is usually talking not one thing it will finish your async stream so if you need breaking out of your loop to finish your stream, you have to to take this into consideration your self.

Personally, I’ve discovered that the best approach to ensure you do some cleanup is to have some methodology in your stream producing object to cancel the stream as an alternative of simply breaking out of an async for loop. That manner, you’ll be able to carry out cleanup and never have a stream that’s sending values though no person is listening.

It’s additionally vital to keep in mind that the sample I confirmed earlier will solely work if one client makes use of your location stream object. You can not have a number of for loops iterating over a single stream in Swift Concurrency as a result of by default, async sequences lack the flexibility to share their iterations with a number of loops.

In Abstract

On this publish, you discovered quite a bit about async streams and how one can produce your personal async sequences. First, you noticed the unfolding strategy of constructing an async stream and also you discovered that this strategy is comparatively simple however won’t be very helpful for those that must bridge present delegate or callback based mostly APIs.

After exploring unfolding for a bit, we took a take a look at the construct closure for async streams. You discovered that this strategy leverages a continuation object that may be referred to as to supply values if and when wanted.

You noticed a really rudimentary instance of an object that will bridge a CLLocationManager into async await, and also you discovered a however about appropriately managing your continuations to forestall sending values into an already accomplished stream.

When you have any questions or feedback for me about this publish, please be happy to succeed in out on Twitter or on Mastodon.