Swift Combine: How to create a single publisher from a list of publishers?

Question

Using Apple's new Combine framework I want to make multiple requests from each element in a list. Then I want a single result from a reduction of all the the responses. Basically I want to go from list of publishers to a single publisher that holds a list of responses.

I've tried making a list of publishers, but I don't know how to reduce that list into a single publisher. And I've tried making a publisher containing a list but I can't flat map a list of publishers.

Please look at the "createIngredients" function

func createIngredient(ingredient: Ingredient) -> AnyPublisher<CreateIngredientMutation.Data, Error> {
    return apollo.performPub(mutation: CreateIngredientMutation(name: ingredient.name, optionalProduct: ingredient.productId, quantity: ingredient.quantity, unit: ingredient.unit))
            .eraseToAnyPublisher()
}

func createIngredients(ingredients: [Ingredient]) -> AnyPublisher<[CreateIngredientMutation.Data], Error> {
    // first attempt
    let results = ingredients
            .map(createIngredient)
    // results = [AnyPublisher<CreateIngredientMutation.Data, Error>]

    // second attempt
    return Publishers.Just(ingredients)
            .eraseToAnyPublisher()
            .flatMap { (list: [Ingredient]) -> Publisher<[CreateIngredientMutation.Data], Error> in
                return list.map(createIngredient) // [AnyPublisher<CreateIngredientMutation.Data, Error>]
            }
}

I'm not sure how to take an array of publishers and convert that to a publisher containing an array.

Result value of type '[AnyPublisher]' does not conform to closure result type 'Publisher'

If I try and use eraseToAnyPublisher() with apollo like `apollo.fetch(query: AllProductsQuery())).eraseToAnyPublisher()` I get the error `Value of type 'Cancellable' has no member 'eraseToAnyPublisher'` - how are you doing it without seeing an error? — daihovey, Jun 17 '20 at 12:24
@daidai I used an extension to apollo to accomplish that. This question is really about merging multiple publishers. — Andre Carrera, Jun 19 '20 at 17:26

score 55 · Accepted Answer · edited Feb 13 '21 at 19:18

Essentially, in your specific situation you're looking at something like this:

func createIngredients(ingredients: [Ingredient]) -> AnyPublisher<[CreateIngredientMutation.Data], Error> {
    Publishers.MergeMany(ingredients.map(createIngredient(ingredient:)))
        .collect()
        .eraseToAnyPublisher()
}

This 'collects' all the elements produced by the upstream publishers and – once they have all completed – produces an array with all the results and finally completes itself.

Bear in mind, if one of the upstream publishers fails – or produces more than one result – the number of elements may not match the number of subscribers, so you may need additional operators to mitigate this depending on your situation.

The more generic answer, with a way you can test it using the EntwineTest framework:

import XCTest
import Combine
import EntwineTest

final class MyTests: XCTestCase {
    
    func testCreateArrayFromArrayOfPublishers() {

        typealias SimplePublisher = Just<Int>

        // we'll create our 'list of publishers' here. Each publisher emits a single
        // Int and then completes successfully – using the `Just` publisher.
        let publishers: [SimplePublisher] = [
            SimplePublisher(1),
            SimplePublisher(2),
            SimplePublisher(3),
        ]

        // we'll turn our array of publishers into a single merged publisher
        let publisherOfPublishers = Publishers.MergeMany(publishers)

        // Then we `collect` all the individual publisher elements results into
        // a single array
        let finalPublisher = publisherOfPublishers.collect()

        // Let's test what we expect to happen, will happen.
        // We'll create a scheduler to run our test on
        let testScheduler = TestScheduler()

        // Then we'll start a test. Our test will subscribe to our publisher
        // at a virtual time of 200, and cancel the subscription at 900
        let testableSubscriber = testScheduler.start { finalPublisher }

        // we're expecting that, immediately upon subscription, our results will
        // arrive. This is because we're using `just` type publishers which
        // dispatch their contents as soon as they're subscribed to
        XCTAssertEqual(testableSubscriber.recordedOutput, [
            (200, .subscription),            // we're expecting to subscribe at 200
            (200, .input([1, 2, 3])),        // then receive an array of results immediately
            (200, .completion(.finished)),   // the `collect` operator finishes immediately after completion
        ])
    }
}

Worth noting that this doesn't preserve the order of the underlying array. The elements of the final array will be sorted in the order each publisher completed. — rpowell, Oct 04 '19 at 04:50
I couldn't find a good solution for that. I'm currently sorting my values after they've all been fetched. — rpowell, Oct 16 '19 at 23:05
Any idea if the new publisher will receive the value when any of the publisher’s value gets changed? — Keyur Tailor, Dec 20 '22 at 19:00

user1687195 · Answer 2 · 2020-06-10T22:05:42.507

I think that Publishers.MergeMany could be of help here. In your example, you might use it like so:

func createIngredients(ingredients: [Ingredient]) -> AnyPublisher<CreateIngredientMutation.Data, Error> {
    let publishers = ingredients.map(createIngredient(ingredient:))
    return Publishers.MergeMany(publishers).eraseToAnyPublisher()
}

That will give you a publisher that sends you single values of the Output.

However, if you specifically want the Output in an array all at once at the end of all your publishers completing, you can use collect() with MergeMany:

func createIngredients(ingredients: [Ingredient]) -> AnyPublisher<[CreateIngredientMutation.Data], Error> {
    let publishers = ingredients.map(createIngredient(ingredient:))
    return Publishers.MergeMany(publishers).collect().eraseToAnyPublisher()
}

And either of the above examples you could simplify into a single line if you prefer, ie:

func createIngredients(ingredients: [Ingredient]) -> AnyPublisher<CreateIngredientMutation.Data, Error> {
    Publishers.MergeMany(ingredients.map(createIngredient(ingredient:))).eraseToAnyPublisher()
}

You could also define your own custom merge() extension method on Sequence and use that to simplify the code slightly:

extension Sequence where Element: Publisher {
    func merge() -> Publishers.MergeMany<Element> {
        Publishers.MergeMany(self)
    }
}

func createIngredients(ingredients: [Ingredient]) -> AnyPublisher<CreateIngredientMutation.Data, Error> {
    ingredients.map(createIngredient).merge().eraseToAnyPublisher()
}

> "That will give you a publisher that sends you single values of the Output." cool! finally I see what `.collect()` does. thanks! :) — Serj Rubens, May 20 '22 at 20:23

score 2 · Answer 3 · answered Jun 02 '21 at 09:32

To add on the answer by Tricky, here is a solution which retains the order of elements in the array. It passes an index for each element through the whole chain, and sorts the collected array by the index.

Complexity should be O(n log n) because of the sorting.

import Combine

extension Publishers {

    private struct EnumeratedElement<T> {
        let index: Int
        let element: T

        init(index: Int, element: T) {
            self.index = index
            self.element = element
        }

        init(_ enumeratedSequence: EnumeratedSequence<[T]>.Iterator.Element) {
            index = enumeratedSequence.offset
            element = enumeratedSequence.element
        }
    }

    static func mergeMappedRetainingOrder<InputType, OutputType>(
        _ inputArray: [InputType],
        mapTransform: (InputType) -> AnyPublisher<OutputType, Error>
    ) -> AnyPublisher<[OutputType], Error> {

        let enumeratedInputArray = inputArray.enumerated().map(EnumeratedElement.init)

        let enumeratedMapTransform: (EnumeratedElement<InputType>) -> AnyPublisher<EnumeratedElement<OutputType>, Error> = { enumeratedInput in
            mapTransform(enumeratedInput.element)
                .map { EnumeratedElement(index: enumeratedInput.index, element: $0)}
                .eraseToAnyPublisher()
        }

        let sortEnumeratedOutputArrayByIndex: ([EnumeratedElement<OutputType>]) -> [EnumeratedElement<OutputType>] = { enumeratedOutputArray in
            enumeratedOutputArray.sorted { $0.index < $1.index }
        }

        let transformToNonEnumeratedArray: ([EnumeratedElement<OutputType>]) -> [OutputType] = {
            $0.map { $0.element }
        }

        return Publishers.MergeMany(enumeratedInputArray.map(enumeratedMapTransform))
            .collect()
            .map(sortEnumeratedOutputArrayByIndex)
            .map(transformToNonEnumeratedArray)
            .eraseToAnyPublisher()
    }
}

Unit test for the solution:

import XCTest
import Combine

final class PublishersExtensionsTests: XCTestCase {

    // MARK: - Private properties

    private var cancellables = Set<AnyCancellable>()

    // MARK: - Tests

    func test_mergeMappedRetainingOrder() {
        let expectation = expectation(description: "mergeMappedRetainingOrder publisher")

        let numbers = (1...100).map { _ in Int.random(in: 1...3) }

        let mapTransform: (Int) -> AnyPublisher<Int, Error> = {
            let delayTimeInterval = RunLoop.SchedulerTimeType.Stride(Double($0))
            return Just($0)
                .delay(for: delayTimeInterval, scheduler: RunLoop.main)
                .setFailureType(to: Error.self)
                .eraseToAnyPublisher()
        }

        let resultNumbersPublisher = Publishers.mergeMappedRetainingOrder(numbers, mapTransform: mapTransform)

        resultNumbersPublisher.sink(receiveCompletion: { _ in }, receiveValue: { resultNumbers in
            XCTAssertTrue(numbers == resultNumbers)
            expectation.fulfill()
         }).store(in: &cancellables)

        waitForExpectations(timeout: 5)
    }
}

score 2 · Answer 4 · answered Dec 17 '21 at 11:29

2

You can do it in one line:

.flatMap(Publishers.Sequence.init(sequence:))

answered Dec 17 '21 at 11:29

Neph Muw

780
7
10

exactly what I was looking for, thankaya – Logan Feb 23 '22 at 00:06

Swift Combine: How to create a single publisher from a list of publishers?

4 Answers4

Linked