RxSwiftでいい感じにQueue処理するRxObservableQueueを作った

サマリ

RxObservableQueue (Carthage, Cocoapods対応しました)
Observableでemitされたitemをqueueとして保持し、好きなタイミングで好きな並列数で取り出す事ができます。
(わかりづらい)

もうちょっとわかりやすいユースケース

例えば、ユーザ操作に対して大量のオブジェクトをネットワークから先読みをする時に使えます。
あめくるで使うために作りました。ここでの要件は下記になります。

• 表示している地図領域の、現在時刻分の画像(10枚程度)を最優先で取得する
• 未来の時間軸12個分(計120枚程度)を現在時刻に近い順に取得する
• 過去の時間軸12個分(計120枚程度)を現在時刻に近い順に取得する
• ユーザが表示領域を変更したら↑の取得処理は全てキャンセル
• 取得・画像処理は負荷が高いので同時処理数は制限したい

上記の処理をこんな感じでかけるようになります。

// 例: Observable<MKMapRect> ユーザが地図領域を変更するとemitされるObservable
visibleMapRect 
    // 地図領域をflatMapLatestし、最後の地図領域分以外はdispose()する
    .flatMapLatest { mapRect -> Observable<(Request, Counter)> in 
        // 地図領域から、先読み処理に必要な情報が入ったRequestをemitするObservableを作成する
        let request: Observable<Request> = makeRequest(mapRect)

        // RxObservableQueueで同時実行数3でキュー作成
        return RxObservableQueue.create(observable: request, maxConcurrentCount: 3)
    }
    .subscribe(onNext: { request, counter in
        // 同時実行数を超えない様にrequestがemitされる
        // 処理が終わったらコンプリーションハンドラでcounter.signal()すると、次のrequestがemitされる
        model.process(request, completionHandler: { counter.signal() })
    })
    .addDisposableTo(bag)