Заменить обратные вызовы с помощью наблюдаемых из RxJava

Question

Я использую прослушиватели в качестве обратных вызовов для наблюдения за асинхронными операциями с Android, но я думаю, что это может быть отличным замещением этих слушателей RxJava, я буду использовать эту библиотеку, но мне это очень нравится, и я всегда использую ее с помощью 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 и лямбда-выражения.

редактировать:

если я вас понимаю хорошо, вы хотите заменить, что слушатель, не завернуть его в наблюдаемый. Я добавил другой пример в ваш комментарий. О .. также вы должны использовать Одиночный, если вы ожидаете только одного элемента.

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 обернутый как Single:

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

Теперь вы выставили API Rx наряду с одной функции обратного вызова в. Предположив, что это какой-то DataProvider свой собственный, теперь вы можете использовать его без решения обратных вызовов, например:

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

Я использовал Rx2 но Прм1 логика та же.

Я также использовал Single вместо наблюдаемого, так как вы ожидаете только одного значения. Интерес представляет собой более выразительный контракт для вашей функции.

Вы не можете испускать ценность от имени наблюдаемого, то есть вызывать что-то вроде myObservable.send (значение). Первое решение - использовать Subject. Другое решение (выше) заключается в создании наблюдаемого с помощью Observable.create() (или Single.create()). Вы вызываете метод обратного вызова и создаете слушателя внутри метода Observable.create(), потому что он находится внутри Observable.create(), который вы можете вызвать методом onSuccess(), метод, который сказал 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); 
     }); 
     } 
    });