Efectos de larga vida: escuchar un stream y emitir acciones sin parar

Hasta aquí el efecto ha sido un encargo con final: pide, espera, envía una acción y muere. Pero el mismo constructor sostiene la figura opuesta, la del efecto que no termina nunca porque su trabajo consiste precisamente en no terminar: escuchar. Una suscripción a notificaciones del sistema, un flujo de posiciones del GPS, los mensajes de un websocket, el tictac de un reloj. La única diferencia técnica con una petición puntual es que la closure no retorna: se queda iterando un stream y emitiendo una acción por cada elemento. Esa diferencia mínima en el código es una diferencia enorme en responsabilidad, porque un efecto que no termina solo se convierte en una fuga si nadie se ocupa de matarlo.

Un efecto que no retorna

La forma canónica cabe en seis líneas. Dentro de .run hay un bucle for await sobre una secuencia asíncrona, y dentro del bucle un send por cada elemento. Mientras la secuencia produzca elementos, la closure no llega a su última línea y el efecto sigue vivo.

enum CancelID { case ubicacion }

case .alAparecer:
  return .run { send in
    for await posicion in ubicacion.actualizaciones() {
      await send(.nuevaPosicion(posicion))
    }
  }
  .cancellable(id: CancelID.ubicacion, cancelInFlight: true)

case .alDesaparecer:
  return .cancel(id: CancelID.ubicacion)

Ese cancelInFlight: true resuelve un problema que aparece siempre: las acciones de aparición se disparan más de una vez —al volver de otra pantalla, al recomponerse una vista— y sin él acabarías con dos, tres o siete suscripciones idénticas emitiendo la misma acción por duplicado. Con él, cada nueva suscripción sustituye a la anterior con la misma identidad, y solo hay una viva por definición.

Si la secuencia puede fallar, usa for try await y traduce el fallo como aprendiste en la lección anterior. Y si el efecto debe reponerse tras una caída, el patrón es un bucle exterior que reconecta con espera creciente, comprobando siempre la cancelación para no quedarte reintentando contra un servidor caído después de que el usuario haya cerrado la pantalla.

return .run { send in
  var espera = Duration.seconds(1)
  while !Task.isCancelled {
    do {
      for try await mensaje in socket.mensajes() {
        await send(.mensaje(mensaje))
        espera = .seconds(1)
      }
    } catch {
      await send(.conexionPerdida)
    }
    try await clock.sleep(for: espera)
    espera = min(espera * 2, .seconds(30))
  }
}
.cancellable(id: CancelID.socket)

La condición del while es la que distingue un reconector de un bucle infinito: sin ella, cancelar el efecto no bastaría, porque tras la cancelación el bucle volvería a intentar conectarse una vez más. Y que la espera se reinicie tras una conexión exitosa es lo que evita el otro extremo, quedarse esperando treinta segundos por una caída que ya se resolvió.

Identidad y ciclo de vida

Un efecto de larga vida sin identidad es un recurso sin dueño, y en un sistema concurrente todo lo que no tiene dueño acaba siendo una fuga. La identidad se da con .cancellable(id:) y puede ser cualquier valor Hashable; la convención es un enum privado por feature, que evita las colisiones que provocaría usar cadenas.

flowchart LR
M[Mundo emite eventos sin parar] --> AS[AsyncStream de la dependencia]
AS --> E[Efecto run con for await]
E -->|una accion por evento| R[Reducer]
R --> S[State siempre al dia]
C[Accion de parada] -->|cancel por id| E
style E fill:#a6e3a1,color:#11111b
style C fill:#f38ba8,color:#11111b

Cancelar a mano no siempre es necesario, y conviene saber cuándo puedes ahorrártelo. Los operadores de composición cancelan por ti los efectos de la feature hija: cuando un estado opcional bajo ifLet pasa a nulo, o cuando un elemento desaparece de la colección de un forEach, todos los efectos que esa hija había lanzado se cancelan automáticamente. La navegación como dato paga aquí uno de sus dividendos: descartar una hoja no solo la quita de la pantalla, también apaga sus suscripciones. Fuera de esos casos —una suscripción de la propia feature raíz, por ejemplo— el par de acciones de aparición y desaparición con su .cancel(id:) sigue siendo tu responsabilidad.

Convertir el mundo en un AsyncStream

Casi ninguna API del sistema habla el idioma de las secuencias asíncronas: hablan el de los delegados, los callbacks y los observadores. La traducción se hace en la dependencia, no en el reducer, y la pieza central es AsyncStream.makeStream, que devuelve la pareja de stream y continuación.

struct ClienteUbicacion {
  var actualizaciones: @Sendable () -> AsyncStream<Posicion>
}

extension ClienteUbicacion: DependencyKey {
  static let liveValue = Self(
    actualizaciones: {
      let (stream, continuation) = AsyncStream.makeStream(of: Posicion.self)
      let observador = Observador { posicion in continuation.yield(posicion) }
      continuation.onTermination = { _ in observador.detener() }
      observador.empezar()
      return stream
    }
  )
}

La línea decisiva es onTermination. Cuando el efecto se cancela, la iteración termina, el stream se da por acabado y esa closure se ejecuta: es el único punto donde el observador del sistema se apaga. Sin ella tendrías un efecto correctamente cancelado y, por debajo, un GPS encendido para siempre. La cancelación de la tarea y la liberación del recurso son dos cosas distintas, y onTermination es el puente entre ambas. Recuerda además que un AsyncStream está pensado para un solo consumidor: la dependencia debe fabricar uno nuevo en cada llamada, no repartir el mismo entre varias features.

Que la dependencia devuelva un stream es lo que vuelve determinista lo continuo. En un test no esperas a que el sistema emita nada: fabricas tú la pareja, sustituyes la dependencia y decides evento por evento cuándo ocurre cada cosa.

let (stream, continuation) = AsyncStream.makeStream(of: Posicion.self)
let store = TestStore(initialState: Mapa.State()) {
  Mapa()
} withDependencies: {
  $0.ubicacion.actualizaciones = { stream }
}

await store.send(.alAparecer)
continuation.yield(Posicion(lat: 40.4, lon: -3.7))
await store.receive(\.nuevaPosicion) { $0.posicion = Posicion(lat: 40.4, lon: -3.7) }
await store.send(.alDesaparecer)