RxAndroidで加入者を停止する方法は？

Question

は私が観察可能な、次のように作成します。

public Observable<FileObjectModel> getAllFiles() { 
    return Observable.create(new Observable.OnSubscribe<FileObjectModel>() { 
     @Override 
     public void call(Subscriber<? super FileObjectModel> subscriber) { 
      //fileList is a huge list. 
      for(int i = 1 ; i < fileList.size(); i ++){ 
       ... 
       subscriber.onNext(fileItem); 
      } 
     } 
    }); 
} 

は、このようにサブスクライブ：

subscription = reposistory.getAllFiles() 
      .compose(scanInject()) 
      .subscribe(...); 

ユーザーのクリックボタンストップは、私はサブスクリプションをキャンセルしたいので、私は呼ん

subscription.unsubscribe(); 

残念ながら、期待どおりに停止しません。ファイルアイテムは引き続き放出されます。

私の間違いを指摘してください。

UPDATE：

Iのデバッグ行ずつと、観察がforループに陥っていることがわかります。このループでは、再帰的メソッドを呼び出し、サブスクライバをパラメータとして渡します。この方法では、すべてのファイルを検索するために、すべてのAndroidファイルシステム（ルートフォルダから開始）をブラウズし、次にそれらを1つずつ発行します。 （システム全体のファイル数が非常に多いため、この方法を使用します。静的リストには保存しないでください） forループに詰まっているので、サブスクリプションオブジェクトは常にnullです。まだ解決策が見つかりません。

Answer 1

加入者がまだ加入している場合、私はチェックをお勧めします：

for(int i = 1 ; i < fileList.size(); i ++){ 
    if (subscriber.isUnsubscribed()) { 
     return; 
    } 
    subscriber.onNext(fileItem); 
} 

これはあまりにも多くの計算過負荷であってはなりません。

Answer 2

Observable.createを使用する場合、多くの可能な下流側演算子に必要なバックプレッシャーを守るために、.onBackpressureXXXを追加する必要があります。疑わしい場合は.onBackpressureBufferを使用してください。

あなたの観測対象が同期しているため、ストリームが終了するまでsubscriptionが設定されていないため、退会できない理由があります。私はとにかく私の質問に答えることを憎むが、私はすでにそれを修正するために根本的な原因とアクションを見つけ

subscriber = ...; 
reposistory.getAllFiles() 
    .compose(scanInject()) 
    .subscribe(subscriber); 

Answer 3

を：私は一般的に、このパターンを避け示唆し、代わりに使用すると思います。私は私の質問の更新部分で述べたように

は、forループは、アイテムの数千を繰り返し、膨大な仕事をしてくれました。そして、プロセッサーは内部についていた。

これを修正するために、私は、観察に異なるスレッドを割り当て、それはこのように、非同期に実行されます。決してnull前と同じように、今

subscription = getAllObjects() 
      .map(this::doScan) 
      .observeOn(Schedulers.newThread()) 
      .subscribeOn(Schedulers.newThread()) 
      .subscribe(subscriber); 

変数subscription、我々はsubscription.unsubscribe()を呼び出すことによって、サブスクリプションを停止するためにそれを使用することができます