Stopp beobachtbare von Schleife

Question

Vor kurzem habe ich auf RxJava gearbeitet 2 und ich habe die Observable.interval getestet()

subscription = Observable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io()).observeOn(AndroidSchedulers.mainThread()); 

subscription.subscribe(new Observer<Long>() { 
      @Override 
      public void onSubscribe(Disposable d) { 

      } 

      @Override 
      public void onNext(Long aLong) { 
       //binding.appBar.mainContent.msg.setText(aLong+""); 
      } 

      @Override 
      public void onError(Throwable e) { 

      } 

      @Override 
      public void onComplete() { 

      } 
     }); 

beobachtbare nach der Aktivität onCreate Methode gestartet wird. Und ich protokolliere die Ausgabe über onNext() Methode. Und ich habe einen Stop Button. Wenn es ausgelöst wird, möchte ich den Abonnementfluss stoppen.

Auch nach dem Klicken auf die Schaltfläche "Stopp" wird das Protokoll fortgesetzt.

stop.setOnClickListener(new View.OnClickListener() { 
      @Override 
      public void onClick(View v) { 
       if (subscription != null) { 
        subscription.unsubscribeOn(Schedulers.io()); 
       } 
      } 
     }); 

Answer 1

Sie haben mit einem Observer abonniert haben, was bedeutet, dass Sie einen Verweis auf die tatsächlichen Disposable von onSubscribe(Disposable) Rückruf zu halten haben, und später ausführen Disposable#dispose() auf dem Objekt.

 


    private Disposable disposable; 

    ... 

    Observable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(
      new Observer() { 
       @Override public void onSubscribe(Disposable d) { 
       disposable = d; 
       } 

      // other callbacks here 

      }); 

    disposable.dispose(); 

 

Stattdessen können Sie Ihr Abonnement folgt ändern:

 


    Disposable disposable = Observable.interval(1, TimeUnit.MILLISECONDS, Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .subscribe(new Consumer() { 
      @Override public void accept(Long aLong) throws Exception { 
      // onNext 
      } 
     }, new Consumer() { 
      @Override public void accept(Throwable throwable) throws Exception { 
      // onError 
      } 
     }, new Action() { 
      @Override public void run() throws Exception { 
      // onComplete 
      } 
     }); 

    disposable.dispose();