RxJava - параллельное выполнение двух вызовов API

Question

У меня есть SyncService.

@Override 
public int onStartCommand(Intent intent, int flags, final int startId) { 
    Timber.i("Starting sync..."); 
    ... 
    RxUtil.unsubscribe(mSubscription); 
    mSubscription = mDataManager.syncEvents() 
      .subscribeOn(Schedulers.io()) 
      .subscribe(new Observer<Event>() { 
       @Override 
       public void onCompleted() { 
        Timber.i("Synced successfully!"); 
        stopSelf(startId); 
       } 

       @Override 
       public void onError(Throwable e) { 
        Timber.w(e, "Error syncing."); 
        stopSelf(startId); 

       } 

       @Override 
       public void onNext(Event event) { 
       } 
      }); 

    return START_STICKY; 
} 

Наблюдаемые события = mDataManager.syncEvents() - вызов API.

Я хочу сделать параллельный вызов:

Single USERINFO = mDataManager.getUserInfo()

и вызвать stopSelf (startId); после завершения этих двух вызовов.

Как я могу это сделать?

Я попробовал RxJava Fetching Observables In Parallel , но это немного другой случай.

Я думаю, что мне нужно использовать .zip или .merge способ. Но в моем случае один вызов метода возвращает Observable (список в Events) и второй Single (один объект UserInfo).

Я создал класс г результат, который может быть результатом .zip метода, но я не знаю, как заполнить его:

public class SyncResponse { 
    List<Event> events; 
    UserInfo userInfo; 
    ... 
} 

Answer 1

Вы можете использовать observable.zip и single.toObservable для достижения этой цели , Я считаю, что вам нужно будет также выполнять планирование потоков в zip-вызове по каждому наблюдаемому отдельно, чтобы делать их параллельно.

Observable.zip(call1.subscribeOn(Schedulers.io()), call2.toObservable().subscribeOn(Schedulers.io(), zip-function) 

Поскольку оба вызова являются разными обернутыми типами.

Zip-функция - это то, где вы можете использовать свой класс для объединения результатов.

Следует отметить, что, поскольку один из ваших наблюдаемых является одиночным, этот метод с zip будет производить только один результат максимум, поскольку сингл отправит событие onComplete сразу после его первого события onNext.

Если вы знаете, что ваш список событий будет завершен, вы можете использовать toList в call1 для их буферизации и испускать их как единый, собранный список событий.

Answer 2

Поскольку у вас есть только два наблюдаемых, вы можете комбинировать их с помощью пары Android, чтобы комбинировать результаты.

mDataManager.syncEvents() 
    .zipWith(mDataManager.getUserInfo().toObservable(), Pair::create) 
    .subscribeOn(Schedulers.io()) 
    .subscribe(pair -> { 
     Event event = pair.first; 
     UserInfo userInfo = pair.second; 

     // your code here 

     Timber.i("Synced successfully!"); 
     stopSelf(startId); 
    }, throwable -> { 
     Timber.w(e, "Error syncing."); 
     stopSelf(startId); 
    }); 

Вы можете сделать то же самое без java8/retrolmbda но почему

Если вам нужно собрать все события до завершения и объединить их все с одной информации о пользователе, было бы немного сложным

Observable<ArrayList<String>> eventsObservable = mDataManager.syncEvents() 
    .collect(ArrayList::new, ArrayList::add); 

mDataManager.getUserInfo() 
    .zipWith(eventsObservable.toSingle(), Pair::create) 
    .subscribeOn(Schedulers.io()) 
    .subscribe(pair -> { 
     UserInfo userInfo = pair.first; 
     List<Event> event = pair.second; 

     // your code here 

     Timber.i("Synced successfully!"); 
     stopSelf(startId); 
    }, throwable -> { 
     Timber.w(e, "Error syncing."); 
     stopSelf(startId); 
    });