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) 
      { 
      } 
     }) 
    } 
} 

ここでレトロフィット観察可能な作成があります（この1つはJavaである。）

/** 
    * 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(); 

サブスクライブ操作は1回しか発生しません。元のオブザーバブルに対しては1つのサブスクリプション操作が行われ、サブスクライブ解除は1つのみであることがわかります。