Combine #6: Operadores de Manipulación de Tiempo

Desplazamiento en el tiempo

delay(for:tolerance:scheduler:options) recibe un flujo de eventos de entrada, los almacena durante el tiempo definido en el parámetro interval (que es un S.SchedulerTimeType.Stride,), y luego los vuelve a emitir, uno a uno, en el scheduler especificado, con el mismo espaciado temporal con que se recibieron.

// Se crea un publisher sobre el que el Timer emitirá valores.
let sourcePublisher = PassthroughSubject<Date, Never>()
// Se crea un publisher retrasado (delay). 
// Se especifica el retraso .seconds()
// Se emiten los valores en el scheduler .main para mostrar en pantalla
let delayedPublisher = sourcePublisher.delay(
  for: .seconds(delayInSeconds),
  scheduler:DispatchQueue.main)
// Se crea un temporizador que emite valores sobre el RunLoop.main
Timer
  .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
  .autoconnect() // Inicializar inmediatamente el temporizador
  .subscribe(sourcePublisher)
  .store(in: &subscriptions)

En el ejemplo anterior se utilizó la versión Publisher del Timer de Foundation. La instancia de tipo Timer.TimerPublisher, conforma el protocolo ConnectablePublisher, lo cual implica que solo se emitirán elementos después de invocar el método .connect(). El operador autoconnect() invoca connect() cuando se suscribe el primer suscriptor.

Acumulando valores

collect(_:options:) recibe un flujo de eventos de entrada, los almacena durante el tiempo definido en el parámetro strategy (que es un Publishers.TimeGroupingStrategy<S>), y luego emite un arreglo de los eventos coleccionados en el scheduler especificado.

// Se crea un publisher sobre el que el Timer emitirá valores.
let sourcePublisher = PassthroughSubject<Date, Never>()
// Se crea un publisher de recolección (collect). 
// Se especifica la ventana de recolección (.byTime o byTimeOrCount)
// Se emiten los valores en el scheduler .main para mostrar en pantalla
let collectedPublisher = sourcePublisher
  .collect(
      .byTime(DispatchQueue.main, .seconds(collectTimeStride))
  )
// Se crea un temporizador que emite valores sobre el RunLoop.main
Timer
  .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
  .autoconnect() // Inicializar inmediatamente el temporizador
  .subscribe(sourcePublisher)
  .store(in: &subscriptions)

La estrategia de agrupamiento (Publishers.TimeGroupingStrategy<S>) puede ser solo por tiempo (.byTime), o por tiempo y conteo (.byTimeOrCount). En el segundo caso, si se llega a acumular tantos eventos como el tope especificado antes del tiempo definido, se emite un arreglo con los eventos coleccionados hasta ese momento.

let collectTimeStride = 2
let collectTimeStride = 4
// Se crea un publisher de recolección (collect). 
// Se especifica la ventana de recolección (.byTime o byTimeOrCount)
// Se emiten los valores en el scheduler .main para mostrar en pantalla
let collectedPublisher = sourcePublisher
  .collect(
      .byTimeOrCount(DispatchQueue.main, 
                     .seconds(collectTimeStride), 
                     collectMaxCount)
  )

En el ejemplo anterior, se emite un arreglo de los eventos coleccionados cuando se alcance el umbral de tiempo de 4 segundos, o el umbral de cantidad de 2 eventos.

Descartando eventos

debounce(for:scheduler:options:) espera el tiempo definido por el parámetro dueTime (de tipo S.SchedulerTimeType.Stride ) después del último elemento emitido por el Publisher de entrada, y luego emite ese último valor. Es MUY importante tener en cuenta que si el Publisher de entrada envía un evento de fin antes del tiempo de espera del debounce, no el operador no podrá re-emitir el evento.

let subject = PassthroughSubject<String, Never>()
// Se crea un debounce con ventana de 1 seg, que re-emite el evento en DQ.main
let debounced = subject
  .debounce(for: .seconds(1.0), scheduler: DispatchQueue.main)
  .share()
// Se crea una suscripción sobre debounced para imprimir los eventos
debounced
  .sink { ... }
  .store(in: &subscriptions)

En el ejemplo anterior se usa el operador share() para crear un solo punto de suscripción al debounce que permite mostrar los mismos resultados al mismo tiempo a todos los suscriptores.

throttle(for:scheduler:latest:) funciona de forma parecida a debounce porque recorta la cantidad de eventos producidos por el flujo de entrada. Sin embargo, hay una diferencia. debounce espera una ventana de tiempo después de recibir el último evento. throttle abre una ventana de tiempo, dada por el parámetro interval, desde que recibe el primer evento, y emite el primer (latest=false) o último (latest=true) elemento recibido. Además, el primer valor de toda la secuencia se emite tan pronto se recibe.

let subject = PassthroughSubject<String, Never>()
// Se crea un throttle con ventana de 1 seg, que re-emite el evento en DQ.main
let throttled = subject
  .throttle(for: .seconds(throttleDelay),
  scheduler: DispatchQueue.main,
  latest: true)
// Se crea una suscripción sobre throttled para imprimir los eventos
throttled
  .sink { ... }
  .store(in: &subscriptions)

Se acabó el tiempo

timeout(_:scheduler:options:customError:) publica un evento de fin (success o failure) si el flujo de entrada (upstream) excede el tiempo definido por interval sin emitir ningún evento. Si customError está en nil, entonces se emitirá un evento de fin exitoso; en caso contrario, se emitirá el error definido en ese closure.

enum TimeoutError: Swift.Error {
  case timedout
}
// Se crea un subject para emitir eventos
let subject = PassthroughSubject<Void, TimeoutError>()
// Se aplica el operador timeout para emitir un evento de fin (en este caso,
// un error .timedout), si no se recibe ningún evento en 5 segundos.
let timeoutSubject = subject
  .timeout(.seconds(5), scheduler: DispatchQueue.main) {
    .timedout
  }

Midiendo tiempo

measureInterval(using:options:) mide y emite el tiempo entre dos eventos recibidos de un flujo de entrada. El tipo de valores emitidos por este Publisher es el TimeInterval del Scheduler pasado por parámetro: Si se pasa DispatchQueue, entonces los valores emitidos están medidos en nano segundos, mientras que al usar RunLoop, se emiten valores en segundos.

cancellable = Timer.publish(every: 1, on: .main, in: .default)
  .autoconnect()
  .measureInterval(using: RunLoop.main)
  .sink { ...) }
// Prints:
//  Stride(magnitude: 1.0013610124588013)
//  Stride(magnitude: 0.9992760419845581)

En el ejemplo anterior se tiene un Timer que emite un evento cada segundo. Recordar que se debe llamar autoconnect() para iniciar el Timer en la primera suscripción. Notar también que los valores emitidos están en el orden de segundos, porque el Scheduler es RunLoop.main.

---

Cuestionario

1. Explica brevemente qué hace delay(for:scheduler:) y cómo conserva el espaciado temporal de los eventos.

2. ¿Cuál es la función del operador autoconnect() en un Timer.TimerPublisher?

3. ¿Qué diferencia hay entre las estrategias .byTime y .byTimeOrCount en el operador collect?

4. ¿En qué se diferencia debounce de throttle?

5. ¿Qué tipo de valor emite measureInterval(using:) y de qué depende su unidad de medida?

6. Si un Publisher con debounce(for:) finaliza antes de que se cumpla el tiempo de espera, ¿qué ocurre? ✅

• [ ] Se emite el último valor igualmente
• [ ] No se emite nada
• [ ] Se retrasa la finalización
• [ ] Se cancela la suscripción

7. En throttle(for:scheduler:latest:), si latest = false, ¿qué valor se emite en cada ventana? ✅

• [ ] El primero recibido
• [ ] El último recibido
• [ ] Todos los valores recibidos
• [ ] Ninguno hasta el final

8. El operador timeout(_:scheduler:options:customError:) emite un error o finaliza exitosamente si... ✅

• [ ] El Publisher termina antes del intervalo
• [ ] El Publisher no emite nada durante el intervalo 
• [ ] El Publisher emite más de un valor
• [ ] Se cancela la suscripción manualmente

9. En measureInterval(using:), los valores emitidos por RunLoop.main están medidos en: ✅

• [ ] Nanosegundos
• [ ] Milisegundos
• [ ] Segundos 
• [ ] Minutos

10. Usando collect(.byTimeOrCount) ocurre una emisión antes de que se termine la ventana de recolección ¿qué pudo haber ocurrido? ✅

• [ ] Se acumuló la cantidad de eventos definida por parámetro
• [ ] El Publisher de entrada envió un evento de fin.
• [ ] Fue un falso positivo
• [ ] La suscripción quedó mal hecha

---

Solución

1. Explica brevemente qué hace delay(for:scheduler:) y cómo conserva el espaciado temporal de los eventos.

Guarda en memoria los eventos recibidos del upstream y los vuelve a emitir, uno a uno, pasado el tiempo de retraso definido por el parámetro interval, respetando el espaciado temporal original.

2. ¿Cuál es la función del operador autoconnect() en un Timer.TimerPublisher?

El Publisher de Timer es Connectable, lo que quiere decir que se debe invocar explícitamente .connect() para que empiece a emitir eventos, a diferencia de los Publishers normales. Que sea autoconnect() quiere decir que se conecta de forma automática cuando recibe la primera suscripción.

3. ¿Qué diferencia hay entre las estrategias .byTime y .byTimeOrCount en el operador collect?

.byTime solo agrupa las entradas y emite el arreglo acumulado en el periodo específicado por parámetro a la estrategia de agrupación. .byTimeOrCount acumula tanto por tiempo y como por cantidad; es decir: emite el arreglo cuando se cumple el periodo o antes, cuando acumule la cantidad especificada.

4. ¿En qué se diferencia debounce de throttle?

debounce emite el último valor recibido, cuando pasa el tiempo de rebote. throttle de forma periódica y emite el primer o último valor recibido en esa ventana de tiempo.

5. ¿Qué tipo de valor emite measureInterval(using:) y de qué depende su unidad de medida?

measureInterval(using:) emite un valor de tipo Stride que representa la distancia entre dos valores, y que también está definido por el Scheduler (Context.SchedulerTimeType.Stride) pasado como argumento al operador.

6. Si un Publisher con debounce(for:) finaliza antes de que se cumpla el tiempo de espera, ¿qué ocurre?

• [✅] No se emite nada
• [ ] Se cancela la suscripción // La suscripción NO SE CANCELA, sino que se completa.

7. En throttle(for:scheduler:latest:), si latest = false, ¿qué valor se emite en cada ventana? 

• [✅] El primero recibido

8. El operador timeout(_:scheduler:options:customError:) emite un error o finaliza exitosamente si… 

• [✅] El Publisher no emite nada durante el intervalo

9. En measureInterval(using:), los valores emitidos por RunLoop.main están medidos en: 

• [✅] Segundos

10. Usando collect(.byTimeOrCount) ocurre una emisión antes de que se termine la ventana de recolección ¿qué pudo haber ocurrido? 

• [✅] Se acumuló la cantidad de eventos definida por parámetro