Combineを用いた非同期処理の直列・並列実行

はじめに

Playground環境で実行しています。

準備

非同期処理を実行する関数の定義

func fetchA() -> Future<String, Error> {
    return Future<String, Error> { promise in
        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
            promise(.success("Aを取得"))
        }
    }
}

func fetchB() -> Future<String, Error> {
    return Future<String, Error> { promise in
        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
            promise(.success("Bを取得"))
        }
    }
}

func fetchC() -> Future<String, Error> {
    return Future<String, Error> { promise in
        DispatchQueue.main.asyncAfter(deadline: .now() + 1) {
            promise(.success("Cを取得"))
        }
    }
}

この３つの関数を直列・並列で実行します。

直列実行

fetchA() 
    .flatMap(maxPublishers: .max(1)) { _ in fetchB() }  // fetchAの結果はクロージャの引数
    .flatMap(maxPublishers: .max(1)) { _ in fetchC() }
    .eraseToAnyPublisher()
    .sink { completion in
        switch completion {
        case .finished:
            print("finished")
        case .failure(_):
            print("failure")
        }
    } receiveValue: { value in
        print(value)
    }
    .store(in: &cancellable)

maxPublishersは同時に実行できる処理の数です。１を指定すると、同期的(順番どおり)に実行されるようです。
詳しくはCombineのflatMapで実行数を制御するを参照してください。

maxPublishersの性質を利用すると、こういった書き方もできます。

let publishers = [
    fetchA(),
    fetchB(),
    fetchC()
]

publishers
    .publisher
    .flatMap(maxPublishers: .max(1)) { $0 }
    .eraseToAnyPublisher()
    .sink { completion in
        switch completion {
        case .finished:
            print("finished")
        case .failure(_):
            print("failure")
        }
    } receiveValue: { value in
        print(value)
    }
    .store(in: &cancellable)

ただし、この書き方の場合は直前の処理の値を利用できないので、注意が必要です。

並列実行

let publishers = [
    fetchA(),
    fetchB(),
    fetchC()
]

publishers
    .publisher
    .flatMap(maxPublishers: .max(publishers.count)) { $0 }
    .eraseToAnyPublisher()
    .sink { completion in
        switch completion {
        case .finished:
            print("finished")
        case .failure(_):
            print("failure")
        }
    } receiveValue: { value in
        print(value)
    }
    .store(in: &cancellable)

今回は３つの処理を同時に実行したいのでmaxPublishersにpublishers.count(この例では3)を指定します。

全てが完了したときの処理はcase .finished内に書くといいと思います。
エラーが流れてきた場合はcase .failureが実行されて、.finishedは実行されません。

エラー処理

非同期処理の返り値をResult型に変換し、
sink内で成功/失敗に応じた分岐処理を書きます。

並列処理における例

publishers
    .publisher
    .flatMap(maxPublishers: .max(3)) { $0 }
    .map { Result.success($0) }
    .catch { Just(Result.failure($0)) }
    .sink(receiveValue: { result in
        switch result {
        case .success(let response):
            print(response)
        case .failure(let error):
            print(error.localizedDescription)
        }
    })
    .store(in: &c)

エラー処理は、下記の記事を参考にしました。
[Swift] Combine の flatMap で failure になると、以降は値が流れなくなる問題に対処する

終わりに

Combineで楽しい非同期処理ライフを！！