Combine #6: Operadores de Manipulación de Tiempo

Source: Dev.to

Desplazamiento en el tiempo

delay(for:tolerance:scheduler:options:) recibe un flujo de eventos de entrada, lo almacena durante el tiempo especificado en el parámetro interval (un SchedulerTimeType.Stride) y luego vuelve a emitirlos, uno a uno, en el scheduler indicado, manteniendo el mismo espaciado temporal con el que fueron recibidos.

// Publisher que será alimentado por el Timer.
let sourcePublisher = PassthroughSubject<Int, Never>()

// Publisher retrasado (delay).
// Se especifica el retraso con .seconds() y se emiten los valores
// en el scheduler .main para mostrarlos en pantalla.
let delayedPublisher = sourcePublisher.delay(
    for: .seconds(delayInSeconds),
    scheduler: DispatchQueue.main
)

// Timer que publica valores en el RunLoop.main.
Timer
    .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
    .autoconnect()                 // Inicia el timer inmediatamente.
    .subscribe(sourcePublisher)    // Conecta el timer al publisher de origen.
    .store(in: &subscriptions)

En el ejemplo anterior se utilizó la versión Publisher del Timer de Foundation.
La instancia Timer.TimerPublisher conforma el protocolo ConnectablePublisher, lo que implica que solo emitirá elementos después de invocar el método .connect().
El operador autoconnect() llama a connect() automáticamente 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 (un Publishers.TimeGroupingStrategy) y luego emite un arreglo con los eventos coleccionados en el scheduler especificado.

// Se crea un publisher sobre el que el Timer emitirá valores.
let sourcePublisher = PassthroughSubject<Int, 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 mostrarlos en pantalla
let collectedPublisher = sourcePublisher
    .collect(
        .byTime(DispatchQueue.main, .seconds(collectTimeStride))
    )

// Se crea un temporizador que emite valores sobre RunLoop.main
Timer
    .publish(every: 1.0 / valuesPerSecond, on: .main, in: .common)
    .autoconnect()
    .subscribe(sourcePublisher)
    .store(in: &subscriptions)

Estrategias de agrupamiento

Publishers.TimeGroupingStrategy puede ser:

En el caso de .byTimeOrCount, si se acumulan tantos eventos como el tope especificado antes de que transcurra el tiempo definido, el arreglo se emite inmediatamente.

let collectTimeStride = 4          // Duración de la ventana temporal (segundos)
let collectMaxCount   = 2          // Número máximo de eventos por ventana

let collectedPublisher = sourcePublisher
    .collect(
        .byTimeOrCount(
            DispatchQueue.main,
            .seconds(collectTimeStride),
            collectMaxCount
        )
    )

En este ejemplo se emitirá un arreglo con los eventos coleccionados cuando se alcance el umbral de tiempo de 4 s o el umbral de cantidad de 2 eventos, lo que ocurra primero.

Descartando eventos

debounce(for:scheduler:options:)

debounce(for:scheduler:options:) espera el tiempo definido por el parámetro dueTime (un SchedulerTimeType.Stride) después del último elemento emitido por el publisher de entrada y, a continuación, vuelve a emitir ese último valor.

Importante: si el publisher de entrada envía un evento de finalización antes de que expire el tiempo de espera del debounce, el operador no podrá volver a emitir ese evento.

import Combine

let subject = PassthroughSubject<Int, Never>()
var subscriptions = Set<AnyCancellable>()

// Debounce con ventana de 1 s, re‑emite en DispatchQueue.main
let debounced = subject
    .debounce(for: .seconds(1.0), scheduler: DispatchQueue.main)
    .share()                     // Un único punto de suscripción

// Suscripción para imprimir los eventos
debounced
    .sink { value in
        print("Debounced value: \(value)")
    }
    .store(in: &subscriptions)

En el ejemplo anterior se usa share() para que varios suscriptores reciban los mismos resultados del debounce sin crear múltiples instancias del operador.

throttle(for:scheduler:latest:)

throttle(for:scheduler:latest:) también reduce la cantidad de eventos producidos por el flujo de entrada, pero su comportamiento difiere de debounce:

import Combine

let subject = PassthroughSubject<Int, Never>()
var subscriptions = Set<AnyCancellable>()

// Throttle con ventana de 1 s, emite el último evento de cada ventana
let throttled = subject
    .throttle(for: .seconds(1.0), scheduler: DispatchQueue.main, latest: true)

throttled
    .sink { value in
        print("Throttled value: \(value)")
    }
    .store(in: &subscriptions)

• Con latest: false se obtendría el primer valor de cada intervalo de 1 s.
• Con latest: true se obtendría el último valor recibido dentro de ese mismo intervalo.

Throttle

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)

Timeout

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 es nil, 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<String, Never>()

// 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) {
        TimeoutError.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 TimeInterval del Scheduler pasado por parámetro:

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

En el ejemplo anterior se tiene un Timer que emite un evento cada segundo.
Recuerda llamar a autoconnect() para iniciar el Timer en la primera suscripción.
Los valores emitidos están en 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. delay(for:scheduler:)
   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. autoconnect() en Timer.TimerPublisher
   El Publisher de Timer es connectable, lo que significa que se debe invocar explícitamente .connect() para que empiece a emitir eventos. autoconnect() hace que la conexión se establezca automáticamente cuando recibe la primera suscripción.

3. Diferencia entre .byTime y .byTimeOrCount en collect

   • .byTime solo agrupa las entradas y emite el arreglo acumulado al cumplirse el periodo especificado.
   • .byTimeOrCount acumula tanto por tiempo como por cantidad; emite el arreglo cuando se cumple el periodo o cuando se alcanza la cantidad especificada, lo que ocurra primero.

4. Diferencia entre debounce y throttle

   • debounce emite el último valor recibido después de que transcurra el tiempo de “rebote”.
   • throttle emite periódicamente y puede emitir el primer o el último valor recibido dentro de cada ventana de tiempo, según el parámetro latest.

5. Valor emitido por measureInterval(using:)
   Emite un valor de tipo Stride (específicamente SchedulerTimeType.Stride) que representa la distancia entre dos valores. La unidad de medida depende del Scheduler usado (nanosegundos para DispatchQueue, segundos para RunLoop, etc.).

6. Si un Publisher con debounce(for:) finaliza antes de que se cumpla el tiempo de espera
   No se emite nada. El último valor pendiente se descarta al completarse el upstream.

7. throttle(for:scheduler:latest:) con latest = false
   Se emite el primer valor recibido en cada ventana de tiempo.

8. timeout(_:scheduler:options:customError:)
   Emite un error (o finaliza exitosamente si customError es nil) cuando el Publisher no emite nada durante el intervalo especificado.

9. Unidad de medida de measureInterval(using:) con RunLoop.main
   Los valores están medidos en segundos.

10. Emisión anticipada con collect(.byTimeOrCount)
    Ocurre porque el Publisher de entrada envió un evento de fin, forzando la emisión del arreglo acumulado antes de que la ventana de tiempo haya expirado.

Preguntas y respuestas

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 (Note: original answer indicated event‑of‑completion; kept as provided)