Как с помощью rxjava2 совместить несколько связанных запросов в наборе данных?

Question

[archduke]

Привет.
Как можно с помощью rxjava2 осуществить такую последовательность действий?:
Есть приблизительно такая структура:

class Holder {
  List<Image> images;
}
class Image {
  String url;
  String localFileUrl;
}

Нужно:

* Пройти по списку Holder'ов;
  * Для каждого Holder'а пройти по Image'ам;
    * Декодировать bitmap из Image.localFileUrl и зааплоадить bitmap на сервер и получить url, 
      записать его в Image.url; 
* После того как обошли все Holder и для каждого Image внутри зааплоадили картинку и получили url, 
  зааплоадить список Holder'ов;

Пробовал делать как то так:

Observable.fromIterable(holders)
      .concatMap(holder -> {
        Observable.fromIterable(holder.images)
          .concatMap(image -> {
            // decode bitmap here
            uploadImage(bitmap)
          })
      })
    // what should I do here to get observable which emits holders,
    // not result from uploadImage(bitmap)?
      .toList()
      .flatMap(? -> uploadeHolders());

Уже не имею идей возиожно ли подобного добиться с rxjava? Подходящего оператора не нашел для подобного, тут видимо нужен какой то специальный подход, но не могу сообразить как надо? Помогите пожалуйста.

Answer 1

[Денис Загаевский]

Observable.fromIterable(holders) //Observable<Holder>
    .concatMap(holder -> Observable.fromIterable(holder.images) //Observable<Image>
        .concatMap(image -> uploadImage(image)) //Observable<Image>
        .toList() //Observable<List<Image>>
        .map(images -> holder)) //Observable<Holder>
    .toList() //Observable<List<Holder>>
    .switchMap(holders -> uploadHolders(holders))
    .subscribe();

Обрати внимание, что в Rx строго рекомендуется иметь иммутабельные данные. То есть поля в твоих классах должны быть final, иначе ты очень сильно огребешь на конкурентности. Если пишешь на джаве, лучше использовать AutoValue для классов с данными.

Comments

[archduke]

Спасибо, работает.
А

.switchMap(holders -> uploadHolders(holders))

так понимаю тут не обязательно именно switchMap использовать, можно и flatMap(учитывая что uploadHolders() будет давать только 1 элемент)?
С иммутабельными данными есть проблема:

uploadImage(image).map(url -> {
            image.url = url;
            return image;
        )

надо делать так после аплоадинга картинки, можно конечно создавать новый обьект Image, но его надо будет каким то образом правильно добавлять в Holder. На конкурентности, как понимаю, могут быть проблемы если после вызова этого кода, другой поток будет пытаться поменять что то(Image или Holder), но внутри этого блока проблем не будет. Если интерфейс блокируется во время это операции, и никакого постороннего влияния на Holder или Image не будет, то проблем с конкурентностью быть не должно?

[Денис Загаевский]

archduke, можно и flat
Не надо ничего добавлять в holder, holder тоже должен быть иммутабельным.

.concatMap(image -> uploadImage(image).map(url ->  new Image(image.localFileUrl, url)))
.toList()
.map(images -> new Holder(images))

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

[archduke]

Еще есть один вопрос. Если нужео с помощью rxjava симмитировать что то вроде stream из java8, обычно рекомендуют делать так:

Observable.just("a", "b", "c", "d", "e", "f")
    .map(s -> s + "1")
    .toList()
    .subscribe(mView::show)
);

но в данном случае нужно еще c Disposable возиться(хотя если вызвать dispose() сразу после subscribe(), то все будет номально, но все равно выглядит как то перегруженно, если все выполняется в одном потоке, нужна просто связка операторов). Такой способ будет лучше или нет?:

mView.show(Observable.just("a", "b", "c", "d", "e", "f")
    .map(s -> s + "1")
    .toList()
    .blockingGet()
);

Ну т.е. если нужно сделать что то без подписок, и ближе к стилю stream java8, то можно преобрадовывать в Single и делать blockingGet()?

[Денис Загаевский]

archduke, Это какой-то адовый оверхед, не надо так. Возьми какой-нибудь бэкпорт апи стримов и не парься. Например https://github.com/Archinamon/Lightweight-Stream-API