Implement an AsyncSequence

👍

· May 20, 2023 · 4 mins read

We have known AsyncSequence is a great part of Swift Concurrency feature and learnt the difference between it and Combine Publishers from this article. We’re going to look into ways of creating of an AsyncSequence in this one.

Conform to AsyncSequence protocol

Firstly, we can create one by conforming to AsyncSequence protocol. For which, we need to define the Element type and provide an implementation of this method func makeAsyncIterator() -> AsyncIterator .

In the sample code below, we implemented a DiceRolling struct to simulate rolling a dice and get the values of dice of each rolling in the sequence. Hence typealias Element = Int . In order to implement makeAsyncIterator method, we created struct AsyncIterator , which had to conform to AsyncIteratorProtocol .

The next() method is the only requirement of AsyncIteratorProtocol, which is the place the values of the sequence are generated. For each call of this method, we decrease the times to make sure we only generate the right amount of the values needed. If need a new value, we simply sleep for two seconds usingTask.sleep(nanoseconds: 2_000_000_000) to simulate the rolling behaviour and return a random value at last through return (1…6).randomElement() . If no more values needed, we return nil to finish this iterator and hence finish the sequence.

struct DiceRolling: AsyncSequence {  
    typealias Element = Int  
    let times: Int  
  
    init(times: Int = 3) {  
        self.times = times  
    }  
  
    struct AsyncIterator: AsyncIteratorProtocol {  
        var times: Int  
  
        mutating func next() async -> Int? {  
            guard !Task.isCancelled, times > 0 else { return nil }  
            do {  
                try await Task.sleep(nanoseconds: 2_000_000_000)  
            } catch {  
                return nil  
                // Ignore error handling here for simpliciry.  
            }  
            times -= 1  
            return (1...6).randomElement()  
        }  
    }  
  
    func makeAsyncIterator() -> AsyncIterator {  
        AsyncIterator(times: times)  
    }  
}  
  
for await diceValue in DiceRolling(times: 5) {  
    print(diceValue)  
}

Use AsyncStream/AsyncThrowingStream

Another way to implement an AsyncSequence is to use AsyncStream or its counterpart AsyncThrowingStream with error throwing feature. Apple provides a great example of using this type to wrap existing code into AsyncSequence here.

The sample code below is to demo how to create an AsyncSequence with similar functionality as above example to simulate rolling dice and get the values of dice in the AsyncSequence. Here, we use a serial queue to handle the task and generate value with two seconds delay.

struct DiceRollingStream {  
    let queue = DispatchQueue(label: "test.asyncstream.queue")  
    let times: Int  
  
    init(times: Int = 3) {  
        self.times = times  
    }  
  
    var stream: AsyncStream<Int> {  
        AsyncStream { continuation in  
            (1...times).forEach { \_ in  
                queue.async {  
                    Thread.sleep(forTimeInterval: 2)  
                    continuation.yield((1...6).randomElement() ?? 1)  
                }  
            }  
            queue.async {  
                continuation.finish()  
            }  
        }  
    }  
}  
  
for await diceValue in DiceRollingStream().stream {  
    print(diceValue)  
}

Use .async from AsyncAlgorithms package

If we have a sequence already and want to convert it to an async one, we can use .async syntax provided by AsyncAlgorithms package.

In the following sample code, we just simple shuffle the array[1, 2, 3, 4, 5, 6] to simulate the dice rolling effect and get AsyncSequence out using .async . Lastly we get first three values by using prefix(3) to simulate we only roll the dice three times.

for await diceValue in [1, 2, 3, 4, 5, 6].shuffled().async.prefix(3) {  
    print(diceValue)  
}

Use AsyncChannel/AsyncThrowingChannel from AsyncAlgorithms package

These are important types provided by AsyncAlgorithms package regarding AsyncSequence. Compared with AsyncStream/AsyncThrowingStream, the main difference here is that they provide back pressure management so that the consumers of the sequence will never be overloaded by sequence generating values too quickly. Please refer to the detail explanation here.

class DiceRollingChannel {  
    let channel = AsyncChannel<Int>()  
  
    init(times: Int = 3) {  
        Task {  
            for _ in 1...times {  
                do {  
                    try await Task.sleep(nanoseconds: 2_000_000_000)  
                } catch {  
                    // Ignore error handling here for simpliciry.  
                    // Can use AsyncThrowingChannel to handle error if needed  
                }  
                await channel.send((1...6).randomElement() ?? 1)  
            }  
            channel.finish()  
        }  
    }  
}  
  
for await diceValue in DiceRollingChannel().channel {  
    print(diceValue)  
}

.values from Combine publishers

Last but not least, for all the Combine publishers, we can simply use .values to get the AsyncSequence as mentioned in this article.

All the code samples are available here in this repo.

Coding life