New Package: Look at Swift Async Algorithms

About a month ago, Apple released the first version of the async swift algorithms package. It provides tools and algorithms to use with the introduced not that far ago asynchronous sequence. The package focuses on implementing already well-known tools like zip as well as new features that transact in time (wow). It also makes available more sophisticated ways of creating and managing asynchronous sequences.

Installation

The new package is distributed through Swift PM. To add it to your project, you need to add it as a dependency in the Xcode project File > Add Packages.

Or add it to your Package.swift file:

.package(url: "https://github.com/apple/swift-async-algorithms"),

Don't forget to also add the dependency to the executable:

.target(name: "<target>", dependencies: [
    .product(name: "AsyncAlgorithms", package: "swift-async-algorithms"),
]),

The module will be available in your project after adding import AsyncAlgorithms.

Creating asynchronous sequences

To test all the beautiful functions the new module provides, we need to create an async sequence at first. And the package introduces new ways of doing so.

Property async

The module adds the following extension to Sequence protocol.

extension Sequence {
  public var async: AsyncLazySequence<Self> { get }
}

Where AsyncLazySequence conforms to AsyncSequence.

public struct AsyncLazySequence<Base: Sequence>: AsyncSequence {
}

extension AsyncLazySequence: Sendable where Base: Sendable {
	...
}
extension AsyncLazySequence.Iterator: Sendable where Base.Iterator: Sendable {
}

let numbers = [1, 2, 3, 4].async
let characters = "Hello, world".async
let items = [1: "one", 2: "two", 3: "three"].async

However, creating AsyncSequence this way does not really bring benefits as all elements are already here and available right away. There are more useful ways of creating AsyncSequence.

AsyncChannel and AsyncThrowingChannel

If you know what Future or Promise in other languages are, then AsyncChannel will be familiar to you. Except that it provides a way of transferring a sequence of values.

Here's a pretty straightforward example of AsyncChannel usage.

let channel = AsyncChannel<String>()
Task {
    for word in ["Hello", "from", "async", "channel"] {
      await channel.send(word)
    }
    await channel.finish()
}

for await message in channel {
    print(message)
}

Hello
from
async
channel

Notice that await keyword is used with send and finish. This is because the channel is actually both ways synchronized. That means that send awaits consumption and vice versa.

AsyncThrowingStream is almost the same except that it provides fail(_ error: Error) method that can be used to throw an exception to the channel's consumer.

let channel = AsyncThrowingChannel<String, Error>()

...

for try await message in channel {
    print(message)
}

And converting back

The module adds initializers for three primary types: Array, Dictionary, and Set that let you transform the async sequence to the regular one by fetching all elements during init.

let table = await Dictionary(uniqueKeysWithValues: zip(keys, values))
let allItems = await Set(items.prefix(10))
let allMessages = await Array(channel)

Manipulating asynchronous sequences

The module also provides new ways of combining asynchronous sequences. These functions are pretty straightforward.

• chain(_ s1: AsyncSequence, _ s2: AsyncSequence)

Chains two or three asynchronous sequences together sequentially where the elements from the result are comprised in order from the elements of the first asynchronous sequence and then the second (and so on) or until an error occurs. Sequences must have the same Element type.

• joined() or joined(separator: AsyncSequence)

Concatenates an asynchronous sequence of asynchronous sequences together where the result is comprised in order from the elements of the first asynchronous sequence and then the second (and so on) or until an error occurs. Similar to chain()except the number of asynchronous sequences to concatenate is not known upfront. The separator also can be specified.

• combineLatest(_ base1: AsyncSequence, _ base2: AsyncSequence)

Combines two or more sequences, producing tuples of the latest values available from the sequence.

• merge(_ base1: AsyncSequence, _ base2: AsyncSequence)

Merges sequences into a new one. The result is a combination of results from two sequences. Sequences must have the same Element type.

• zip(_ base1: AsyncSequence, _ base2: AsyncSequence)

The same as a regular zip but for AsyncSequence. Differs from combineLatest as it waits until the second value is available and does not use the last value.

Time-related functions

Sounds awesome, but Swift is not powerful enough to put await before the time itself. When events can potentially happen faster than the desired consumption rate, there are ways to handle the situation. These functions allow linking AsyncSequences with time. They can be applied to any AsyncSequence.

For both listed methods, a custom clock can be specified. By default, it's ContinuousClock

Debounce

 public func debounce<C: Clock>(
    for interval: C.Instant.Duration, 
    tolerance: C.Instant.Duration? = nil, 
    clock: C
  ) -> AsyncDebounceSequence<Self, C>

The debounce algorithm produces elements after a particular duration has passed between events. If there are a lot of events happening, debounce will wait until at least interval of time elapsed from the last event before emitting value.

seq.debounce(for: .seconds(1))

In this case, it transforms a potentially fast asynchronous sequence of events into one that waits for a window of 1 second with no events to elapse before emitting a value.

Throttle

extension AsyncSequence {
  public func throttle<C: Clock, Reduced>(
    for interval: C.Instant.Duration, 
    clock: C, 
    reducing: @Sendable @escaping (Reduced?, Element) async -> Reduced
  ) -> AsyncThrottleSequence<Self, C, Reduced>
  
  public func throttle<Reduced>(
    for interval: Duration, 
    reducing: @Sendable @escaping (Reduced?, Element) async -> Reduced
  ) -> AsyncThrottleSequence<Self, ContinuousClock, Reduced>
  
  public func throttle<C: Clock>(
    for interval: C.Instant.Duration, 
    clock: C, 
    latest: Bool = true
  ) -> AsyncThrottleSequence<Self, C, Element>
  
  public func throttle(
    for interval: Duration, 
    latest: Bool = true
  ) -> AsyncThrottleSequence<Self, ContinuousClock, Element>
}

The throttle algorithm produces elements such that at least a specific interval has elapsed between them. If values are produced by the base AsyncSequence the throttle does not resume its next iterator until the period has elapsed or unless a terminal event is encountered. Similarly to debounce, a custom clock can be specified.

seq.throttle(for: .seconds(1))

In this case, the throttle transforms a potentially fast asynchronous sequence of events into one that waits for a window of 1 second to elapse before emitting a value.

Final notes

It's actually frankly entertaining to watch how Swift unfolds new features and how they are developed. Definitely check the project's GitHub mentioned in references to check out the module's source code.

If you feel not really confident with relatively new swift concurrency features, check out my quick guide to async/await in Swift.

References