Publisher#

再実行#

エラーが返ってきたときに、n 回再実行するというのが非常に簡単に書けます。

並列実行#

同時に n 個並列に実行するというのも簡単に書けます。

Future#

一つの値と完了か失敗を返せる。

Just#

一つの値を返して完了。

Deferred#

Future がインスタンスを作成した段階で実行されるのに対して、sinkされたタイミングでしか実行されない。

Fail#

失敗だけ返して完了。

Record#

複数値を返して完了か失敗を返せる。

Publisher#

Publisher はタスクの集合みたいなものなので、処理したい内容っぽいのを配列として用意します。

今回は適当に 1 から 10 までの数を入力とし、それを Publisher としましょう。

// Publisherの内容を保持するために必要
var task = Set<AnyCancellable>()

// タスクにわたす値の配列
let publisher = Array(Range(1 ... 10).map({ $0 })).publisher

ここでtaskが必要になる理由ですが、Publisherはキャンセル可能なタスクなのでこれを使って値を保持しておかないとインスタンスが消えた瞬間に強制的にタスク自体がキャンセルされるためです。

実装#

とりあえずよくわからなくてもいいので以下のコードを見ます。

publisher
    .receive(on: DispatchQueue.main)
    .map({ addValue(value: $0)})
    .sink(receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("FINISHED")
        case .failure:
            print("ERROR")
        }
    }, receiveValue: { value in
        print(value)
    })
    .store(in: &task)

実行結果#

出力結果は以下のようになります。

11
12
13
14
15
16
17
18
19
20
FINISHED

で、これだけでは何も面白くないので関数自体を Publisher 化します。

Publisher#

こちらはmapをflatMapに書き換えるだけです。

publisher
    .receive(on: DispatchQueue.main)
    // mapからflatMapに変更
    .flatMap({ addValue(value: $0) })
    .sink(receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("FINISHED")
        case .failure(let error):
            // エラーの内容を表示
            print("ERROR", error)
        }
    }, receiveValue: { value in
        // 受け取った値を表示
        print("RECEIVE", value)
    })
    .store(in: &task)

Publisher 関数#

// 通常の関数
func addValue(value: Int) -> Int {
    value + 10
}

// Publisher化した関数
// 返り値はAnyPublisher型にする
func addValue(value: Int) -> AnyPublisher<Int, Never> {
    // Deferredにすることで即時実行されない
    Deferred {
        Future { promise in
            // Returnの代わりにこう書く
            promise(.success(value + 10))
        }
    }
    // よくわからないが必須のおまじない
    .eraseToAnyPublisher()
}

エラーを返さない場合の書き方についてはこれを丸暗記すると良いそうです。

エラーを返す場合#

一定の確率でエラーを発生させるコードを考えます。

以下のコードは 10%の確率でエラーを発生させます。

// エラー型を定義
enum APIError: Error {
    case invalidValue
}

func addValue2(value: Int) -> AnyPublisher<Int, APIError> {
    Deferred {
        Future { promise in
            if Int.random(in: 0 ... 9) == 0 {
                // エラーを返す
                promise(.failure(.invalidValue))
            } else {
                // 成功したので10足した数を返す
                promise(.success(value + 10))
            }
        }
    }
    .eraseToAnyPublisher()
}

実行してみる#

エラーが発生すると、その時点で処理が終わりreceiveCompletionが呼ばれます。

RECEIVE 19
ERROR invalidValue // エラー発生

エラーが発生しない場合はそのまま普通に終わります。

リトライする#

publisher
    .receive(on: DispatchQueue.main)
    .flatMap({ addValue(value: $0) })
    .retry(10) // リトライ回数を書く
    .sink(receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("FINISHED")
        case .failure(let error):
            print("ERROR", error)
        }
    }, receiveValue: { value in
        print("RECEIVE", value)
    })
    .store(in: &task)

::: tip リトライ回数について

リトライを宣言すると、どこかでエラーが発生した場合に「最初からやり直す」という処理になってしまう。

失敗したものだけリトライする方法がないか探してみる。

:::

並列処理する#

publisher
    .receive(on: DispatchQueue.main)
    // 同時に実行するPublisher数を指定
    .flatMap(maxPublishers: .max(5), { addValue(value: $0)})
    .sink(receiveCompletion: { completion in
        switch completion {
        case .finished:
            print("FINISHED")
        case .failure(let error):
            print("ERROR", error)
        }
    }, receiveValue: { value in
        print("RECEIVE", value)
    })
    .store(in: &task)