さまざまなパラメータで結果のカウントとキャッシングを保証する方法

Question

リクエストパラメータに基づいて異なる結果になるAPIエンドポイントがあります。パラメータはpage,per_page,queryなどです。

fun getItems(params : Map<String, String>) : Single<ItemsResponse> 

data class ItemsResponse(
    val hasMore : Boolean, 
    val items : List<Items> 
) 

APIは信頼性が低く、per_page未満の値を返す可能性があります。私は、常に私が必要とする結果の数を取得し、次の要求サイクルのために残りをキャッシュするようにしたい。
例えば何か

val page : Int = 1 

fun fetchItems(requestedItems : Int = 20) : Single<List<Items>> { 
    ... 
    .map { buildParams(page, perPage, query) } 
    .flatMap { api.getItems(it) } 
    .doOnSuccess { page++ } 
    .buffer(requestedItems) 
} 

fun buildParams(page: Int, perPage: Int, query : String) : Map<String, String> { 
    ... 
} 

シナリオ例：

• 発信者は、最初に20の項目を要求します。 page: 1とapi.getItems()から
• コール、per_pageが常に20
  • コールは16の項目
• コールapi.getItems()
  • page: 2とコールリターンに19の項目
• 20を返しますアイテムは呼び出し元に返され、15残りのアイテムは次の発呼者の要求のためにキャッシュされました。
• 発信者は2回目に20アイテムをリクエストします。 page: 3
  • 呼にapi.getItems()から
  • 電話で呼び出し元に戻された12の項目
• 20アイテムを返す（15古い5最後の応答から）7つの残りの項目は次の発信者の要求のためにキャッシュされました。
• などなど。

これはプロデューサ - コンシューマパターンのようですが、RxJava2で実行できますか？

Answer 1

編集：追加情報

に基づいて必要なもの：RxJava 2つの拡張ライブラリを：compile "com.github.akarnokd:rxjava2-extensions:0.17.0"

import hu.akarnokd.rxjava2.expr.StatementObservable 
import io.reactivex.Observable 
import io.reactivex.functions.BooleanSupplier 
import io.reactivex.subjects.PublishSubject 
import java.util.concurrent.Callable 
import java.util.concurrent.ConcurrentLinkedQueue 
import java.util.concurrent.ThreadLocalRandom 

var counter = 0; 

fun service() : Observable<String> { 
    return Observable.defer(Callable { 
     val n = ThreadLocalRandom.current().nextInt(21) 
     val c = counter++; 
     Observable.range(1, n).map({ v -> "" + c + " | " + v }) 
    }) 
} 

fun getPage(pageSignal : Observable<Int>, pageSize: Int) : Observable<List<String>> { 
    return Observable.defer(Callable { 
     val queue = ConcurrentLinkedQueue<String>() 

     pageSignal.concatMap({ _ -> 
      StatementObservable.whileDo(
        service() 
          .toList() 
          .doOnSuccess({ v -> v.forEach { queue.offer(it) }}) 
          .toObservable() 
        , BooleanSupplier { queue.size < pageSize }) 
        .ignoreElements() 
        .andThen(
          Observable.range(1, pageSize) 
            .concatMap({ _ -> 
             val o = queue.poll(); 
             if (o == null) { 
              Observable.empty() 
             } else { 
              Observable.just(o) 
             } 
            }) 
            .toList() 
            .toObservable() 
        ) 
     }) 
    }) 
} 

fun main(args: Array<String>) { 

    val pages = PublishSubject.create<Int>(); 

    getPage(pages, 20) 
      .subscribe({ println(it) }, { it.printStackTrace() }) 

    pages.onNext(1) 

    pages.onNext(2) 
}