java - 在返回到订阅者之前，如何拦截一个可观察对象并在 RxJava 中修改它？

Question

我目前正在尝试点击一个服务并返回一个对象列表，在它返回给订阅者之前，我想对列表中的每个对象进行另一个同步调用以进行另一个服务调用以设置丢失 field 。我成功地进行了所有调用，但是订阅者中返回的对象具有我需要设置为空的字段。这是我的代码示例:

示例服务:

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

因为更新条目的中间调用是异步的，我认为您不能坚持使用 List<ExampleObject> , 但应该操纵 ExampleObject直接来自 Observable :

mService.getExampleObject()
    // Spread the list
    .flatMap(list -> Observable.from(list))
    // Update your object
    // Here we zip the object with the missing object,
    // so that when the missing object is obtained,
    // we update the entry and emit it.
    .flatMap(entry -> Observable.zip(
           Observable.just(entry), 
           mDocsService.getMissingObjectByFoo(entry.getFoo()), 
           (entry, missingObject) -> {
               entry.setBar(missingObject);
               return entry;
           })
    )
    // if you really want a map after all
    .toList();

边注:

如果您对 map 中的功能没问题，则可以跳过 zip取决于外部变量(条目)。这是我尽量避免的事情，但无论如何:

    .flatMap(entry -> mDocsService.getMissingObjectByFoo(entry.getFoo())
                                  .map(missingObject -> {
                                      entry.setBar(missingObject);
                                      return entry;
                                   })
    )