RxJavaサブスクリプションが正しくサブスクライブされない

Question

ストリームの現在の値でビューを更新する必要があるMVCアプリケーションがあります。私は、ストリームを取得するには、コントローラで使用

public Observable<Integer> getStreamInstance(){ 
    if(stream == null){ 
     this.stream = Observable.create((Subscriber<? super Integer> subscriber) -> { 
      new HeartbeatStream(frequence,subscriber).start(); 
     }); 
    } 

    return stream; 
} 

：私はこの方法を持っているモデルでは 。私は単に現在の値に鑑みてラベルを更新するスタート方法で

public void start(){ 
    this.sb = stream.subscribe((Integer v) -> { 
     view.updateCurrValue(v); 
    }); 
} 

public void stop(){ 
    this.sb.unsubscribe();  
} 

：次に、コントローラに私はこれらの2つの方法を持っています。 登録を解除して更新を停止しようとするまでうまくいきます。実際には、ビュー内のボタン「停止」を押すと、ラベルは現在の値で更新され続け、「開始」をもう一度押すと、ラベルには2つの異なるストリームの値が表示されます。最初の "開始"と思われる2番目の "開始"の2番目のプレスで作成されています。 どこが間違っていますか？

EDIT：

public class HeartbeatStream extends Thread{ 

private Subscriber<? super Integer> subscriber; 
private int frequence; 
private HeartbeatSensor sensor; 

public HeartbeatStream(int freq, Subscriber<? super Integer> subscriber){ 
    this.frequence = freq; 
    this.subscriber = subscriber; 
    sensor = new HeartbeatSensor(); 
} 

public void run(){ 

    while(true){   
     try { 
      subscriber.onNext(sensor.getCurrentValue()); 
      Thread.sleep(frequence); 
     } catch (Exception e) { 
      subscriber.onError(e); 
     } 
    } 


} 

これはHeartbeatStreamクラスです。 HeartbeatSensorは、ハートビートの頻度をシミュレートする値を定期的に生成するクラスです。

Answer 1

観察可能で、すべてのアイテムが放出され、onCompleteのコールバックが呼び出されることを一度あなたのオブザーバーを解除：そのための演算子があります。

この例を見るhttps://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/creating/ObservableSubscription.java

Answer 2

私はあなたのHeartbeatStreamクラスで何が起こっているのか分かりませんが、unsubscribeは扱っていないと思います。

をObservable.createで作成する場合は、subscriber.isUnsubscribed()で明示的にサブスクライブ解除を処理する必要があります。

可能であれば、Observableを作成するユーティリティメソッドをいくつか使用します。これらはすべてあなたのためにこのすべてを処理します。たとえば、Observable.just()またはObservable.from()です。

これで問題が解決しない場合は、HeartbeatStreamクラスを投稿してください。

は詳細についてはドキュメントを参照してください。 https://github.com/ReactiveX/RxJava/wiki/Creating-Observables https://github.com/ReactiveX/RxJava/wiki/Async-Operators

Answer 3

私はあなたが定期的に画面更新をトリガーするいくつかのイベントを通知しようとした推測しています。デザインによって

Observable<Long> timer = Observable.interval(period, TimeUnit.MILLISECONDS, 
    AndroidSchedulers.mainThread()); 

SerialSubscription serial = new SerialSubscription(); 

public void start() { 
    serial.set(timer.subscribe(v -> view.updateCurrValue(v))); 
} 

public void stop() { 
    serial.set(Subscriptions.unsubscribed()); 
} 

public void onDestroy() { 
    serial.unsubscribe(); 
}