How to process data using RxJava in Android

Question

The goal is to make two types of requests using the REST library in an Android application, get two types of objects, and fill one with another.

You can, of course, gash asink task and consistently perform all operations. And you can try to use RXjava, but I’m familiar with it the second day, and I still don’t understand much, not to mention lambda expressions, articles on Habré are also far gone, almost nothing is clear.

Karoch. There are two classes 1. offerClass - each class when sparring json gets its id. contains an empty orderStatus class that can be retrieved by another request using id. 2. orderStatus - the class contains information for a specific offerClass id.

There is a variable offers = new ArrayList (); which when onCreate () is zero, and is given to the adapter to build the View.

In the onCreate () method, the downloadData () function is called which, using RXjava, downloads and fills the data, and returns it to the adapter in the main stream.

What I want to do:

1.выполнить асинхронный запрос getActiveSimpleOrder(1) 2.по нему получить List<offerClass> и присвоить его offers. 3. наполнить каждый объект offers (offer - статутом, зная id) for(offerClass offer:offers) // для каждого объекта offers { 4. getOrderStatus(offer.getId()) //Получив статус для конкретного объекта. и далее offer.setStatus(response.body()) // Наполняем объект статусом. } 5. когда все выполнится передать offers в основной поток и вызвать 6. adapter.notifyDataSetChanged().

After the answers a new code was generated:

In the main fragment method:

  public void onStart() { initializeData(); // Получаем данные }

Function code:

 private void initializeData() { int rest_id = 1; ApiInterface api = ApiModule.getClient(); api.getActiveSimpleOrder(rest_id) .subscribeOn(Schedulers.io()) .flatMap(Observable::from) .flatMap(offer -> api.getStatus(offer.getId()) .doOnNext(offer::setStatus) ******* .map(status -> offer) ******************* ) .toList() .observeOn(AndroidSchedulers.mainThread()) .subscribe(adapter::setOffers); }

Api code:

 public class ApiModule { private static ApiInterface apiInterface; private static String baseUrl = ""http://85.143.221.39/EatgidServer/"; public static ApiInterface getClient() { if (apiInterface == null) { Retrofit client = new Retrofit.Builder() .baseUrl(baseUrl) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .addConverterFactory(GsonConverterFactory.create()) .build(); apiInterface = client.create(ApiInterface.class); } return apiInterface; } }

The string with asterisks is underlined in red: the compiler compiles anyway. which is strange. varning is as follows:

The application works with the error:

LogCat:

  06-13 19:59:53.954 13429-13429/eatgid.com.restaurant E/AndroidRuntime: FATAL EXCEPTION: main Process: eatgid.com.restaurant, PID: 13429 java.lang.IllegalStateException: Exception thrown on Scheduler.Worker thread. Add `onError` handling. at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:60) at android.os.Handler.handleCallback(Handler.java:815) at android.os.Handler.dispatchMessage(Handler.java:104) at android.os.Looper.loop(Looper.java:194) at android.app.ActivityThread.main(ActivityThread.java:5631) at java.lang.reflect.Method.invoke(Native Method) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:959) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:754) Caused by: rx.exceptions.OnErrorNotImplementedException: 1465819531027 at rx.Observable$27.onError(Observable.java:8139) at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:157) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:254) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:186) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at android.os.Handler.handleCallback(Handler.java:815) at android.os.Handler.dispatchMessage(Handler.java:104) at android.os.Looper.loop(Looper.java:194) at android.app.ActivityThread.main(ActivityThread.java:5631) at java.lang.reflect.Method.invoke(Native Method) at java.lang.reflect.Method.invoke(Method.java:372) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:959) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:754) Caused by: com.google.gson.JsonSyntaxException: 1465819531027 at com.google.gson.internal.bind.DateTypeAdapter.deserializeToDate(DateTypeAdapter.java:74) at com.google.gson.internal.bind.DateTypeAdapter.read(DateTypeAdapter.java:59) at com.google.gson.internal.bind.DateTypeAdapter.read(DateTypeAdapter.java:41) at com.google.gson.internal.bind.ReflectiveTypeAdapterFactory$1.read(ReflectiveTypeAdapterFactory.java:116) at com.google.gson.internal.bind.ReflectiveTypeAdapterFactory$Adapter.read(ReflectiveTypeAdapterFactory.java:216) at retrofit2.converter.gson.GsonResponseBodyConverter.convert(GsonResponseBodyConverter.java:37) at retrofit2.converter.gson.GsonResponseBodyConverter.convert(GsonResponseBodyConverter.java:25) at retrofit2.ServiceMethod.toResponse(ServiceMethod.java:116) at retrofit2.OkHttpCall.parseResponse(OkHttpCall.java:211) at retrofit2.OkHttpCall.execute(OkHttpCall.java:174) at retrofit2.adapter.rxjava.RxJavaCallAdapterFactory$RequestArbiter.request(RxJavaCallAdapterFactory.java:160) at rx.Subscriber.setProducer(Subscriber.java:211) at rx.Subscriber.setProducer(Subscriber.java:205) at rx.Subscriber.setProducer(Subscriber.java:205) at rx.Subscriber.setProducer(Subscriber.java:205) at retrofit2.adapter.rxjava.RxJavaCallAdapterFactory$CallOnSubscribe.call(RxJavaCallAdapterFactory.java:141) at retrofit2.adapter.rxjava.RxJavaCallAdapterFactory$CallOnSubscribe.call(RxJavaCallAdapterFactory.java:127) at rx.Observable$2.call(Observable.java:162) at rx.Observable$2.call(Observable.java:154) at rx.Observable$2.call(Observable.java:162) at rx.Observable$2.call(Observable.java:154) at rx.Observable$2.call(Observable.java:162) at rx.Observable$2.call(Observable.java:154) at rx.Observable.unsafeSubscribe(Observable.java:8314) at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:235) at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:145) at rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54) at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:368) at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:330) at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:807) at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.slowpath(OnSubscribeFromIterable.java:97) at rx.internal.operators.OnSubscribeFromIterable$IterableProducer.request(OnSubscribeFromIterable.java:73) at rx.Subscriber.setProducer(Subscriber.java:211) at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:49) at rx.internal.operators.OnSubscribeFromIterable.call(OnSubscribeFromIterable.java:32) at rx.Observable.unsafeSubscribe(Observable.java:8314) at rx.internal.op

Here I wrote a working version very close in fact to ru.stackoverflow.com/questions/532152/…
Highlighting a line with asterisks as errors is a glitch of the analyzer in the IDE, it unfortunately sometimes mimics the lambda.
And the error of work you now have is not in the code, more precisely not in logic.
You have a non-existent address requested by HttpException: HTTP 400 Bad Request
I apologize for this mistake, I forgot to update it really, but the error persists, as I understand the problem in the parsin dates, because the status contains several date objects that are pulled out of the Status by the String getDate method.
either that the library cannot parse the date object, you can somehow handle onError so that the application does not crash, even add a toaster, and on OnComplited flop something, how to prescribe it?
if subscriber is main therad where should i override the above methods?
To handle the error, you can add the second parameter to the subscribe method: .subscribe(adapter::setOffers, throwable -> throwable.printStackTrace());
I am writing from memory so that the name of the printStackTrace method could not be written correctly.

Accepted Answer · 2016-06-09T22:37:08

It should be something like this:

 ApiInterface api = ApiModule.getClient(); // создадим только один сервис с апи, не надо дергать getClient() несколько раз api.getActiveSimpleOrder(1) // запрашиваем список .subscribeOn(Schedulers.io()) // запросы и дальнейшие преобразования будут выполняться в потоке для операций ввода/вывода .flatMap(Observable::from) // превращаем список в последовательность что бы работать дальше отдельно с каждым заказом .flatMap(offer -> api.getOrderStatus(offer.getId()) // запрашиваем статус .doOnNext(offer::setStatus) // записываем полученный статус в объект заказа .map(status -> offer) // небольшая хитрость что бы дальше шли объекты заказов а не статусов ) .toList() // преобразуем последовательность обратно в список .observeOn(AndroidSchedulers.mainThread()) // дальнейшие действия над данными будем производить в главном потоке .subscribe(adapter::setOffers); // подписываемся на получение списка инициируя тем самым старт запросов

The adapter should have a public void setOffers(List<offerClass> offers) method in which you should write the incoming list as you need and call adapter.notifyDataSetChanged() .

If the Observable::from syntax is not familiar to you, then this is just an abbreviation for offers -> Observable.from(offers) , and adapter::setOffers respectively, from offers -> adapter.setOffers(offers) .

He himself realized that the object from the list offers this offer? or should it be the name of the class, or is it the name of the result of working out api as it is at all?

In the line .flatMap(Observable::from) we say that further down the chain will go instead of the object of class List<offerClass> objects from this list and accordingly class offerClass . Then in the line .flatMap(offer -> api.getOrderStatus(offer.getId()) we write a lambda function that takes the offerClass class as a parameter (the compiler itself understands, and we can not specify the class ourselves), we set the name to this parameter I called it offer myself. This function will be applied to each offer from the received list. Then we create an observabl status request ( api.getOrderStatus(offer.getId()) ) in the body of the lambda function, in the next line ( .doOnNext(offer::setStatus) ) say that each received orderStatus we should send in the method offer.setStatus() then stitched. .map(status -> offer) we gnoriruem class status orderStatus so he does not need us to continue, and return to processing offer class offerClass for us in the end to get a list of these classes. Why do you have this line swear I do not understand, please bring the full name and the error message.

What exactly do we convert to toList?

.toList() is the inverse operation of .flatMap(Observable::from) , that is, this command waits until all the offerClass are processed and assembled into the List<offerClass> , on which we further subscribe.

why do we perform operations of Schedulers.io ()? it's safe? Are there other threads you can choose?

RxJava has the following scheduler options:

AndroidSchedulers.mainThread () - to perform operations in the main thread
Schedulers.io () - to perform long requests to resources (network requests, requests to the database, etc.) in the thread pool
Schedulers.computation () - for computational operations (for example, it is recommended to perform the map operator, but we will not complicate the code so far)

There are others, you can read more about them here in the section RxJava.

unless we need to make Observable :: from and the status object received by the api.getOrderStatus command (offer.getId ())

Not necessary. We already receive not a list, but immediately the status of this process in parts we have nothing there.

when trying to compile your code: java.lang.IllegalArgumentException: Unable to create an adapter for rx.Observable> for method ApiInterface.getActiveSimpleOrder

You have not connected CallAdapterFactory required to convert a Retrofit request to an Observable. It is necessary to add the code:

  Retrofit client = new Retrofit.Builder() .baseUrl(baseUrl) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) // без этого ретрофит не сможет возвращать Observable объекты .addConverterFactory(GsonConverterFactory.create(gson)) .build();

And if you have not connected this library yet, then connect it:

 compile 'com.squareup.retrofit:adapter-rxjava:{версия ретрофита}'

He himself realized that the object from the list offers this offer?
or should it be the name of the class, or is it the name of the result of working out api as it is at all?
2. complains about the string .map (status -> offer) something there about the lover and aper boundes.
unless we need to do Observable :: from and the status object received by the command api.getOrderStatus (offer.getId ())
what kind of .map trick (status -> offer) what does this command actually do?
Why, for example, do not have a status counter equal to the orders. Size and do not respond when the last one is received?
when trying to compile your code: java.lang.IllegalArgumentException: Unable to create a call adapter for rx.Observable <java.util.List <eatgid.com.restaurant.adapter.offerClass >> for method ApiInterface.getActiveSimpleOrder

How to process data using RxJava in Android

1 answer 1

More articles: