RxJava 1.0コードをRxJava 2.0にアップデートする

Question

古いAndroidアプリケーションコードを更新しています。RxJava/RxAndroid 1.0で動作していたコードをRxJava/RxAndroid 2.0に変換する方法がわかりません。私が変換しようとしているコードは、信号強度インジケータを取得し、毎秒の値でチャートを更新する責任がありました。私がRxJava 2.0でハッキングしたものは、まったく動作せず、数秒後にSIGABORTでアプリがクラッシュした。私は何が欠けているのか分からないのですか？

マイRxJava 1つのコード：

rx.Observer myObserver = new rx.Observer<Float>() { 
      @Override 
      public void onCompleted() { 
      } 

      @Override 
      public void onError(Throwable e) { 

      } 

      @Override 
      public void onNext(Float aFloat) 
      { 
       addEntry(aFloat); 
      } 
     }; 

     subscription = rx.Observable.interval(1, TimeUnit.SECONDS) 
          .flatMap(new Func1<Long, rx.Observable<Float>>() { 
           @Override 
           public rx.Observable<Float> call(Long aLong) { 
            return getSignalStrength(); 
           } 
          }) 
          .subscribeOn(Schedulers.io()) 
          .observeOn(AndroidSchedulers.mainThread()) 
          .subscribe(myObserver); 


private rx.Observable<Float> getSignalStrength() 
{ 
    return rx.Observable.create(subscriber -> 
    { 
     WifiManager wifiManager = (WifiManager) getContext().getSystemService(Context.WIFI_SERVICE); 
     Integer value = wifiManager.getConnectionInfo().getRssi(); 
     subscriber.onNext(value.floatValue()); 
     subscriber.onCompleted(); 
    }); 
} 

そして私はRxJava 2.0をしようと思い付いた：

 DisposableObserver<Float> disposableObserver = new DisposableObserver<Float>() 
     { 
      @Override 
      public void onNext(Float value) 
      { 
       addEntry(value); 
      } 

      @Override 
      public void onError(Throwable e) 
      { 
      } 

      @Override 
      public void onComplete() 
      { 
      } 
     }; 

     Disposable disposable = Observable.interval(1, TimeUnit.SECONDS) 
       .flatMap(new Function<Long, ObservableSource<Float>>() 
       { 
        @Override 
        public ObservableSource<Float> apply(Long aLong) throws Exception 
        { 
         return signalStrengthInfo.getSignalStrength(); 
        } 
       }) 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribeWith(disposableObserver); 


private Observable<Float> getSignalStrength() 
{ 
    return Observable.create(subscriber -> 
    { 
     wifiManager = (WifiManager) context.getSystemService(Context.WIFI_SERVICE); 
     Integer signalStrength = wifiManager.getConnectionInfo().getRssi(); 
     subscriber.onNext(Float.valueOf(signalStrength)); 
     subscriber.onComplete(); 
    }); 
} 

Answer 1

Rx2の中にあなたのコードは本当に巻き込まれます。

ただ単純な解決策（kotlinには申し訳ありません）がありますが、Javaでコメントする場合に問題がある場合はこちらをご覧ください。

val disposable = Observable.interval(1, TimeUnit.SECONDS) 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .flatMap { getSignalStrength() } 
       .doOnNext { addEntry(it) } 
       .subscribe() 


private fun getSignalStrength(): Observable <Float> { 
      val wifiManager = this.getSystemService(Context.WIFI_SERVICE) as WifiManager 
      val signalStrength = wifiManager.connectionInfo.rssi 
      return Observable.just(signalStrength.toFloat()) 
    }