Combinar el marco serializar operaciones asincrónicas

¿Cómo consigo que las canalizaciones asíncronas que constituyen el marco de trabajo de Combine se alineen sincrónicamente (en serie)?

Supongamos que tengo 50 URL de las que quiero descargar los recursos correspondientes, y digamos que quiero hacerlo de uno en uno. Sé cómo hacerlo con Operation / OperationQueue, por ejemplo, utilizando una subclase de Operation que no se declara terminada hasta que se completa la descarga. ¿Cómo haría lo mismo con Combine?

Por el momento, todo lo que se me ocurre es mantener una lista global de las URL restantes y abrir una, configurar esa tubería para una descarga, hacer la descarga y, en sinkla tubería, repetir. Eso no parece muy combinado.

Intenté hacer una matriz de las URL y asignarla a una variedad de editores. Sé que puedo "producir" un editor y hacer que se publique en el futuro utilizando flatMap. Pero aún sigo haciendo todas las descargas simultáneamente. No hay ninguna forma combinada de recorrer la matriz de manera controlada, ¿o sí?

(También imaginé hacer algo con Future, pero me confundí irremediablemente. No estoy acostumbrado a esta forma de pensar).

Solo probé brevemente esto, pero a primera vista parece que cada solicitud espera a que finalice la solicitud anterior antes de comenzar.

Estoy publicando esta solución en busca de comentarios. Sea crítico si esta no es una buena solución.

extension Collection where Element: Publisher {

    func serialize() -> AnyPublisher<Element.Output, Element.Failure>? {
        // If the collection is empty, we can't just create an arbititary publisher
        // so we return nil to indicate that we had nothing to serialize.
        if isEmpty { return nil }

        // We know at this point that it's safe to grab the first publisher.
        let first = self.first!

        // If there was only a single publisher then we can just return it.
        if count == 1 { return first.eraseToAnyPublisher() }

        // We're going to build up the output starting with the first publisher.
        var output = first.eraseToAnyPublisher()

        // We iterate over the rest of the publishers (skipping over the first.)
        for publisher in self.dropFirst() {
            // We build up the output by appending the next publisher.
            output = output.append(publisher).eraseToAnyPublisher()
        }

        return output
    }
}

Una versión más concisa de esta solución (proporcionada por @matt):

extension Collection where Element: Publisher {
    func serialize() -> AnyPublisher<Element.Output, Element.Failure>? {
        guard let start = self.first else { return nil }
        return self.dropFirst().reduce(start.eraseToAnyPublisher()) {
            $0.append($1).eraseToAnyPublisher()
        }
    }
}

Puede crear un Suscriptor personalizado donde reciba Subscribers.Demand.max (1). En ese caso, el suscriptor solicitará el siguiente valor solo cuando reciba uno. El ejemplo es para Int.publisher, pero un retraso aleatorio en el mapa imita el tráfico de red :-)

import PlaygroundSupport
import SwiftUI
import Combine

class MySubscriber: Subscriber {
  typealias Input = String
  typealias Failure = Never

  func receive(subscription: Subscription) {
    print("Received subscription", Thread.current.isMainThread)
    subscription.request(.max(1))
  }

  func receive(_ input: Input) -> Subscribers.Demand {
    print("Received input: \(input)", Thread.current.isMainThread)
    return .max(1)
  }

  func receive(completion: Subscribers.Completion<Never>) {
    DispatchQueue.main.async {
        print("Received completion: \(completion)", Thread.current.isMainThread)
        PlaygroundPage.current.finishExecution()
    }
  }
}

(110...120)
    .publisher.receive(on: DispatchQueue.global())
    .map {
        print(Thread.current.isMainThread, Thread.current)
        usleep(UInt32.random(in: 10000 ... 1000000))
        return String(format: "%02x", $0)
    }
    .subscribe(on: DispatchQueue.main)
    .subscribe(MySubscriber())

print("Hello")

PlaygroundPage.current.needsIndefiniteExecution = true

Impresión del patio ...

Hello
Received subscription true
false <NSThread: 0x600000064780>{number = 5, name = (null)}
Received input: 6e false
false <NSThread: 0x60000007cc80>{number = 9, name = (null)}
Received input: 6f false
false <NSThread: 0x60000007cc80>{number = 9, name = (null)}
Received input: 70 false
false <NSThread: 0x60000007cc80>{number = 9, name = (null)}
Received input: 71 false
false <NSThread: 0x60000007cc80>{number = 9, name = (null)}
Received input: 72 false
false <NSThread: 0x600000064780>{number = 5, name = (null)}
Received input: 73 false
false <NSThread: 0x600000064780>{number = 5, name = (null)}
Received input: 74 false
false <NSThread: 0x60000004dc80>{number = 8, name = (null)}
Received input: 75 false
false <NSThread: 0x60000004dc80>{number = 8, name = (null)}
Received input: 76 false
false <NSThread: 0x60000004dc80>{number = 8, name = (null)}
Received input: 77 false
false <NSThread: 0x600000053400>{number = 3, name = (null)}
Received input: 78 false
Received completion: finished true

ACTUALIZACIÓN finalmente encontré .flatMap(maxPublishers: ), lo que me obliga a actualizar este interesante tema con un enfoque un poco diferente. Por favor, vea que estoy usando la cola global para la programación, no solo un retraso aleatorio, solo para asegurarme de que recibir una transmisión serializada no sea un comportamiento "aleatorio" o "afortunado" :-)

import PlaygroundSupport
import Combine
import Foundation

PlaygroundPage.current.needsIndefiniteExecution = true

let A = (1 ... 9)
    .publisher
    .flatMap(maxPublishers: .max(1)) { value in
        [value].publisher
            .flatMap { value in
                Just(value)
                    .delay(for: .milliseconds(Int.random(in: 0 ... 100)), scheduler: DispatchQueue.global())
        }
}
.sink { value in
    print(value, "A")
}

let B = (1 ... 9)
    .publisher
    .flatMap { value in
        [value].publisher
            .flatMap { value in
                Just(value)
                    .delay(for: .milliseconds(Int.random(in: 0 ... 100)), scheduler: RunLoop.main)
        }
}
.sink { value in
    print("     ",value, "B")
}

huellas dactilares

1 A
      4 B
      5 B
      7 B
      1 B
      2 B
      8 B
      6 B
2 A
      3 B
      9 B
3 A
4 A
5 A
6 A
7 A
8 A
9 A

Basado en escrito aquí

.publicar por fascículos()?

definido por Clay Ellis la respuesta aceptada podría ser reemplazada por

.publisher.flatMap (maxPublishers: .max (1)) {$ 0}

mientras que la versión "no serializada" debe usar

.publisher.flatMap {$ 0}

"ejemplo del mundo real"

import PlaygroundSupport
import Foundation
import Combine

let path = "postman-echo.com/get"
let urls: [URL] = "... which proves the downloads are happening serially .-)".map(String.init).compactMap { (parameter) in
    var components = URLComponents()
    components.scheme = "https"
    components.path = path
    components.queryItems = [URLQueryItem(name: parameter, value: nil)]
    return components.url
}
//["https://postman-echo.com/get?]
struct Postman: Decodable {
    var args: [String: String]
}


let collection = urls.compactMap { value in
        URLSession.shared.dataTaskPublisher(for: value)
        .tryMap { data, response -> Data in
            return data
        }
        .decode(type: Postman.self, decoder: JSONDecoder())
        .catch {_ in
            Just(Postman(args: [:]))
    }
}

extension Collection where Element: Publisher {
    func serialize() -> AnyPublisher<Element.Output, Element.Failure>? {
        guard let start = self.first else { return nil }
        return self.dropFirst().reduce(start.eraseToAnyPublisher()) {
            return $0.append($1).eraseToAnyPublisher()
        }
    }
}

var streamA = ""
let A = collection
    .publisher.flatMap{$0}

    .sink(receiveCompletion: { (c) in
        print(streamA, "     ", c, "    .publisher.flatMap{$0}")
    }, receiveValue: { (postman) in
        print(postman.args.keys.joined(), terminator: "", to: &streamA)
    })


var streamC = ""
let C = collection
    .serialize()?

    .sink(receiveCompletion: { (c) in
        print(streamC, "     ", c, "    .serialize()?")
    }, receiveValue: { (postman) in
        print(postman.args.keys.joined(), terminator: "", to: &streamC)
    })

var streamD = ""
let D = collection
    .publisher.flatMap(maxPublishers: .max(1)){$0}

    .sink(receiveCompletion: { (c) in
        print(streamD, "     ", c, "    .publisher.flatMap(maxPublishers: .max(1)){$0}")
    }, receiveValue: { (postman) in
        print(postman.args.keys.joined(), terminator: "", to: &streamD)
    })

PlaygroundPage.current.needsIndefiniteExecution = true

huellas dactilares

.w.h i.c hporves ht edownloadsa erh appeninsg eriall y.-)       finished     .publisher.flatMap{$0}
... which proves the downloads are happening serially .-)       finished     .publisher.flatMap(maxPublishers: .max(1)){$0}
... which proves the downloads are happening serially .-)       finished     .serialize()?

Me parece muy útil en otros escenarios también. Intente utilizar el valor predeterminado de maxPublishers en el siguiente fragmento y compare los resultados :-)

import Combine

let sequencePublisher = Publishers.Sequence<Range<Int>, Never>(sequence: 0..<Int.max)
let subject = PassthroughSubject<String, Never>()

let handle = subject
    .zip(sequencePublisher.print())
    //.publish
    .flatMap(maxPublishers: .max(1), { (pair)  in
        Just(pair)
    })
    .print()
    .sink { letters, digits in
        print(letters, digits)
    }

"Hello World!".map(String.init).forEach { (s) in
    subject.send(s)
}
subject.send(completion: .finished)

En todos los demás marcos reactivos esto es realmente fácil; solo se usa concatpara concatenar y aplanar los resultados en un solo paso y luego se pueden obtener reducelos resultados en una matriz final. Apple hace esto difícil porque Publisher.Concatenateno tiene sobrecarga que acepte una variedad de editores. Hay rarezas similares con Publisher.Merge. Tengo la sensación de que esto tiene que ver con el hecho de que devuelven editores genéricos anidados en lugar de devolver un solo tipo genérico como rx Observable. Supongo que puedes llamar a Concatenateen un bucle y luego reducir los resultados concatenados en una sola matriz, pero realmente espero que aborden este problema en la próxima versión. Ciertamente, es necesario concatenar a más de 2 publicadores y fusionar más de 4 publicadores (y las sobrecargas para estos dos operadores ni siquiera son consistentes, lo cual es extraño).

EDITAR:

Volví a esto y descubrí que de hecho puedes concatenar una variedad arbitraria de editores y emitirán en secuencia. No tengo idea de por qué no hay una función como ConcatenateManyhacer esto por usted, pero parece que siempre y cuando esté dispuesto a usar un editor borrado no es tan difícil escribir uno. Este ejemplo muestra que la fusión emite en orden temporal mientras que concat emite en el orden de combinación:

import PlaygroundSupport
import SwiftUI
import Combine

let p = Just<Int>(1).append(2).append(3).delay(for: .seconds(0.25), scheduler: RunLoop.main).eraseToAnyPublisher()
let q = Just<Int>(4).append(5).append(6).eraseToAnyPublisher()
let r = Just<Int>(7).append(8).append(9).delay(for: .seconds(0.5), scheduler: RunLoop.main).eraseToAnyPublisher()
let concatenated: AnyPublisher<Int, Never> = [q,r].reduce(p) { total, next in
  total.append(next).eraseToAnyPublisher()
}

var subscriptions = Set<AnyCancellable>()

concatenated
  .sink(receiveValue: { v in
    print("concatenated: \(v)")
  }).store(in: &subscriptions)

Publishers
  .MergeMany([p,q,r])
  .sink(receiveValue: { v in
    print("merge: \(v)")
  }).store(in: &subscriptions)

De la pregunta original:

Intenté hacer una matriz de las URL y asignarla a una variedad de editores. Sé que puedo "producir" un editor y hacer que se publique en el futuro utilizando flatMap. Pero aún sigo haciendo todas las descargas simultáneamente. No hay ninguna forma combinada de recorrer la matriz de manera controlada, ¿o sí?

Aquí hay un ejemplo de juguete para reemplazar el problema real:

let collection = (1 ... 10).map {
    Just($0).delay(
        for: .seconds(Double.random(in:1...5)),
        scheduler: DispatchQueue.main)
        .eraseToAnyPublisher()
}
collection.publisher
    .flatMap() {$0}
    .sink {print($0)}.store(in:&self.storage)

Esto emite los enteros del 1 al 10 en orden aleatorio llegando a tiempos aleatorios. El objetivo es hacer algo con collectioneso hará que emita los enteros del 1 al 10 en orden.

Ahora vamos a cambiar solo una cosa: en la línea

.flatMap {$0}

agregamos el maxPublishersparámetro:

let collection = (1 ... 10).map {
    Just($0).delay(
        for: .seconds(Double.random(in:1...5)),
        scheduler: DispatchQueue.main)
        .eraseToAnyPublisher()
}
collection.publisher
    .flatMap(maxPublishers:.max(1)) {$0}
    .sink {print($0)}.store(in:&self.storage)

Presto, ahora hacer emiten los números enteros de 1 a 10, en orden, con intervalos aleatorios entre ellos.

Apliquemos esto al problema original. Para demostrarlo, necesito una conexión a Internet bastante lenta y un recurso bastante grande para descargar. Primero, lo haré con ordinario .flatMap:

let eph = URLSessionConfiguration.ephemeral
let session = URLSession(configuration: eph)
let url = "https://photojournal.jpl.nasa.gov/tiff/PIA23172.tif"
let collection = [url, url, url]
    .map {URL(string:$0)!}
    .map {session.dataTaskPublisher(for: $0)
        .eraseToAnyPublisher()
}
collection.publisher.setFailureType(to: URLError.self)
    .handleEvents(receiveOutput: {_ in print("start")})
    .flatMap() {$0}
    .map {$0.data}
    .sink(receiveCompletion: {comp in
        switch comp {
        case .failure(let err): print("error", err)
        case .finished: print("finished")
        }
    }, receiveValue: {_ in print("done")})
    .store(in:&self.storage)

El resultado es

start
start
start
done
done
done
finished

lo que muestra que estamos haciendo las tres descargas simultáneamente. Ok, ahora cambia

    .flatMap() {$0}

a

    .flatMap(maxPublishers:.max(1) {$0}

El resultado ahora es:

start
done
start
done
start
done
finished

Así que ahora estamos descargando en serie, que es el problema originalmente resuelto.

adjuntar

De acuerdo con el principio de TIMTOWTDI, podemos encadenar a los editores appendpara serializarlos:

let collection = (1 ... 10).map {
    Just($0).delay(
        for: .seconds(Double.random(in:1...5)),
        scheduler: DispatchQueue.main)
        .eraseToAnyPublisher()
}
let pub = collection.dropFirst().reduce(collection.first!) {
    return $0.append($1).eraseToAnyPublisher()
}

El resultado es una editorial que serializa a las editoriales retrasadas en la colección original. Probémoslo suscribiéndolo:

pub.sink {print($0)}.store(in:&self.storage)

Efectivamente, los enteros ahora llegan en orden (con intervalos aleatorios entre).

Podemos encapsular la creación de pubuna colección de editores con una extensión en Colección, como lo sugiere Clay Ellis:

extension Collection where Element: Publisher {
    func serialize() -> AnyPublisher<Element.Output, Element.Failure>? {
        guard let start = self.first else { return nil }
        return self.dropFirst().reduce(start.eraseToAnyPublisher()) {
            return $0.append($1).eraseToAnyPublisher()
        }
    }
}

Aquí hay una página de código de juegos que muestra un posible enfoque. La idea principal es transformar las llamadas API asíncronas en una cadena de Futureeditores, haciendo así una canalización en serie.

Entrada: rango de int de 1 a 10 que asíncronamente en la cola de fondo convertida en cadenas

Demostración de llamada directa a API asíncrona:

let group = DispatchGroup()
inputValues.map {
    group.enter()
    asyncCall(input: $0) { (output, _) in
        print(">> \(output), in \(Thread.current)")
        group.leave()
    }
}
group.wait()

Salida:

>> 1, in <NSThread: 0x7fe76264fff0>{number = 4, name = (null)}
>> 3, in <NSThread: 0x7fe762446b90>{number = 3, name = (null)}
>> 5, in <NSThread: 0x7fe7624461f0>{number = 5, name = (null)}
>> 6, in <NSThread: 0x7fe762461ce0>{number = 6, name = (null)}
>> 10, in <NSThread: 0x7fe76246a7b0>{number = 7, name = (null)}
>> 4, in <NSThread: 0x7fe764c37d30>{number = 8, name = (null)}
>> 7, in <NSThread: 0x7fe764c37cb0>{number = 9, name = (null)}
>> 8, in <NSThread: 0x7fe76246b540>{number = 10, name = (null)}
>> 9, in <NSThread: 0x7fe7625164b0>{number = 11, name = (null)}
>> 2, in <NSThread: 0x7fe764c37f50>{number = 12, name = (null)}

Demostración de la tubería combinada:

Salida:

>> got 1
>> got 2
>> got 3
>> got 4
>> got 5
>> got 6
>> got 7
>> got 8
>> got 9
>> got 10
>>>> finished with true

Código:

import Cocoa
import Combine
import PlaygroundSupport

// Assuming there is some Asynchronous API with
// (eg. process Int input value during some time and generates String result)
func asyncCall(input: Int, completion: @escaping (String, Error?) -> Void) {
    DispatchQueue.global(qos: .background).async {
            sleep(.random(in: 1...5)) // wait for random Async API output
            completion("\(input)", nil)
        }
}

// There are some input values to be processed serially
let inputValues = Array(1...10)

// Prepare one pipeline item based on Future, which trasform Async -> Sync
func makeFuture(input: Int) -> AnyPublisher<Bool, Error> {
    Future<String, Error> { promise in
        asyncCall(input: input) { (value, error) in
            if let error = error {
                promise(.failure(error))
            } else {
                promise(.success(value))
            }
        }
    }
    .receive(on: DispatchQueue.main)
    .map {
        print(">> got \($0)") // << sideeffect of pipeline item
        return true
    }
    .eraseToAnyPublisher()
}

// Create pipeline trasnforming input values into chain of Future publishers
var subscribers = Set<AnyCancellable>()
let pipeline =
    inputValues
    .reduce(nil as AnyPublisher<Bool, Error>?) { (chain, value) in
        if let chain = chain {
            return chain.flatMap { _ in
                makeFuture(input: value)
            }.eraseToAnyPublisher()
        } else {
            return makeFuture(input: value)
        }
    }

// Execute pipeline
pipeline?
    .sink(receiveCompletion: { _ in
        // << do something on completion if needed
    }) { output in
        print(">>>> finished with \(output)")
    }
    .store(in: &subscribers)

PlaygroundPage.current.needsIndefiniteExecution = true

Usar flatMap(maxPublishers:transform:)con .max(1), p. Ej.

func imagesPublisher(for urls: [URL]) -> AnyPublisher<UIImage, URLError> {
    Publishers.Sequence(sequence: urls.map { self.imagePublisher(for: $0) })
        .flatMap(maxPublishers: .max(1)) { $0 }
        .eraseToAnyPublisher()
}

Dónde

func imagePublisher(for url: URL) -> AnyPublisher<UIImage, URLError> {
    URLSession.shared.dataTaskPublisher(for: url)
        .compactMap { UIImage(data: $0.data) }
        .receive(on: RunLoop.main)
        .eraseToAnyPublisher()
}

y

var imageRequests: AnyCancellable?

func fetchImages() {
    imageRequests = imagesPublisher(for: urls).sink(receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("done")
        case .failure(let error):
            print("failed", error)
        }
    }, receiveValue: { image in
        // do whatever you want with the images as they come in
    })
}

Eso resultó en:

Pero debemos reconocer que recibes un gran éxito de rendimiento haciéndolos secuencialmente, así. Por ejemplo, si lo aumento hasta 6 a la vez, es más del doble de rápido:

Personalmente, recomendaría descargar solo secuencialmente si es absolutamente necesario (lo cual, al descargar una serie de imágenes / archivos, casi seguramente no es el caso). Sí, realizar solicitudes simultáneamente puede hacer que no terminen en un orden particular, pero solo usamos una estructura que es independiente del orden (por ejemplo, un diccionario en lugar de una matriz simple), pero las ganancias de rendimiento son tan significativas que generalmente vale la pena.

Pero, si desea que se descarguen secuencialmente, el maxPublishersparámetro puede lograrlo.