Trabajando con iteraciones asíncronas usando AsyncSequence

Libranner Santos

22 mayo, 20234min de lectura

En algunos casos, es necesario iterar sobre una secuencia de valores que se emiten de forma asíncrona. Swift proporciona una API que permite hacer esto fácilmente utilizando un simple bucle for, gracias al protocolo AsyncSequence.

Características de AsyncSequence

• Causará una pausa en cada iteración y continuará cuando finalice su ejecución.
• Podemos utilizar continue y break, al igual que en cualquier otro bucle.
• Puedes utilizar las mismas funciones que estás acostumbrado a utilizar con secuencias regulares. Funciones como map, reduce, dropFirst...
• Es similar a una secuencia, pero de forma asíncrona. Cada elemento se entrega de forma asíncrona.
• Puede lanzar una excepción.
• Finaliza al llegar al final o al ocurrir un error. Si ocurre un error, devolverá nil para cualquier llamada posterior a next en su iterador.
• Si la secuencia asíncrona puede lanzar una excepción, utlizamos for try await in.
• Si necesitamos ejecutar la iteración de forma concurrente a otras tareas en curso, puedes crear una nueva tarea asíncrona que encapsule la iteración. Por ejemplo:

Task {
 for await element in list {
  ...
 }
}

Task {
 for await element in anotherList {
  ...
 }
}

En este caso tenemos dos AsyncSequence, pero como las encapsulamos dentro de un Task cada una, el código puede ejecutarse en un hilo diferente y no bloquear el resto del código en la función.

Cómo cancelar una iteración

Para esto necesitamos tener una referencia al Task y ejecutar el método cancel(). Veamos un ejemplo:

let iterator = Task {
  for await event in asyncEvents {
    ...
  }
}

iterator.cancel()

Creando un AsyncSequence usando AsyncStream

Podemos crear secuencias asíncronas utilizando AsyncStream. Veamos un ejemplo, comenzando con la clase ParkingMonitor:

class ParkingMonitor {
  var handler: ((Vehicle) -> Void)?
  func start() {}
  func stop() {}
}

Los métodos start() y stop() inician y detienen el proceso de seguimiento para los vehículos que entran a un parqueo. Mientras que handler nos permite asignar un closure que recibe un Vehicle.

Para convertir esta clase en un AsyncSequence, utilizamos el siguiente código:

let vehicles = AsyncStream(Vehicle.self) { continuation in
    let monitor = ParkingMonitor()
    monitor.handler = { vehicle in
        continuation.yield(vehicle)
    }
    continuation.onTermination = { @Sendable _ in
        monitor.stop()
    }
    monitor.start()
}

Con este código creamos una instancia de AsyncStream que acepta objetos de tipo Vehicle, y pasamos un cierre en el cual configuramos nuestro ParkingMonitor. Gracias a continuation, podemos utilizar el método yield(_ value:) para emitir valores. También, gracias a onTermination, indicamos que queremos llamar a stop() del monitor cuando dejemos de utilizar este AsyncStream.

Opcionalmente, podemos pasar el parámetro bufferingPolicy. Por defecto, usa la política .unbounded, que almacena un número ilimitado de elementos en el búfer. También puedes elegir almacenar solo los más antiguos (.bufferingOldest(Int)) o los más nuevos (.bufferingNewest(Int)), según tus necesidades.

Gracias a que ahora tenemos un AsyncStream, podemos utilizar for await para imprimir los modelos de los vehículos que ingresan al estacionamiento:

for await vehicle in vehicles {
    print(vehicle.model)
}

APIs de Swift que usan AsyncSequence

Algunas APIs disponibles en Swift hacen uso de AsyncSequence para facilitar el uso de las mismas y hacer uso del nuevo modelo de concurrencia.

Estas son algunas de las APIs disponibles:

• Puedes leer bytes de manera asíncrona desde un FileHandle.

for try await line in FileHandle.standardInput.bytes.lines {
  ...
}

• Leer bytes de manera asíncrona o leer líneas de manera asíncrona desde una URL:

let url = URL(fileURLWithPath: "/tmp/temp.txt")
for try await line in url.lines {
  ...
}

• Leer bytes de forma asíncrona desde un URLSession:

let (bytes, response) = try await URLSession.shared.bytes(from: url)

guard let httpResponse = response as? HTTPURLResponse,
      httpResponse.statusCode == 200 /*OK*/
else {
    throw MyNetworkingError.invalidServerResponse
}

for try await byte in bytes {
    ...
}

• Puedes usar async con el API de notificaciones:

let center = NotificationCenter.default
let notification = await
  center.notifications(named: .NSPersistentStoreRemoteChange).first {
    $0.userInfo[NSStoreUUIDKey] == storeUUID
}