将回调替换为来自 RxJava 的可观察量

java rx-java

2022-09-01 05:59:42

我使用监听器作为回调来观察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");
    }
});

谢谢！

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

答案 2

您正在寻找：Completable.create

可满足：表示没有任何值的延迟计算，但仅指示完成或异常。该类遵循与反应流类似的事件模式：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*));