java - RxAndroid 链式操作

Question

我正在使用 RxAndroid + Retrofit，我想做的是将 2 个操作合并为一个。例如我有两种方法。第一个在 onNext block 中返回一些 URL。

apiRequests.getApi().getURL()
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<CreateUserResponse>() {
                @Override
                public void onCompleted() {
                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(CreateUserResponse createUserResponse) {
                    secondMethos(createUserResponse.getURL()); // HERE GOES THE URL

                }
            });

第二种方法使用此 URL 执行其他操作

public void secondMethod(URL){
    Observable.create(new Observable.OnSubscribe<Object>() {


        @Override
        public void call(Subscriber<? super Object> subscriber) {

        }
    });

    Observable.create(new Observable.OnSubscribe<List<Arg>>() {
        @Override
        public void call(Subscriber<? super List<Arg>> subscriber) {
            subscriber.onNext(test(URL));
            subscriber.onCompleted();

        }
    }).subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<List<Arg>>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(List<Arg> params) {

                }
            });


}

是否可以将这些方法合而为一？谢谢

Answer 1

您可以简单地映射这些值

    apiRequests.getApi().getURL()
            .map(new Func1<CreateUserResponse, List<Arg>>() {
                public List<Arg> call(CreateUserResponse response) {
                    return test(response.getURL());
                }
            })
            .subscribeOn(Schedulers.newThread())
            .observerOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<List<Arg>>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(List<Arg> params) {

                }

请注意，如果您希望每个 Arg 都调用 onNext而不是List<Arg>您可以使用 flatMapInterable，如下所示。

    apiRequests.getApi().getURL()
            .flatMapIterable(new Func1<CreateUserResponse, List<Arg>>() {
                public List<Arg> call(CreateUserResponse response) {
                    return test(response.getURL());
                }
            })
            .subscribeOn(Schedulers.newThread())
            .observerOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<Arg>() {
                @Override
                public void onCompleted() {

                }

                @Override
                public void onError(Throwable e) {

                }

                @Override
                public void onNext(Arg params) {

                }
            });