Как перехватить наблюдаемый объект и изменить его в 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. Я думаю, что вы хотите flatmap на молнию всех ваших звонков, так, что-то вроде этого:

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 апи:

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

Имейте в виду, что это будет сделать п + 1 сетевые вызовы, 1 для списка ExampleObject с, а затем 1 на ExampleObject в этом списке. Если это на всех возможно, я настоятельно рекомендую вам поговорить с разработчиком API, чтобы получить информацию о поиске по API-интерфейсу. Просто знайте, что это будет использовать некоторую пропускную способность!

Answer 2

Потому что ваш посредник вызова обновить запись является асинхронным, я не думаю, что вы можете использовать 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 зависят от внешней переменной (вход). Это то, что я стараюсь избегать, но здесь это все равно:

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