バックプレッシャーメカニクスを使用してページングでデータを取得する

Question

バックプレッシャーを使用してデータページングを処理しようとしています。私はすべての要求（N）のためにサーバーからN個の新しいチャンクを要求する観測可能性を持っています。ユーザーがボタンをクリックした後、新しいデータを取得したい。

私の主な目標は、私はジップが

Observable.zip(dataFromServer(), userActions(), (data, ø) -> data).subscribe(showData()); 

ここで素晴らしいことだと思うよしかし、理由はジップの内側Mの大きさにRxRingBufferの観察可能な契約

interface Api { 
    void openConnection(); //subscription 
    boolean hasNextPage(); //completion 
    Single<Data> nextPage();//onNext and onError 
    void closeConnection(); //unsubscription 
} 

により、このAPIを置き換えることです演算子は、サブスクリプション後すぐにサーバーからM個のデータチャンクを要求します。これは完全に不適切です。バッファサイズを1に変更することはできますが（私はできません）、上部構造の動作は不適切です。データは、最初のユーザー操作の前に要求されますが、最初のユーザー操作の後に最初に要求されます。私のアプローチが全く間違っているようです。

Answer 1

私は適切な動作を達成することができました。

//model of user actions 
val userActions = Observable.interval(1000, TimeUnit.MILLISECONDS).share();//user clicks every second// 
//model of my observable with data which supports backpressure 
val dataFromServer = Observable.range(1, 10);//data has 10 pages// 

dataFromServer 
    .doOnNext(it -> System.out.println("Received from server" + it)) 
    .flatMap(it -> userActions.take(1).ignoreElements().startWith(it), 1) 
    .doOnNext(it -> System.out.println("Before data showing" + it)) 
    .subscribe(showData());