RxJava 1.0-Code auf RxJava 2.0 aktualisieren

Question

Ich aktualisiere einen alten Android-App-Code und kann nicht herausfinden, wie Code, der in RxJava/RxAndroid 1.0 funktioniert, in RxJava/RxAndroid 2.0 konvertiert wird. Der Code, den ich zu konvertieren versuche, war dafür verantwortlich, den Signalstärkeindikator zu erhalten und jede Sekunde ein Diagramm mit seinem Wert zu aktualisieren. Was ich in RxJava 2.0 gehackt habe funktioniert überhaupt nicht und nach wenigen Sekunden stürzt die App mit SIGABORT ab. Ich kann nicht herausfinden, was ich vermisse?

Mein RxJava 1 Code:

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(); 
    }); 
} 

Und was ich kam mit dem Versuch, RxJava 2.0 zu tun:

 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

Ihr Code in Rx2 ist wirklich verstrickt.

Sehen Sie einfach einfache Lösung (Entschuldigung für Kotlin), aber wenn Sie ein Problem mit machen es mit Java haben, nur ping mich im Kommentar.

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