Combine #5: Operadores para combinar Streams

Prepending (Anexando al principio)

prepend(Output...) recibe una lista variádica de valores y los emite antes del flujo de datos emitido por el Publisher de entrada ("upstream"). El orden es importante. En el siguiente ejemplo, 3 y 4 se imprimen de último. 1 y 2 se imprimen antes que ellos. -1 y 0 se imprimen de primero.

let publisher = [3, 4].publisher
publisher
    .prepend(1, 2)
    .prepend(-1, 0)
    .sink { ... }
// -1 0 1 2 3 4

prepend(Sequence) recibe un objeto secuencia, como Array, Strideable o Set, y emite sus valores antes del flujo de datos emitido por el Publisher de entrada ("upstream").

• Tener en cuenta que, para el caso de Set, no se puede garantizar el orden de la emisión de sus valores.

let publisher = [5, 6].publisher // Ultimo
publisher
    .prepend([3, 4]) // Penúltimo
    .prepend(stride(from: 6, to: 9, by: 2)) // Antepenúltimo
    .prepend(Set(1...2)) // Primero
    .sink { ... }
// 1 2 6 8 3 4 5 6

prepend(Publisher) permite emitir los valores de un segundo Publisher, p2 , ANTES del Publisher original, p1. p1 va a empezar a trabajar y emitir eventos solo después de que p2 emita el evento de fin .finished. Si no hay evento de fin, p1 nunca va a emitir sus valores.

let p1 = [3, 4].publisher
let p2 = [1, 2].publisher
p1
  .prepend(p2)
  .sink { ... }
// 1 2 3 4

Appending (Anexando al final)

append(Output...) recibe una lista variádica de valores y los emite después del flujo de datos emitido por el Publisher de entrada ("upstream"), que debe terminar con un .finished. Aunque parezca trivial, vale aclarar: Si no se recibe evento .finished, Combine no tiene forma de saber que debe agregar los valores del append(Output...).

let publisher = [1].publisher
publisher
  .append(2, 3)
  .append(4)
  .sink { ... }
// 1 2 3 4

append(Sequence) emite los elementos de una secuencia, como Array, Strideable o Set, después del flujo de datos emitido por el Publisher de entrada ("upstream"). Se necesita que el Publisher de entrada emita un evento de fin, .finished, antes de anexar los demás elementos al flujo.

let publisher = [1, 2, 3].publisher
publisher
  .append([4, 5])
  .append(Set([6, 7]))
  .append(stride(from: 8, to: 11, by: 2))
  .sink { ... }
// 1 2 3 4 5 7 6 8 10

append(Publisher) permite emitir los valores de un segundo Publisher, p2 , DESPUÉS del Publisher original, p1. p2 va a empezar a trabajar y emitir eventos solo después de que p1 emita el evento de fin .finished. Si no hay evento de fin, p2 nunca va a emitir sus valores.

let p1 = [1, 2].publisher
let p2 = [3, 4].publisher
p1
  .append(p2)
  .sink { ... }
// 1 2 3 4

Combinando Publishers

switchToLatest() se aplica a un emisor de emisores (e.g. Publisher<Publisher<X, Never>, Never>), retornando una suscripción de tipo Publisher<X, Never>, como si aplanara a los emisores. Para lograr esto, switchToLatest() re-emite los valores emitidos por el último emisor emitido. Al mismo tiempo, cancela a todos los emisores emitidos antes.
Un caso de estudio donde podríamos usar este operador es un botón que dispara una petición web. Si presiona dos veces el botón, iniciaría una segunda petición. ¿Cómo se cancela la primera? - Para esto existe switchToLatest()

let publisher1 = PassthroughSubject<Int, Never>()
let publisher2 = PassthroughSubject<Int, Never>()
// Se crea el emisor de emisores
let publishers = PassthroughSubject<PassthroughSubject<Int, Never>, Never>()
publishers
  .switchToLatest()
  // Suscripción a valores de tipo Int
  .sink(receiveValue: { intValue in print(Value) })
// Se emite el primer emisor
publishers.send(publisher1)
publisher1.send(1)
publisher1.send(2)
// Se emite el segundo emisor, lo cual cancela al primero
publishers.send(publisher2)
publisher1.send(3)
publisher1.send(4)
publisher2.send(5)
// Se finalizan el último emisor, y el emisor de emisores
publisher2.send(completion: .finished)
publishers.send(completion: .finished)
// SALIDA
// 1 2 5

merge(with:) permite combinar las salidas de hasta ocho Publishers que emitan el mismo tipo de dato. El Publisher resultante terminará cuando el último Publisher termine.

let publisher1 = PassthroughSubject<Int, Never>()
let publisher2 = PassthroughSubject<Int, Never>()
publisher1
    .merge(with: publisher2)
    .sink( ... )
publisher1.send(1) // 1
publisher2.send(2) // 2
publisher1.send(completion: .finished)
publisher2.send(3) // 3
publisher2.send(completion: .finished) // Completed

combineLatest() junta la salida de hasta cuatro Publishers, que pueden emitir datos de diferentes tipos. Este operador emite una tupla con el último valor emitido por todos los Publishers combinados, cada vez que cualquiera de ellos emite un valor. Para ello se requiere que todos hayan emitido al menos un valor. Por otro lado, existe una versión de este operador que recibe un closure para transformar los valores emitidos por los Publishers en un nuevo valor de un tipo específico.

let p1 = PassthroughSubject<Int, Never>()
let p2 = PassthroughSubject<String, Never>()
p1
  .combineLatest(publisher2)
  .sink( ... )
p1.send(1) // Se recibe, pero no hay emisión porque p2 no ha emitido nada
p1.send(2) // Se recibe, pero no hay emisión porque p2 no ha emitido nada
p2.send("a") // P1: 2, P2: a
p2.send("b") // P1: 2, P2: b
p1.send(3) // P1: 3, P2: b
p2.send("c") // P1: 3, P2: c
p1.send(completion: .finished)
p2.send("d") // P1: 3, P2: d
p2.send(completion: .finished) // Completed

zip() combina la salida de hasta cuatro Publishers, emitiendo una tupla con los valores emitidos por ellos en el mismo índice. Esto significa que espera a que todos los Publishers emitan el valor i-ésimo para poder producir la tupla i-ésima. Igual que para el operador combineLatest(), zip() también tiene una versión que recibe una función de transformación.

let p1 = PassthroughSubject<Int, Never>()
let p2 = PassthroughSubject<String, Never>()
p1
  .zip(p2)
  .sink( ... )
p1.send(1) // No emite porque falta p2
p1.send(2) // No emite porque falta p2
p2.send("a") // P1: 1, P2: a
p2.send("b") // P1: 2, P2: b
p1.send(3) // No emite porque falta p1
p2.send("c") // P1: 3, P2: c
p2.send("d") // No emite porque falta p1
p1.send(completion: .finished)
p2.send(completion: .finished) // Se ignora "d". Imprime Completed

---

Cuestionario

1. ¿Qué diferencia hay entre prepend(Output...) y prepend(Sequence)?

2. ¿Qué sucede si el Publisher pasado a prepend(Publisher) nunca envía un evento .finished?

3. ¿Por qué append(Output…) requiere que el Publisher de entrada envíe .finished antes de emitir los valores añadidos?

4. Explica brevemente la diferencia entre combineLatest() y zip().

5. ¿Qué hace switchToLatest() con los Publishers previamente emitidos cuando aparece uno nuevo?

6. Si se usa prepend(Set(1...3)), ¿qué se debe tener en cuenta?

• [] Los valores se ordenan automáticamente
• [] No se garantiza el orden de emisión
• [] Los valores se duplican
• [] Los valores se invierten

7. append(Publisher) emitirá los valores del segundo Publisher...

• [] En paralelo con el primero
• [] Solo después de que el primero termine
• [] Antes de que el primero empiece
• [] Solo si el segundo tiene el mismo número de valores

8. merge(with:) termina su emisión cuando...

• [] El primer Publisher termina
• [] Todos los Publishers hayan terminado
• [] El último Publisher empieza
• [] El primer Publisher emite un .finished

9. En combineLatest(p1, p2), se emitirá una tupla solo cuando...

• [] Cualquiera emita un valor, sin importar los anteriores
• [] Ambos hayan emitido al menos un valor 
• [] El primero haya terminado
• [] Se emitan valores con el mismo índice

10. ¿Cuál de estos operadores espera emparejar valores por índice antes de emitir?

• [] merge(with:)
• [] zip()
• [] combineLatest()
• [] switchToLatest()

---

Solución

1. ¿Qué diferencia hay entre prepend(Output...) y prepend(Sequence)?

Tanto prepend(Output...) como prepend(Sequence) emiten los valores pasados por argumento antes del Publisher de entrada ("upstream"). Sin embargo, el primero recibe un argumento variádico (varios valores del mismo tipo separados por coma), mientras que el segundo recibe una colección que conforma Sequence.

2. ¿Qué sucede si el Publisher pasado a prepend(Publisher) nunca envía un evento .finished?

Nunca empieza a emitir los eventos del Publisher de entrada (upstream).

3. ¿Por qué append(Output...) requiere que el Publisher de entrada envíe .finished antes de emitir los valores añadidos?

Porque al recibir el .finished del Publisher de entrada empieza a emitir los valores pasados como argumento variádico.

4. Explica brevemente la diferencia entre combineLatest() y zip().

Ambos combinan los valores emitidos por hasta cuatro Publishers, y ambos tienen versiones que reciben una función de transformación. combineLatest() emite una tupla con el último valor emitido por cada uno de los Publishers, y requiere que todos hayan emitido al menos un valor para poder empezar. zip() empareja los valores i-ésimos emitidos por cada Publisher en la i-ésima tupla.

5. ¿Qué hace switchToLatest() con los Publishers previamente emitidos cuando aparece uno nuevo?

Los cancela.

6. Si se usa prepend(Set(1...3)), ¿qué se debe tener en cuenta?

• [✅] No se garantiza el orden de emisión

7. append(Publisher) emitirá los valores del segundo Publisher...

• [✅] Solo después de que el primero termine

8. merge(with:) termina su emisión cuando...

• [✅] Todos los Publishers hayan terminado

9. En combineLatest(p1, p2), se emitirá una tupla solo cuando...

• [✅] Ambos hayan emitido al menos un valor

10. ¿Cuál de estos operadores espera emparejar valores por índice antes de emitir?

• [✅] zip()