Combine #13: Manejo de Recursos

En ocasiones, se necesita compartir el resultado de un origen de datos determinado entre múltiples suscriptores, en lugar de duplicar los esfuerzos. Para ello, se usan los operadores share() y multicast(_:).

El operador share()

La mayoría de Publishers de Combine son estructuras (struct) que solo describen un pipeline, sin guardar un estado compartido. No se está creando una instancia viva que mantenga datos, sino un valor que describe cómo se producirá el flujo cuando alguien se suscriba. Por esta razón, cada suscripción inicia una nueva secuencia de publicación (i.e. arranca su propio trabajo nuevo).

share() devuelve un Publisher.Share, que se implementa sobre multicast(_:) y PassthroughSubject, creando una única suscripción al Publisher original, que se comparte entre múltiples suscriptores. Debido a que combine usa una clase internamente para mantener referencias a los suscriptores, se dice que hay un "estado compartido".

let shared = URLSession.shared
  .dataTaskPublisher(for: URL(string: "https://ejemplo.com")!)
  .map(\.data)
  .share()

let subscription1 = shared.sink(...)
let subscription2 = shared.sink(...)
// Solo se hace una vez la petición web, y el resultado se comparte 
// por las dos suscripciones

Cuando share() recibe la primera suscripción, inicia el trabajo del Publisher de entrada. Esto puede implicar que el flujo termine (emitiendo un evento de fin) antes de que ocurran otras suscripciones. En este caso, las nuevas suscripciones recibirán solo el evento de fin.

En el siguiente ejemplo, la petición termina antes de que ocurra la segunda suscripción, por lo que esta solo recibe un evento de fin.

let shared = URLSession.shared
  .dataTaskPublisher(for: URL(string: "https://ejemplo.com")!)
  .map(\.data)
  .share()

let subscription1 = shared.sink(...)
var subscription2: AnyCancellable? = nil
DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
  subscription2 = shared.sink(...)
}

El operador multicast(_:)

multicast(_:) usa un Subject para emitir valores a varios suscriptores y retorna un ConnectablePublisher, lo cual significa que se suscribirá al Publisher de entrada cuando reciba la señal connect().

// Subject para emitir eventos del multicast. 
// Recibe <Data, URLError> del flujo de entrada
let subject = PassthroughSubject<Data, URLError>()
// Se crea el Publisher multicast, envolviendo el subject
let multicasted = URLSession.shared
  .dataTaskPublisher(for: URL(string: "https://ejemplo.com")!)
  .map(\.data)
  .multicast(subject: subject)
let subscription1 = multicasted.sink(...)
var subscription2: AnyCancellable? = nil
// Crear una segunda suscripción al multicast
DispatchQueue.main.asyncAfter(deadline: .now() + 5) {
  subscription2 = multicasted.sink(...)
}
// Suscribir el multicast al dataTaskPublisher
DispatchQueue.main.asyncAfter(deadline: .now() + 6) {
  let cancellable = multicasted.connect()
}

Como multicast(_:) retorna un ConnectablePublisher, también se puede usar autoconnect() para suscribirlo al Publisher de entrada tan pronto reciba la primera suscripción.

En escenarios donde solo se requiere hacer un trabajo una vez y emitir un valor, se puede usar CurrentValueSubject.

Future

Future es un Publisher (implementado como class y no como struct) que recibe un closure con un Future.Promise que eventualmente produce un solo valor, y luego termina. Dentro del closure se ejecuta cierto trabajo y luego se llama a la promesa con el resultado o con un error.

Al crear la instancia de Future, el closure se ejecuta inmediatamente sin esperar a ningún suscriptor. Además, almacena el resultado de la promesa y lo emite a todos sus suscriptores, actuales y futuros.

let future = Future<Int, Error> { fulfill in
  do {
    let result = try performSomeWork()
    fulfill(.success(result))
  } catch {
    fulfill(.failure(error))
  }
}
let subscription1 = future.sink( ... )
let subscription2 = future.sink( ... )

---

Cuestionario

1. ¿Por qué, en la mayoría de los Publishers de Combine, cada suscripción repite el trabajo del flujo? ✅

• [ ] Porque los Publishers son clases que mantienen un estado interno.
• [ ] Porque Combine crea un nuevo hilo por suscripción.
• [ ] Porque los Publishers son estructuras (struct) que solo describen el pipeline y no comparten estado. 
• [ ] Porque los suscriptores deben conectarse manualmente con connect().

2. ¿Qué hace el operador share() en un Publisher? ✅

• [ ] Crea copias separadas del flujo para cada suscriptor.
• [ ] Cancela las suscripciones duplicadas.
• [ ] Crea una sola suscripción al Publisher original y comparte sus resultados entre varios suscriptores. 
• [ ] Convierte el Publisher en un Subject.

3. ¿Qué ocurre si una segunda suscripción a un Publisher compartido con share() se realiza después de que el flujo ya terminó? ✅

• [ ] Recibe los valores anteriores almacenados.
• [ ] No recibe nada, solo el evento de finalización. 
• [ ] Se vuelve a ejecutar todo el flujo original.
• [ ] Causa un error en tiempo de ejecución.

4. ¿En qué se basa internamente el operador share()? ✅

• [ ] En Future y Just.
• [ ] En multicast(_:) y PassthroughSubject. 
• [ ] En CurrentValueSubject.
• [ ] En DispatchQueue.

5. ¿Qué diferencia principal tiene multicast(_:) frente a share()? ✅

• [ ] multicast(_:) no necesita un Subject.
• [ ] multicast(_:) devuelve un ConnectablePublisherque requiere llamarse con connect(). 
• [ ] multicast(_:) repite el trabajo por cada suscriptor.
• [ ] share() permite múltiples conexiones manuales, mientras que multicast(_:) no.

6. ¿Qué ventaja ofrece el método autoconnect() en un ConnectablePublisher creado con multicast(_:)? ✅

• [ ] Permite que el Publisher se conecte automáticamente al recibir la primera suscripción. 
• [ ] Permite cancelar automáticamente las suscripciones.
• [ ] Convierte el Publisher en un Future.
• [ ] Guarda el resultado para suscriptores futuros.

7. ¿Qué característica distingue a Future de otros Publishers como share() o multicast(_:)? ✅

• [ ] Es un Publisher struct sin estado.
• [ ] Solo emite errores, nunca valores.
• [ ] Se ejecuta solo después de recibir una suscripción.
• [ ] Es una clase que ejecuta su trabajo inmediatamente al crearse y guarda el resultado para todos los suscriptores.

---

Solución

1. ¿Por qué, en la mayoría de los Publishers de Combine, cada suscripción repite el trabajo del flujo?

• [✅] Porque los Publishers son estructuras (struct) que solo describen el pipeline y no comparten estado.

2. ¿Qué hace el operador share() en un Publisher?

• [✅] Crea una sola suscripción al Publisher original y comparte sus resultados entre varios suscriptores.

3. ¿Qué ocurre si una segunda suscripción a un Publisher compartido con share() se realiza después de que el flujo ya terminó?

• [✅] No recibe nada, solo el evento de finalización.

4. ¿En qué se basa internamente el operador share()?

• [✅] En multicast(_:) y PassthroughSubject.

5. ¿Qué diferencia principal tiene multicast(_:) frente a share()?

• [✅] multicast(_:) devuelve un ConnectablePublisher que requiere llamarse con connect().

6. ¿Qué ventaja ofrece el método autoconnect() en un ConnectablePublisher creado con multicast(_:)?

• [✅] Permite que el Publisher se conecte automáticamente al recibir la primera suscripción.

7. ¿Qué característica distingue a Future de otros Publishers como share() o multicast(_:)?

• [✅] Es una clase que ejecuta su trabajo inmediatamente al crearse y guarda el resultado para todos los suscriptores.