백 프레셔 메카닉을 사용하여 페이징으로 데이터를 가져옵니다.

Question

데이터 페이징을 처리하는 데 backpressure를 사용하려고합니다. 나는 모든 요청 (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를 대체하는 것입니다 operator 서브 스크립 션 이후의 상위 구성은 즉시 서버로부터 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());