在返回订阅服务器之前,如何拦截可观察对象并在RxJava中修改它？

Question

我正在尝试命中一个服务并返回一个对象列表,在它返回给订阅者之前,我想对列表中的每个对象进行另一个同步调用,以进行另一个服务调用来设置一个丢失的字段.我已成功完成所有调用,但订阅者返回的对象具有此字段,我需要将其设置为null.这是我的代码示例:

示例服务:

rx.Observable<List<ExampleObject>> getExampleObject();
rx.Observable<MissingObject> getMissingObjectByFoo(@Path("foo") String foo);

示例类:

public class ExampleObject {
    String foo;
    MissingObject bar;

    public String getFoo() {
        return this.foo;
    }

    public void setFoo(String value) {
        this.foo = value;
    }

    public MissingObject getBar() {
        return this.bar;
    }

    public void setBar(MissingObject value) {
        this.bar = value;
    }
}

示例实施:

mService.getExampleObject().flatMap(new Func1<List<ExampleObject>, Observable<?>>() {
            @Override
            public Observable<List<ExampleObject>> call(List<ExampleObject> exampleObjects) {
                for (ExampleObject entry : exampleObjects) {
                    String foo = entry.getFoo();
                    mService.getMissingObjectByFoo(foo)
                            .subscribeOn(mScheduler.backgroundThread())
                            .observeOn(mScheduler.mainThread())
                            .subscribe(new Subscriber<MissingObject>() {
                        @Override
                        public void onCompleted() {

                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onNext(MissingObject missingObject) {
                            entry.setBar(missingObject);
                        }
                    });
                }
                return Observable.just(exampleObjects);
            };

Answer 1

您正在寻找zip运算符，如下所述：Zip Operator。我认为您想将所有调用平面映射到一个 zip，所以，像这样：

    mService.getExampleObject().flatMap(new Func1<List<ExampleObject>, Observable<ExampleObject>>() {
        @Override
        public Observable<List<ExampleObject>> call(List<ExampleObject> exampleObjects) {
            List<Observable<ExampleObject>> allTheObservables = new ArrayList<Observable<ExampleObject>>();
            for (ExampleObject entry : exampleObjects) {
                allTheObservables.add(mService.getMissingObjectByFoo(foo).map(new Func1<MissingObject, ExampleObject>() {
                    @Override
                    public ExampleObject call(MissingObject missingObject) {
                        return entry.setBar(missingObject);
                    }
                }));
            }
            return Observable.zip(allTheObservables, new FuncN<ExampleObject>() {
                @Override
                public ExampleObject call(ExampleObject... args) {
                    return Arrays.asList(args);
                }
            });
        }
    });

如果这不起作用，或者存在语法问题，这里有一个使用 github api 的具体示例：

    service.getContributorsObservable("square", "dagger")
        .flatMap(new Func1<List<Contributor>, Observable<List<String>>>() {
            @Override
            public Observable<List<String>> call(List<Contributor> contributors) {
                List<Observable<String>> allTheObservables = new ArrayList<>(contributors.size());
                for (final Contributor contributor : contributors) {
                    allTheObservables.add(service.getContributorsObservable(contributor.login).map(new Func1<User, String>() {

                        @Override
                        public String call(User user) {
                            return contributor.login + " is " + user.name;
                        }
                    }));
                }
                return Observable.zip(allTheObservables, new FuncN<List<String>>() {
                    @Override
                    public List<String> call(Object... args) {
                        return Arrays.asList((String[]) args);
                    }
                });
            }
        });

请记住，这将进行 n+1 次网络调用，其中 1 次用于 s 列表ExampleObject，然后针对ExampleObject该列表中的每个进行 1 次。如果可能的话，我强烈建议您与 API 的维护者联系，以便在 API 端处理信息查找。只需知道这将使用一些带宽！