AsyncSequence y AsyncStream

Al terminar, PartidoEnVivo tendrá eventos: AsyncStream<EventoDePartido>: un flujo que va soltando los goles, tarjetas y cambios del partido minuto a minuto, y que puedes recorrer con for await. Además tendrá una narración escrita con map, un filtro de goles y primerGol(), que se queda con el primer gol y deja de jugar el resto. El marcador gol a gol llega en la lección siguiente.

Importa porque un partido en directo no es una lista que ya existe: sus datos van llegando. Swift los modela con secuencias asíncronas, el equivalente de Flow en Kotlin, y entender en qué se parecen y en qué no evita varios errores.

Temas: AsyncSequence y AsyncStream, y su parecido con Flow.

AsyncSequence y AsyncStream

Una secuencia es algo que se recorre con for … in. Una secuencia asíncrona (AsyncSequence) es igual, pero cada elemento puede tardar en llegar, así que el bucle se escribe for await y se pausa entre elemento y elemento. Se puede recorrer solo desde código async.

AsyncStream es la forma más directa de fabricar una: le das una closure que recibe una continuación (continuation), y con ella entregas valores (yield) y cierras el flujo (finish()).

Ejemplo mínimo: frío gracias a una propiedad

La closure de AsyncStream { … } se ejecuta cuando se crea el stream, no cuando alguien empieza a recorrerlo: lo que entrega antes de que llegue el consumidor queda guardado en un búfer (una cola de espera donde se guardan los valores que nadie ha recogido todavía; en AsyncStream no tiene límite de tamaño por defecto). Un stream ya recorrido está agotado: recorrerlo otra vez no da nada. Entonces, ¿cómo se consigue que cada espectador vea un partido nuevo desde el minuto 0, como el flow { } de Kotlin? Con una propiedad calculada (var x: Tipo { … }, la que viste en Propiedades: calculadas, didSet y lazy): cada vez que la lees, ejecuta su código y crea un stream nuevo.

// Un struct con una propiedad calculada: cada lectura de `atras` crea un stream nuevo.
struct Cuenta {
    var atras: AsyncStream<Int> {
        // c es la continuación: con ella se entregan valores.
        AsyncStream { c in
            print("(la closure corre)")
            // yield entrega un valor al consumidor, uno detrás de otro.
            for i in [3, 2, 1] { c.yield(i) }
            // finish cierra el stream: el for await de quien lo recorre termina.
            c.finish()
        }
    }
}
 
let cuenta = Cuenta()
let s = cuenta.atras                                    // aquí se crea el stream y corre la closure
print("stream creado")
for await n in s { print(n) }                           // recorre los tres valores
for await n in s { print("segundo recorrido", n) }      // no imprime nada: ya está agotado
for await n in cuenta.atras { print("nuevo stream", n) } // otra lectura: un stream nuevo, y la closure corre otra vez

Salida:

(la closure corre)
stream creado
3
2
1
(la closure corre)
nuevo stream 3
nuevo stream 2
nuevo stream 1

Fíjate en que (la closure corre) sale antes de stream creado, y en que el segundo recorrido no imprime nada. Cada lectura de cuenta.atras es un stream nuevo; el mismo stream, recorrido dos veces, no se repite.

Kotlin y Swift, lado a lado

Un solo consumidor por valor significa que, si dos tareas recorren el mismo stream, se reparten los valores: una recibe unos y la otra los demás. Para varios oyentes de lo mismo se necesita otra cosa, que verás en las lecciones de SwiftUI.

onTermination: cuando el consumidor se va

Si el productor es una Task que has lanzado tú, hay que pararla cuando el consumidor se marcha (porque hizo break, porque lo cancelaron o porque terminó). Para eso está continuacion.onTermination: una closure que corre cuando el stream termina o cuando quien lo recorre se cancela. Sin esa línea, el partido seguiría jugándose para nadie.

Un matiz sobre cuándo corre tras un break. Si recorres el stream sin guardarlo (for await x in reloj(), o con first { … }, como hace primerGol), corre al salir del bucle. Si lo guardas en una constante (let guardado = reloj()), onTermination espera a que esa constante deje de existir, que es al acabar la función. El ejercicio de abajo lo muestra.

Operadores, some y Never

AsyncSequence trae map, filter, compactMap y first(where:), como las colecciones, y cada uno devuelve otra secuencia asíncrona: no corre nada hasta el for await. compactMap deja pasar solo los elementos para los que la closure devuelve un valor y descarta los nil.

El tipo de retorno some AsyncSequence<String, Never> (some, lo viste en some y any) dice «una secuencia asíncrona que entrega String» sin nombrar el tipo exacto. Los dos datos entre < y > son, en orden, qué entrega (String) y cómo puede fallar (Never, que significa que nunca falla). Esa forma de escribirlo pide macOS 15, la versión que pusiste en platforms en async/await y Task.

Ejemplo mínimo: un semáforo con productor, onTermination y operadores

Un solo ejemplo con casi todas las piezas, sobre un semáforo que cambia de color en vez de un partido. No es tu código de Marcador: pégalo en tu Playground. Fíjate en la copia let colores = self.colores: capturar self es que una closure o una tarea guarde una referencia al objeto, y mientras la tarea exista, el objeto no se libera (ARC). Con la copia, la tarea guarda solo la lista de colores, no el semáforo.

// Una clase con la lista de colores por los que pasa el semáforo.
final class Semaforo {
    let colores = ["verde", "ámbar", "rojo", "verde"]
 
    // Propiedad calculada: cada lectura crea un stream nuevo (frío).
    var cambios: AsyncStream<String> {
        let colores = self.colores                 // una copia: así la tarea no guarda una referencia a self
        return AsyncStream { continuacion in       // continuacion: con ella se entregan valores
            let tarea = Task {                     // el productor: una Task que lanzamos nosotros
                for color in colores {
                    // Si cancelan la tarea, Task.sleep lanza; el catch sale del bucle.
                    do { try await Task.sleep(for: .milliseconds(20)) } catch { break }
                    continuacion.yield(color)      // entrega un valor al consumidor
                }
                continuacion.finish()              // cierra el stream: el for await termina
            }
            // Corre al terminar o cuando el consumidor se va: para al productor.
            continuacion.onTermination = { _ in    // _ : el parámetro que no usamos
                tarea.cancel()                     // sin esto, seguiría cambiando para nadie
            }
        }
    }
 
    // map devuelve otra secuencia asíncrona; nada corre hasta el for await.
    var avisos: some AsyncSequence<String, Never> {   // entrega String y nunca falla
        cambios.map { "Ahora está en \($0)" }
    }
 
    // compactMap con el tipo de retorno escrito antes de in: deja pasar solo los rojos.
    var soloRojos: some AsyncSequence<String, Never> {
        cambios.compactMap { color -> String? in
            color == "rojo" ? color : nil
        }
    }
}
 
let semaforo = Semaforo()
for await aviso in semaforo.avisos { print(aviso) }                 // cuatro avisos, uno cada 20 ms
for await rojo in semaforo.soloRojos { print("Rojo detectado:", rojo) }
// first mira los colores hasta encontrar el primero que cumple y cancela el resto.
let primero = await semaforo.cambios.first { $0 != "verde" }
print(primero ?? "ninguno")                                          // ámbar

Salida:

Ahora está en verde
Ahora está en ámbar
Ahora está en rojo
Ahora está en verde
Rojo detectado: rojo
ámbar

tic 1
tic 2
productor cancelado
salí del for
fin

func reloj() -> AsyncStream<Int> {
    AsyncStream { c in
        let t = Task {
            var s = 0
            while !Task.isCancelled {
                s += 1
                c.yield(s)
                try? await Task.sleep(for: .milliseconds(20))
            }
            c.finish()
        }
        c.onTermination = { _ in
            print("productor cancelado")
            t.cancel()
        }
    }
}
 
func hastaElDos() async {
    for await s in reloj() {
        print("tic", s)
        if s == 2 { break }
    }
    print("salí del for")
}
 
await hastaElDos()
try? await Task.sleep(for: .milliseconds(100))
print("fin")

Paso · eventos en PartidoEnVivo

Objetivo: PartidoEnVivo.eventos como AsyncStream<EventoDePartido>, la narración, el filtro de goles y primerGol(). Al terminar, torneo-cli imprime la narración de un partido, un segundo recorrido y el primer gol.

Trabajas en el paquete Torneo que creaste en Tu primer paquete de Swift, en la carpeta ~/Marcador-Swift/Torneo. Todos los comandos de esta lección se escriben en la Terminal, dentro de esa carpeta:

cd ~/Marcador-Swift/Torneo

No hay ficheros nuevos: lo que ya existe se modifica.

Piezas nuevas de este paso:

• AsyncStream { continuacion in … }, continuacion.yield(x), continuacion.finish() y continuacion.onTermination = { _ in … }: las que acabas de ver en el semáforo. { _ in … } es una closure con un parámetro que no usas; el _ lo ignora.
• for await evento in origen { … }: recorre una secuencia asíncrona.
• eventos.compactMap { evento -> EventoDePartido? in … }: el tipo que devuelve la closure puede escribirse a la vista, entre la lista de parámetros y in.
• eventos.map(describir): describir es la función de EventoDePartido.swift que convierte un evento en texto; se pasa como closure sin llamarla, como en Closures con listas.
• guion.eventos.map(\.minuto).max() ?? 0: \.minuto es una ruta de propiedad (key path, de Colecciones y key paths) usada como función; max() da el mayor, o nil si no hay elementos, y ?? 0 pone un 0 en ese caso.
• private let alTerminarLaEmision: (() -> Void)? y alTerminar?(): una closure opcional, que no recibe nada ni devuelve nada, y se llama solo si no es nil (el ? antes de los paréntesis).
• do { try await … } catch { break }: si la espera lanza un error (porque cancelaron la tarea), sale del bucle.
• El init de PartidoEnVivo gana dos parámetros con valor por defecto, entre guion y alLiberarse (que ya tenías): msPorMinuto, los milisegundos reales por minuto de partido (10 si no se dice), y alTerminarLaEmision, una closure opcional que se llama cuando termina la emisión. main.swift y los tests usan esos nombres.

Intenta escribirlo tú, en este orden:

1. En PartidoEnVivo, los dos parámetros nuevos del init y eventos, un stream que cada vez que se recorre juega el partido desde el principio: entrega los eventos del minuto 0, y luego, minuto a minuto hasta el 90 (o el último evento), espera lo que dura un minuto y entrega los de ese minuto; al acabar, se cierra. Si el consumidor se va, el productor se para y se llama a la closure de fin de emisión. El productor usa copias del guion y del tiempo, no el propio partido, para no mantenerlo vivo: una closure que lo captura (guarda una referencia a él) lo mantiene con vida mientras exista (ARC explica quién mantiene vivo a quién, y Retain cycles lo lleva a los closures).
2. A partir de eventos: narracion, cada evento descrito como texto; golesEnFrio, solo los goles; y primerGol, que devuelve el primer gol y deja de jugar el resto.
3. Al final de main.swift, con el título «== Un AsyncStream frío: la narración del partido ==» y un partido a 2 ms por minuto: la narración, con cada línea sangrada; cuántos eventos tiene un segundo recorrido; y el primer gol.

Lo que debes ver al ejecutar, tras el título:

  12' Gol de Ana (Rayo FC)
  30' Tarjeta amarilla para Iván
  55' Gol de Iván (Toros)
  80' Gol de Ana (Rayo FC)
  85' Cambio: sale Ana, entra Luis
Segundo recorrido: otro partido completo (5 eventos)
Primer gol: 12' Gol de Ana (Rayo FC) (el resto del partido se canceló)

Cuando termines, compara con la solución.

    public let guion: Partido
    public let msPorMinuto: Int
 
    // Se llama cuando termina la emisión de `eventos` (por final o por cancelación).
    private let alTerminarLaEmision: (() -> Void)?
 
    // Se llama en `deinit`, es decir, justo cuando ARC libera el objeto. Sirve
    // para que las pruebas (y la consola) vean que de verdad se liberó.
    private let alLiberarse: (() -> Void)?
 
    // `strong` es lo normal: esta referencia mantiene vivo al narrador.
    public var narrador: Narrador?
 
    public init(
        guion: Partido,
        msPorMinuto: Int = 10,
        alTerminarLaEmision: (() -> Void)? = nil,
        alLiberarse: (() -> Void)? = nil
    ) {
        self.guion = guion
        self.msPorMinuto = msPorMinuto
        self.alTerminarLaEmision = alTerminarLaEmision
        self.alLiberarse = alLiberarse
    }

    public var eventos: AsyncStream<EventoDePartido> {
        let guion = self.guion
        let ms = self.msPorMinuto
        let alTerminar = self.alTerminarLaEmision
        return AsyncStream { continuacion in
            let tarea = Task {
                for evento in guion.eventos where evento.minuto == 0 {
                    continuacion.yield(evento)
                }
                let ultimoMinuto = max(minutosDelPartido, guion.eventos.map(\.minuto).max() ?? 0)
                for minuto in 1...ultimoMinuto {
                    // El productor no puede lanzar errores por el stream: si nos cancelan, salimos.
                    do { try await Task.sleep(for: .milliseconds(ms)) } catch { break }
                    for evento in guion.eventos where evento.minuto == minuto {
                        continuacion.yield(evento)
                    }
                }
                continuacion.finish()
            }
            continuacion.onTermination = { _ in
                tarea.cancel()
                alTerminar?()
            }
        }
    }
 
    // Operadores: `AsyncSequence` trae `map`, `filter`, `compactMap`, `first(where:)`…
    // y cada uno devuelve OTRA secuencia asíncrona: nada corre hasta el `for await`.
    public var narracion: some AsyncSequence<String, Never> {
        eventos.map(describir)
    }
 
    // `compactMap` deja pasar solo los goles y descarta lo demás.
    public var golesEnFrio: some AsyncSequence<EventoDePartido, Never> {
        eventos.compactMap { evento -> EventoDePartido? in
            if case .gol = evento { evento } else { nil }
        }
    }
 
    // `first(where:)` se queda con el primero y CANCELA el resto (al salir del
    // `for await` interno, `onTermination` cancela la tarea): el partido no se sigue
    // jugando por nada. Es el `first()` de Kotlin.
    public func primerGol() async -> EventoDePartido? {
        await eventos.first { evento in
            if case .gol = evento { true } else { false }
        }
    }

// Un título para separar esta parte de la salida anterior.
print("== Un AsyncStream frío: la narración del partido ==")
// Un partido en vivo muy rápido: 2 ms por minuto de juego.
let enVivoRapido = PartidoEnVivo(guion: Ejemplo.rayoContraToros, msPorMinuto: 2)
// La narración es una secuencia asíncrona de frases; aún no ha corrido nada.
let narracion = enVivoRapido.narracion
// `for await` recorre la narración y se pausa entre una frase y la siguiente.
for await linea in narracion { print("  \(linea)") }
 
// Cada recorrido es un partido nuevo: el frío se repite entero.
// Un contador de eventos…
var cuantos = 0
// …que suma uno por cada evento del segundo partido.
for await _ in enVivoRapido.eventos { cuantos += 1 }
print("Segundo recorrido: otro partido completo (\(cuantos) eventos)")
 
// `primerGol()` devuelve el primer gol y cancela el resto del partido.
if let gol = await enVivoRapido.primerGol() {
    print("Primer gol: \(describir(gol)) (el resto del partido se canceló)")
}

Ejecuta, en la Terminal y desde ~/Marcador-Swift/Torneo, xcrun swift test y xcrun swift run torneo-cli. Sin contar las líneas de título que empiezan por ==, la salida de torneo-cli termina así:

  12' Gol de Ana (Rayo FC)
  30' Tarjeta amarilla para Iván
  55' Gol de Iván (Toros)
  80' Gol de Ana (Rayo FC)
  85' Cambio: sale Ana, entra Luis
Segundo recorrido: otro partido completo (5 eventos)
Primer gol: 12' Gol de Ana (Rayo FC) (el resto del partido se canceló)

Esta lección no añade tests: los de los streams llegan en El marcador gol a gol, junto con el Marcador.