Mapping Swift Combine Future to Another Future

Mapping Swift Combine Future to another Future

You can't use dribs and drabs and bits and pieces from the Combine framework like that. You have to make a pipeline — a publisher, some operators, and a subscriber (which you store so that the pipeline will have a chance to run).

 Publisher
     |
     V
 Operator
     |
     V
 Operator
     |
     V
 Subscriber (and store it)

So, here, getItem is a function that produces your Publisher, a Future. So you can say

getItem (...)
    .map {...}
    ( maybe other operators )
    .sink {...} (or .assign(...))
    .store (...)

Now the future (and the whole pipeline) will run asynchronously and the result will pop out the end of the pipeline and you can do something with it.

Now, of course you can put the Future and the Map together and then stop, vending them so someone else can attach other operators and a subscriber to them. You have now assembled the start of a pipeline and no more. But then its type is not going to be Future; it will be an AnyPublisher<NSImage,Error>. And there's nothing wrong with that!

Using Combine operators to transform Future into Publisher

You are trying to map to another publisher. Most of the time, this is a sign that you need flatMap. If you use map instead, you'll get a publisher that publishes another publisher, which is almost certainly not what you want.

However, flatMap requires that the upstream publisher (the promise) has the same failure type as the publisher that you are mapping to. However, they aren't the same in this case, so you need to call mapError on the data session publisher to change its error type:

return Future<String?, Error> { promise in
    promise(.failure(NSError()))
}
// flatMap and notice the change in return type
.flatMap { idToken -> Publishers.MapError<URLSession.DataTaskPublisher, Error> in
    if idToken != nil {
        request.addValue("Bearer \(idToken!)", forHTTPHeaderField: "Authorization")
    }
    return URLSession.shared.dataTaskPublisher(for: request)
        // change the error type
        .mapError { $0 as Error } // "as Error" isn't technically needed. Just for clarity
}
.tryMap { result -> Response<T> in
    let value = try decoder.decode(T.self, from: result.data)
    return Response(value: value, response: result.response)
}
.receive(on: DispatchQueue.main)
.map(\.value)
.eraseToAnyPublisher()

Simple Future chaining that have different value types using Combine

So close! Just erase the type:

var users: AnyPublisher<[User], Error> {
    if didAlreadyImportUsers {
        return fetchUsers.eraseToAnyPublisher()
    } else {
        return importUsers.setFailureType(to: Error.self)
            .combineLatest(fetchUsers)
            .map { $0.1 }.eraseToAnyPublisher()
    }
}

You might say: Wait, that's not what I said to do; I want to return it as a Future. Well, don't want that! It isn't a Future; it's some horrible Future-plus-CombineLastest-plus-Map thingy. You don't want to know what it is. All you want to know is its output type and its failure type.

The whole point here is that one publisher is as good as another; they can be used interchangeably, provided their types match. That is why AnyPublisher is provided: it's so you can do exactly that. Basically, AnyPublisher is how you cast between publisher types: you just cast everything to AnyPublisher.

Swift Combine Future with multiple values?

Generally you can use a PassthroughSubject to publish custom outputs. You can wrap a PassthroughSubject (or multiple PassthroughSubjects) in your own implementation of Publisher to ensure that only your process can send events through the subject.

Let's mock a VideoFrame type and some input frames for example purposes:

typealias VideoFrame = String
let inputFrames: [VideoFrame] = ["a", "b", "c"]

Now we want to write a function that synchronously processes these frames. Our function should report progress somehow, and at the end, it should return the output frames. To report progress, our function will take a PassthroughSubject<Double, Never>, and send its progress (as a fraction from 0 to 1) to the subject:

func process(_ inputFrames: [VideoFrame], progress: PassthroughSubject<Double, Never>) -> [VideoFrame] {
    var outputFrames: [VideoFrame] = []
    for input in inputFrames {
        progress.send(Double(outputFrames.count) / Double(inputFrames.count))
        outputFrames.append("output for \(input)")
    }
    return outputFrames
}

Okay, so now we want to turn this into a publisher. The publisher needs to output both progress and a final result. So we'll use this enum as its output:

public enum ProgressEvent<Value> {
    case progress(Double)
    case done(Value)
}

Now we can define our Publisher type. Let's call it SyncPublisher, because when it receives a Subscriber, it immediately (synchronously) performs its entire computation.

public struct SyncPublisher<Value>: Publisher {
    public init(_ run: @escaping (PassthroughSubject<Double, Never>) throws -> Value) {
        self.run = run
    }

    public var run: (PassthroughSubject<Double, Never>) throws -> Value

    public typealias Output = ProgressEvent<Value>
    public typealias Failure = Error

    public func receive<Downstream: Subscriber>(subscriber: Downstream) where Downstream.Input == Output, Downstream.Failure == Failure {
        let progressSubject = PassthroughSubject<Double, Never>()
        let doneSubject = PassthroughSubject<ProgressEvent<Value>, Error>()
        progressSubject
            .setFailureType(to: Error.self)
            .map { ProgressEvent<Value>.progress($0) }
            .append(doneSubject)
            .subscribe(subscriber)
        do {
            let value = try run(progressSubject)
            progressSubject.send(completion: .finished)
            doneSubject.send(.done(value))
            doneSubject.send(completion: .finished)
        } catch {
            progressSubject.send(completion: .finished)
            doneSubject.send(completion: .failure(error))
        }
    }
}

Now we can turn our process(_:progress:) function into a SyncPublisher like this:

let inputFrames: [VideoFrame] = ["a", "b", "c"]
let pub = SyncPublisher<[VideoFrame]> { process(inputFrames, progress: $0) }

The run closure is { process(inputFrames, progress: $0) }. Remember that $0 here is a PassthroughSubject<Double, Never>, exactly what process(_:progress:) wants as its second argument.

When we subscribe to this pub, it will first create two subjects. One subject is the progress subject and gets passed to the closure. We'll use the other subject to publish either the final result and a .finished completion, or just a .failure completion if the run closure throws an error.

The reason we use two separate subjects is because it ensures that our publisher is well-behaved. If the run closure returns normally, the publisher publishes zero or more progress reports, followed by a single result, followed by .finished. If the run closure throws an error, the publisher publishes zero or more progress reports, followed by a .failed. There is no way for the run closure to make the publisher emit multiple results, or emit more progress reports after emitting the result.

At last, we can subscribe to pub to see if it works properly:

pub
    .sink(
        receiveCompletion: { print("completion: \($0)") },
        receiveValue: { print("output: \($0)") })

Here's the output:

output: progress(0.0)
output: progress(0.3333333333333333)
output: progress(0.6666666666666666)
output: done(["output for a", "output for b", "output for c"])
completion: finished

Swift Combine return result of first future after evaluation of second

You need to .map the Void result of the second publisher (voidFuture) back to the result of the first publisher (intFuture), which is what the .flatMap would emit:

func combinedFuture() -> AnyPublisher<Int, Error> {
    intFuture().flatMap { result in
        voidFuture().map { _ in result }
    }
    .eraseToAnyPublisher()
}

Combine - mapping errors to different types

You need catch to replace errors thrown by the upstream publisher with a new downstream Publisher.

Inside the catch, you can wrap the error in a Just, which is a Publisher that emits a single value immediately.

service.fetchPeople()
    .map { ViewState.loaded($0) }
    .catch { Just(ViewState.error($0.localizedDescription)) }
    .eraseToAnyPublisher()

Swift Combine: Do asynchronous work in a Future?

The premise of what you're trying to do is wrong. Future, though it can only return at most a single result, doesn't semantically stand for that. It's just a specific type of a publisher.

You should be returning an AnyPublisher at the function boundary, and not try to avoid it, which would make your code more robust to changes (e.g. what if you needed to wrap the Future in a Deferred? - a common practice)

process(value: Value) -> AnyPublisher<Value, Never> {
   ...
}

And if a subscriber can only handle a single result, they could simply ensure that with first():

process(value)
    .first()
    .sink {...}
    .store(in: &storage)

---

But if you insist, you could use a .sink inside Future's closure, if the closure captured the reference to the AnyCancellable and released it on completion:

process(value: Value) -> Future<Value, Never> {
    guard !childProcessors.isEmpty else {
        return Future { $0(.success(value)) }
    }
    let just = Just(value).eraseToAnyPublisher()
    let combined = childProcessors.reduce(just) { (publisher, processor) -> AnyPublisher<Value, Never> in
        publisher.flatMap { processor.process(value: $0).eraseToAnyPublisher() }
    }

    var c: AnyCancellable? = nil
    return Future { promise in
        c = combined.sink(receiveCompletion: { 
            withExtendedLifetime(c){}; c = nil 
        }) {
            promise(.success($0))
        }
    }
}

---

Related Topics