RxKotlin은 - 동적 관찰자 배열

Question

는 내가 단일 작업에 관찰자의 동적 목록을 가지고하는 방법을 그림을 시도하고 개조 2

과 함께 RxKotlin을 사용하고 있습니다. 전체 작업이/

을 실패 할 때까지, 나는 (캐시/메모리에 저장) 데이터 조작을하고 모든 통지해야하는 작업이 완료되면

첫 번째 관찰자가 작업을 트리거해야하고, 모든 추가 관찰자 기다려야합니다 관찰자들. 여기

내가 무슨 짓을했는지 :

class UserManager 
{ 
    val observers = ArrayList<Observer<ArrayList<User>>>() 
    var isFetchingUsers = false 

    fun getUsers(observer: Observer<ArrayList<User>>) 
    { 
     observers.add(observer) 

     if (isFetchingUsers) 
     { 
      return 
     } 

     api.getUserList.observeOn(AndroidSchedulers.mainThread()).subscribe(object : Observer<UserListResponse> 
     { 
      override fun onNext(response: UserListResponse) 
      { 
       // Do some manipulations on the response and notify all 

       observers.forEach { 
        it.onNext(response.getUsers()) 
       } 
      } 

      override fun onError(e: Throwable) 
      { 
       observers.forEach { 
        it.onError(Throwable()) 
       } 
      } 

      override fun onComplete() 
      { 
       isFetchingUsers = false 
       observers.clear() 
      } 

      override fun onSubscribe(d: Disposable) 
      { 
      } 
     }) 
    } 
} 

가 여기 개조 관찰 생성 (이 자바에 ..)

/** 
    * Get users 
    */ 
    public Observable<UserListResponse> getUserList() 
    { 
     return mService.getUserList().subscribeOn(Schedulers.io()); 
    } 

나는이

을 수행하기위한 더 나은 방법이있을거야

감사!

Answer 1

관찰 가능 항목에 share() 연산자를 사용할 수 있습니다. 첫 번째 구독에서만 관찰 대상이 생성 프로세스를 수행합니다. 마지막 구독자가 구독 취소되면 관찰 대상은 삭제 프로세스를 수행합니다.

Observable<Long> v; 
... 
Observable<Long> sharedObservable = v 
    .doOnSubscribe(() -> logger.debug("subscribe")) 
    .doOnUnsubscribe(() -> logger.debug("unsubscribe")) 
    .share(); 

... 
Subscription v1 = sharedObservable.subscribe(); 
Subscription v2 = sharedObservable.subscribe(); 
... 
v1.unsubscribe(); 
v2.unsubscribe(); 

구독 작업은 한 번만 발생합니다. 원래의 관찰 가능 항목에는 단 하나의 구독 작업 만 발생하고 한 번만 구독 취소된다는 것을 알 수 있습니다.