RxJava + retrofit, get a list and add additional information for each element

Question

I play RXJava, modify Android. I am trying to do the following:

I need to periodically interrogate the call that will give me Observable> (From here I could do this)

As soon as I get this list, I want to iterate through each delivery and call the other methods that will give me ETA (so just more information). I want to add this new information to the delivery and return a complete list with additional information attached to each item.

I know how to do this without rxjava as soon as I get the list, but I would like to practice.

This is my code:

pollDeliveries = Observable.interval(POLLING_INTERVAL, TimeUnit.SECONDS, Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR)) .map(tick -> RestClient.getInstance().getApiService().getDeliveries()) .doOnError(err -> Log.e("MPB", "Error retrieving messages" + err)) .retry() .subscribe(deliveries -> { MainApp.getEventBus().postSticky(deliveries); }); 

This gives me a list of supplies. Now I would like to complete the second part.

I hope I was clear enough. Thanks

Answer 1

Finally, I found a good way to do this.

 private void startPolling() { pollDeliveries = Observable.interval(POLLING_INTERVAL, TimeUnit.SECONDS, Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR)) .flatMap(tick -> getDeliveriesObs()) .doOnError(err -> Log.e("MPB", "Error retrieving messages" + err)) .retry() .subscribe(this::parseDeliveries, Throwable::printStackTrace); } private Observable<List<Delivery>> getDeliveriesObs() { return RestClient.getInstance().getApiService().getDeliveries() .flatMap(Observable::from) .flatMap(this::getETAForDelivery) .toSortedList((d1, d2) -> { if (d1.getEta() == null) { return -1; } if (d2.getEta() == null) { return 1; } return d1.getEta().getDuration().getValue() > d2.getEta().getDuration().getValue() ? 1 : -1; }); } 

Release step by step.

• First, we create an Observable that starts each POLLING_INTERVAL value with the getDeliveriesObs () method, which will return the final list
• We use the modification to receive the observed call.
• We use flatMap to flatten the reinstallation list and in the following plan we collect the delivery point one by one.
• Then we get the estimated arrival time set inside the Delivery object and return it
• We sort the order list by estimated time of arrival.
• In case of an error, we print and repeat so that the interval does not stop
• Finally, we sign up to sort the list and with ETA inside, then we just return it or something else that you need to do.

It works correctly, and it is pretty good, I start to love rxjava :)

Answer 2

I did not spend much time on Java 8 lambdas, but here is an example of matching each object with another object, and then getting List<...> on the other end in plain ol 'Java 7:

 List<Delivery> deliveries = ...; Observable.from(deliveries).flatMap(new Func1<Delivery, Observable<ETA>>() { @Override public Observable<ETA> call(Delivery delivery) { // Convert delivery to ETA... return someEta; } }) .toList().subscribe(new Action1<List<ETA>>() { @Override public void call(List<ETA> etas) { } }); 

Of course, it would be nice to accept the Retrofit response (supposedly Observable<List<Delivery>> ?) And just watch each one. For this, we ideally use something like flatten() , which does not appear in RxJava in the near future .

To do this, you can do something like this (much better with lambdas). You would replace Observable.from(deliveries) in the above example with the following:

 apiService.getDeliveries().flatMap(new Func1<List<Delivery>, Observable<Delivery>>() { @Override public Observable<Delivery> call(List<Delivery> deliveries) { return Observable.from(deliveries); } }).flatMap(...)