스트림을 변경하지 않는 부작용 구독을 어떻게 가질 수 있습니까?

Question

시나리오 :

나는 데이터 목록을 가지고 있습니다 (업무용 트랜잭션). 다음 두 가지 조건이 충족되는 경우이 데이터는 오래된 간주됩니다

• 사용자 (UI /보기)가 데이터 소스에 가입 (AN rx.Observable<List<Transaction>>)
• > = 5 분 마지막 네트워크 동기화 이후 통과 . 이 값은 데이터베이스에서 직렬화되고 관찰됩니다.

참고 : UI /보기가 데이터 소스에 가입하지 않고,> = 5 오분이 지난 경우, 나는 (아무도 듣고 있지 않기 때문에)

일부를 새 데이터를 동기화하지 않으 아래의 코드 샘플에 대한 세부 사항 : -, 새로운 네트워크 요구를 가져 오기 위해 실행됩니다하지 않을 경우 마지막 5 분 동기가 있었는지

• data.observe() 반환 rx.Observable<List<Transaction>>
• void syncIfLast5Minutes() 명령 적으로 확인합니다 새 데이터 이제

, 나는 쉽게 같은 것을 할 수있는 :

data.observe() 
    .doOnSubscribe(transactions -> syncIfLast5Minutes() 

을하지만 데이터가 UI에 의해 초기 가입에 오래된 경우에만 확인합니다. UI가 5 분 후에도 가입되어있는 경우 여기에서 자동 새로 고침이 실행되지 않습니다 (하지만 원합니다). 데이터가 오래된 경우 제가

• 원래 스트림에 영향을 미치지 않을 것이다 부작용 연산자를 찾고

  는

• 구독 감시하는 타이머를 시작한다 (> = 5 분 경과)
• unsusbscribing는 반응 적이를 달성하기 RxJava의 관용적 방법이 있나요이 타이머

취소?

Answer 1

부작용 (데이터 새로 고침)을 수행하고 ignoreElements이 적용된 Observable.interval으로 스트림을 병합하여 원하는 것을 얻을 수 있습니다.

import java.util.concurrent.TimeUnit; 

import rx.Observable; 
import rx.subjects.PublishSubject; 

public class Main { 

    public static void main(String[] args) throws InterruptedException { 
     PublishSubject<String> data = PublishSubject.create(); 
     //refresh data every second for this demo 
     //for your use case it's every 5 minutes 
     data.mergeWith(Observable.interval(1, TimeUnit.SECONDS) 
       // cause timed side-effect that updates data stream 
       .doOnNext(n -> data.onNext(n + "")) 
       .ignoreElements() 
       .cast(String.class)) 
      .doOnNext(System.out::println) 
      .toBlocking() 
      .subscribe(); 
    } 

} 

여러 가입자가 당신이 .share보고 할 수 있습니다 스트림을 사용하는 경우가 있으므로 하나의 새로 고침 작업은 매 5 분을 발생합니다

이

은 작업 예입니다.