Rxリミット+タイムアウトを伴う高速生成ストリームのバッファリング

Question

RxJavaでWebサービスコールを最適化しようとしていますが、応答に時間がかかりすぎることはありません。そのために は、私はこのようなセレクタを閉じるようdebounce()でbuffer(closingSelector)演算子を使用しています：

Observable<BaseCall<T, R>> burstyMulticast = requestStream.share(); 
Observable<BaseCall<T, R>> burstyDeBounced = burstyMulticast.debounce(windowSize, windowUnit); 
burstyMulticast.buffer(burstyDeBounced).subscribe(/* call external WS with batches */); 

それはrequestStreamが速すぎて生成する場合、それはWSが一度に処理するには大きすぎる巨大なバッチを放出することを除いて、正常に動作し、何とかバッチサイズを制限したいと思います。 したがって、バッファにX項目があるか、アップストリームから最後に到着してからYが経過した場合に閉じるイベントを発行するclosingSelectorが必要です。

Operatorというカスタムを実装する以外の良い解決策は見つからないようですが、OperatorDebounceWithTimeに似ていますが、最後のバッファではなくバッファ内のすべての要素を返す内部バッファがあります。

これを実現する簡単な方法はありますか？いくつかの操作を組み合わせることによって？

編集：私は、コード片が上記別の問題があることに気づい

質問を投稿した後：要求がデバウンスタイムアウトよりも継続的に速く流れる場合（requestStreamwindowSizeよりも早く生産しているが）、その後burstyDeBouncedは何も放出しません受信ストリームに十分長いポーズがあるまで、すべての要求がバッファされます。

Answer 1

私はカスタムオペレーションを実装しました：https://gist.github.com/zsoltm/79462b37c0943b4fbef2ee3968155f27これはうまくいくようです。私は改善のための提案をしてうれしいです。

Answer 2

あなたは小さなものにデバウンスソースの大容量バッファを分割することができます：

Observable<BaseCall<T, R>> burstyMulticast = requestStream.share(); 
Observable<BaseCall<T, R>> burstyDeBounced = burstyMulticast 
    .debounce(windowSize, windowUnit); 

burstyMulticast.buffer(burstyDeBounced) 
.onBackpressureBuffer() 
.concatMapIterable(list -> Lists.partition(list, windowSizeLimit)) 
.subscribe(...); 

Lists.partition

をGoogleグアバからです。