AsyncStream: convertir callbacks en secuencias

Casi ninguna fuente real de eventos nace hablando el idioma de AsyncSequence. Un delegado llama a un método, un observador recibe una notificación, una biblioteca en C invoca un puntero a función, un temporizador dispara un bloque. Todas esas APIs son de empuje: deciden cuándo hablar y no aceptan que nadie les pida el siguiente valor. AsyncStream es la pieza que traduce ese mundo al de tracción sin obligar a reescribir la fuente: crea una secuencia asíncrona vacía y te entrega un mando a distancia —la continuación— que puedes guardar donde haga falta y usar desde cualquier hilo para empujar elementos dentro. Entender bien ese objeto, su semántica de encolado y su contrato de terminación es la diferencia entre un adaptador correcto y una fuga de memoria que nadie encuentra.

El adaptador y su mando a distancia

La forma canónica envuelve una API de callbacks en el inicializador que recibe una clausura de construcción.

func posiciones(del sensor: Sensor) -> AsyncStream<Posicion> {
    AsyncStream { continuation in
        sensor.alRecibir = { punto in
            continuation.yield(punto)
        }
        sensor.alTerminar = {
            continuation.finish()
        }
        sensor.empezar()
    }
}

La clausura se ejecuta una sola vez, cuando alguien crea el iterador, y su único cometido es entregar la continuación a quien vaya a producir. La continuación es Sendable: puede viajar a otro hilo, guardarse en una propiedad, capturarse en un bloque de C. yield no suspende nunca y puede llamarse desde cualquier contexto, síncrono o asíncrono, aislado o no; esa es exactamente la propiedad que hace posible el puente, porque el código antiguo que la llama no sabe nada de tareas ni de actores.

En el otro extremo, el consumidor no percibe nada de esta maquinaria:

for await p in posiciones(del: sensor) {
    dibujar(p)
}

Hay una regla que separa el uso correcto del incorrecto y conviene enunciarla pronto: si la continuación no sobrevive a la clausura de construcción, el flujo está roto. Registrarla en un objeto, en una propiedad o en un manejador es lo que la mantiene viva; dejarla morir al terminar la clausura produce un flujo que no emite jamás y que además nunca termina, porque nadie podrá llamar a finish.

Semántica de yield y contrato de terminación

yield devuelve un valor que casi todo el mundo descarta y que contiene información importante.

func yield(_ value: Element) -> YieldResult   // el resultado es descartable
// .enqueued(remaining: Int)  el elemento entro en el bufer
// .dropped(Element)          la politica de bufer lo descarto
// .terminated                el flujo ya habia terminado

Ese resultado es el único canal por el que el productor se entera de lo que ocurre aguas abajo: si el búfer está lleno y la política descarta, si el consumidor ya se marchó. Ignorarlo es aceptable en un adaptador simple; en un productor caro —uno que lee un fichero o descodifica imágenes— es la señal que permite dejar de trabajar en vano.

La terminación tiene tres puertas, y las tres conducen al mismo estado final:

continuation.finish()                      // fin normal
continuation.finish(throwing: ErrorRed.caida)   // solo en AsyncThrowingStream
// y la tercera: que la continuacion se destruya sin que nadie llame a finish

finish es idempotente: la segunda llamada no hace nada, igual que un yield posterior, que devolverá .terminated. Los elementos ya encolados se entregan antes de que el bucle vea el final, de modo que terminar no equivale a truncar. Y la tercera puerta importa más de lo que parece: si la continuación se libera sin haber llamado a finish, el flujo se cierra igualmente. Es una red de seguridad contra tareas eternamente colgadas, no una invitación a olvidarse de cerrar.

La variante que lanza cambia solo la firma del error:

func lineas(de socket: Socket) -> AsyncThrowingStream<String, Error> {
    AsyncThrowingStream { continuation in
        socket.alRecibir = { continuation.yield($0) }
        socket.alFallar  = { continuation.finish(throwing: $0) }
        socket.alCerrar  = { continuation.finish() }
    }
}

El consumidor pasa entonces a escribir for try await, y el error interrumpe el bucle como cualquier otro lanzamiento. Un detalle que se olvida: tras un error el flujo está terminado, no pausado. No hay reanudación posible ni reintento implícito; reintentar es responsabilidad de quien consume, envolviendo el bucle.

stateDiagram-v2
[*] --> Inactivo
Inactivo --> Activo: se crea el iterador y corre la clausura
Activo --> Activo: yield encola un elemento
Activo --> Terminado: finish normal
Activo --> Terminado: finish con error
Activo --> Terminado: la continuacion se libera
Terminado --> [*]: onTermination y limpieza
note right of Terminado
  yield posterior devuelve terminated
  finish posterior no hace nada
end note

Tres formas de construir, tres situaciones

El inicializador con clausura no es la única vía, y elegir mal complica código que podría ser trivial.

El desdoblamiento sirve cuando la fuente ya es de tracción y solo hay que repetir una operación asíncrona hasta agotarla. No hay continuación en juego.

let paginas = AsyncStream(unfolding: {
    await api.siguientePagina()      // devolver nil termina el flujo
})

El patrón makeStream, incorporado en Swift 5.9, devuelve la pareja de golpe y resuelve de raíz el problema de la continuación fugitiva, además de permitir que productor y consumidor se creen en momentos distintos.

let (flujo, cont) = AsyncStream.makeStream(of: Posicion.self, bufferingPolicy: .bufferingNewest(32))

final class Rastreador {
    let continuacion: AsyncStream<Posicion>.Continuation
    init(_ c: AsyncStream<Posicion>.Continuation) { continuacion = c }
    func recibir(_ p: Posicion) { continuacion.yield(p) }
    deinit { continuacion.finish() }
}

Ese deinit es el idioma más limpio para atar la vida del flujo a la vida del productor. Y el inicializador con clausura sigue siendo la mejor opción cuando la suscripción a la fuente debe ocurrir justo al empezar a consumir y cancelarse al dejar de hacerlo, porque su clausura se ejecuta de forma perezosa y onTermination le da un lugar natural donde deshacer el registro.

Los tropiezos habituales

Cuatro fallos concentran la mayoría de los problemas reales.

Un solo consumidor. AsyncStream está diseñada para un único bucle. Dos consumidores sobre el mismo flujo no reciben copias: se reparten los elementos de forma impredecible. Si necesitas difusión, construye un tipo que la implemente por encima.

Consumir dos veces. Iterar de nuevo un flujo ya consumido no lo reinicia. Una AsyncStream es un objeto de un solo uso; la receta reutilizable es la función que la fabrica.

Producir sin límite. Si el productor emite más rápido de lo que el consumidor procesa, la política de búfer por defecto —ilimitada— hace crecer la memoria hasta donde llegue. Es el tema entero de la lección siguiente y el motivo de que makeStream pida la política de forma explícita.

Confundir el aislamiento. Que yield sea seguro desde cualquier hilo no convierte en seguro lo que hagas alrededor. Mutar estado compartido dentro del callback de la fuente sigue siendo tu problema, y el compilador en modo estricto lo señalará como tal.