Instruction

AsyncStream

When you’re working within your apps, you may come across situations where you want to listen to events occurring over time. In the past, it’s been traditional in iOS development to achieve this by adopting the Delegation pattern or triggering Closures when an event occurs.

While these techniques are acceptable approaches, one of the issues they have is their inability to participate within the Swift Concurrency model. This is because neither of the techniques can directly participate within the async / await approach that Swift Concurrency relies on to suspend work and progress with it when the work is complete.

Solving this problem is not trivial, as there are plenty of apps in the world that use these techniques. Moving them to Swift Concurrency would be a large engineering task. The engineers at Apple were aware of this and provided an API called AsyncStream to make the migration to Swift Concurrency smoother.

So, what does AsyncStream do? Take a look at the following example:

class TripMonitor {
  var newTripHandler: ((Trip) -> Void)?

  func startMonitoringForTrips() {
    // some code to begin monitoring for new trips.
    let newTrip = Trip()
    
    newTripHandler?(newTrip)
  }

  func stopMonitoringForTrips() {
    // Stop monitoring for new trips.
  }
}

class TripStore {
  var tripMonitor: TripMonitor
  
  init(tripMonitor: TripMonitor) {
    self.tripMonitor = tripMonitor
    
    self.tripMonitor.newTripHandler = { [weak self] trip in
      self?.handleTrip(trip: trip)
    }
    
    self.tripMonitor.startMonitoringForTrips()
  }
  
  func handleTrip(trip: Trip) {
    // Handle trip
  }
}

Above, you have two classes, TripMonitor and TripStore. TripMonitor is responsible for listening to new Trips being created and informing other classes of the trip using the newTripHandler closure. TripStore uses the TripMonitor class to listen to new trips using the newTripHandler closure. When a new trip is passed through the closure, it’s passed to handleTrip().

As mentioned earlier, there’s an issue with this code when it comes to adopting Swift Concurrency. There’s no way for the closure to contribute to Swift Concurrency’s asynchronous patterns because it doesn’t know how to deal with async / await keywords. You can solve this issue using the AsyncStream API:

class TripMonitor {
  var newTripHandler: ((Trip) -> Void)?

  func startMonitoringForTrips() {

    // some code to begin monitoring for new trips.
    // ...

    let newTrip = Trip()
    newTripHandler?(newTrip)
  }

  func stopMonitoringForTrips() {
    // Stop monitoring for new trips.
  }
}

// 1
extension TripMonitor {

  var trips: AsyncStream<Trip> {
    // 2
    AsyncStream { continuation in
      newTripHandler = { trip in
        // 3
        print("Yielding New Trip: \(trip)")
        continuation.yield(trip)
      }
      // 4
      continuation.onTermination = { @Sendable _ in
        self.stopMonitoringForTrips()
      }

      self.startMonitoringForTrips()
    }
  }
}


class TripStore {
  
  // 5
  let tripMonitor = TripMonitor()

  // 6
  func listenForTrips() async {
    // 7
    for await trip in tripMonitor.trips {
      print("New Trip: \(trip)")
      handleTrip(trip: trip)
    }
    
    print("Stream finished.")
  }

  func handleTrip(trip: Trip) {
    // Handle trip
  }
}

Go through the new code step by step:

1. An extension is added to TripMonitor, containing a computed property called trips. Its type is an AsyncStream passing Trip types.
2. Inside the property, you create an AsyncStream. The Stream itself passes through a continuation parameter, which you will use later. Inside the Stream, you create TripMonitor and assign a closure to newTripHandler.
3. When you newTripHandler is triggered, you pass the trip through to the closure, which passes it to continuation.yield(). This continuation informs the AsyncStream that there’s a new value to inform subscribers about.
4. When you cancel AsyncStream, either through a Task or from cleanup of an object, the onTermination() method is called on the continuation. This gives the AsyncStream an opportunity to stop any work it’s doing. In this case we stop monitoring for trips.
5. The TripMonitor is created inside TripStore, so it can be used later to recieve trips from the AsyncStream.
6. In TripStore, a new method, listenForTrips(), is created. It’s an async method, meaning it can be called asynchronously.
7. Inside listenForTrips(), the AsyncStream is called from TripMonitor. Because it’s part of Swift Concurrency, you can use the for await syntax to wait for new events and process them.

With a few lines, you’ve wrapped a legacy closure-based API in AsyncStream. Providing a wrapper around callbacks and delegated methods so they can be used with Swift Concurrency is one of the main use cases of AsyncStreams.

So, how does AsyncStream achieve this? The magic is that AsyncStream conforms to AsyncSequence, a protocol allowing implementers to receive values asynchronously. It also allows implementers to make use of the for / await / in syntax, so for loops can wait for the next value in the sequence before commencing its next iteration.

You can cancel an AsyncStream if the Task it’s working within is canceled. In this case, the onTermination() callback is called to stop the AsyncStream.

Now that you know about AsyncStream, have a go at using it in the next section.