用RxJava中的可觀察元素替換回調

Question

我使用監聽器作爲回調來觀察與Android的異步操作，但我認爲用RxJava代替這個監聽器會很好，但是我真的很喜歡它，Im總是和它一起使用Android項目。

這裏是我的代碼重構：

public void getData(final OnResponseListener listener){ 
    if(data!=null && !data.isEmpty()){ 
     listener.onSuccess(); 
    } 
    else{ 
     listener.onError(); 
    } 
} 

一個簡單的回調：

public interface OnResponseListener { 
    public void onSuccess(); 
    public void onError(); 
} 

而 「觀察員」：

object.getData(new OnResponseListener() { 
    @Override 
    public void onSuccess() { 
     Log.w(TAG," on success"); 
    } 

    @Override 
    public void onError() { 
     Log.e(TAG," on error"); 
    } 
}); 

謝謝！

Answer 1

例如，您可以使用Observable.fromCallable來爲您的數據創建可觀察項。

public Observable<Data> getData(){ 
    return Observable.fromCallable(() -> { 
     Data result = null; 
     //do something, get your Data object 
     return result; 
    }); 
} 

然後使用您的數據

getData().subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(data -> { 
       //do something with your data 
      }, error -> { 
       //do something on error 
      }); 

二手rxjava 1.x和lambda表達式。

編輯：

如果我理解你很好，你想更換聽衆，不是包裝成可觀察的。我參考你的評論添加了其他例子。哦..也應該使用單個如果您只需要一個項目。

public Single<Data> getData() { 
     return Single.create(singleSubscriber -> { 
      Data result = object.getData(); 
      if(result == null){ 
       singleSubscriber.onError(new Exception("no data")); 
      } else { 
       singleSubscriber.onSuccess(result); 
      } 
     }); 
    } 

getData().subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(data -> { 
       //do something with your data 
      }, error -> { 
       //do something on error 
      }); 

Answer 2

我該如何重構你的代碼;旁邊GetData方法，我想補充GetData方法包裝成單：

public void getData(final OnResponseListener listener){ 
    if(data!=null && !data.isEmpty()){ 
     listener.onSuccess(); 
    } 
    else{ 
     listener.onError(); 
    } 
} 

public Single<Boolean> getDataSingle() { 
    return Single.create(new SingleOnSubscribe<Boolean>() { 
     @Override 
     public void subscribe(SingleEmitter<Boolean> e) throws Exception { 
      getData(new OnResponseListener() { 
       @Override 
       public void onSuccess() { 
        e.onSuccess(true); 
       } 

       @Override 
       public void onError() { 
        e.onSuccess(false); 
       } 
      }); 
     } 
    }); 
} 

或者與Java 8：

public Single<Boolean> getDataSingle() { 
    return Single.create(e -> getData(
      new OnResponseListener() { 
       @Override 
       public void onSuccess() { 
        e.onSuccess(true); 
       } 

       @Override 
       public void onError() { 
        e.onSuccess(false); 
       } 
      }) 
    ); 
} 

現在你已經暴露的Rx API一起回調的一個。假設它是某種你自己的DataProvider的，你現在就可以使用它，而有回調處理，像這樣：

dataProvider.getDataSingle() 
     .map(result -> result ? "User exist" : "User doesn't exist") 
     .subscribe(message -> display(message)); 

我用Rx2的，但與RX1的邏輯是一樣的。

我還使用了Single而不是Observable，因爲您只等待一個值。興趣是您的功能更具表現力的合同。

您不能代表Observable發出值，即調用類似myObservable.send（value）的值。第一個解決方案是使用Subject。另一個解決方案（上面的解決方案）是使用Observable.create（）（或Single.create（））創建observable。您調用回調方法並在Observable.create（）方法內部創建監聽器，因爲它位於Observable.create（）內部，您可以調用onSuccess（）方法，該方法告訴Observable傳遞值。

這是我用來包裝回調成observable。起初有點複雜，但容易適應。

我問你另一個例子。比方說，你想作爲一個小吃吧，以顯示一個EditText的變化：

View rootView; 
EditText editTextView; 

//Wrap Android addTextChangedListener into an Observable 
Observable<String> textObservable = Observable.create(consumer -> 
     editTextView.addTextChangedListener(new TextWatcher() { 
      @Override 
      public void beforeTextChanged(CharSequence s, int start, int count, int after) { 

      } 

      @Override 
      public void onTextChanged(CharSequence s, int start, int before, int count) { 

      } 

      @Override 
      public void afterTextChanged(Editable s) { 
       consumer.onNext(s.toString()); 
      } 
     }) 
); 

//Use it 
textObservable.subscribe(text -> Snackbar.make(rootView, text, Snackbar.LENGTH_SHORT).show()); 

Answer 3

您正在尋找Completable.create：

Completable：表示無任何價值，但只完成指示的延遲計算或例外。該類遵循與Reactive-Streams：onSubscribe（onError | onComplete）類似的事件模式？

Completable.create(subscriber -> { 
    object.getData(new OnResponseListener() { 
     @Override 
     public void onSuccess() { 
      subscriber.onCompleted(); 
     } 

     @Override 
     public void onError() { 
      subscriber.onError(* put appropriate Throwable here *); 
     } 
    } 
}) 
...//apply Schedulers 
.subscribe((() -> *success*), (throwable -> *error*)); 

Answer 4

Maybe.<String>create(new MaybeOnSubscribe<String>() { 
     @Override 
     public void subscribe(MaybeEmitter<String> e) throws Exception { 
     OnSuccessListener(uri->{ 
      e.onSuccess(uri)); 
     }) 
     .addOnFailureListener(throwable -> { 
      e.onError(throwable); 
     }); 
     } 
    });