Get JSON object one by one from JSON array in Android using Retrofit and RxJava

Question

I am using retrofit to hit my network api which returns json array. I am doing so using following code-

Observable<MyJson[]> response = (Observable<MyJson[]>)mNetworkService.getReadyObserverable(mNetworkService.getNetworkServiceApi().getMyDataJsons(), MyJson.class, true, useCache);
        mAirlineSubscription = response.subscribe(new Observer<MyJson[]>() {
            @Override
            public void onCompleted() {
                Log.d(TAG, "getData completed..");
            }

            @Override
            public void onError(Throwable e) {
                Log.e(TAG, "onError: "  + e.getLocalizedMessage());
            }

            @Override
            public void onNext(MyJson[] myJsonData) {
                //I want here json data one by one
            }

But my problem is json data array get downloaded completely and then only onNext gets called. Here I want onNext should get called when retrofit downloads first json object from myData[] json array and keep continue until my all json objects of myData[] get downloaded. This way my UI would look more responsive and better in terms of user interaction.

Can anyone help to fix this?

Answer 1

i have same question i explain it base of own project, you should use my retrofit request is

@GET("StudentApi/getAll")
    Observable<List<Person>> getPersons();

and i have an interface to run server request and it has this method for run getPersons() and return list of json array of persons info

Observable<Person> getAllPersons();

and the important part is body of up method and should like bellow

@Override
    public Observable<Person> getAllPersons() {
        Observable<List<Person>> observable =
                serviceGenerator.getService().getPersons();
        return observable
                .flatMap(new Function<List<Person>, Observable<Person>>() {
                    @Override
                    public Observable<Person> apply(List<Person> persons) throws Exception {
                        return Observable.fromIterable(persons);
                    }
                });
}

i found that fromIterable method return objects of json array one by one and finally in an activity i get json objects from up methods like bellow

public void getPersons(){
      personsRepository.getAllPersons()
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeOn(Schedulers.io())
                .subscribeWith(new DisposableObserver<Person>() {
                    @Override
                    public void onNext(Person value) {
                        Log.e("person info  is",value.getpName() + value.getpFamily());
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        Log.e("onError",throwable.toString());
                    }

                    @Override
                    public void onComplete() {
                        Log.e("onComplete","onComplete");
                    }
                });
    }

i hope is useful :)

Answer 2

Receiving elements One by One, as fast as possible?
Don't want to wait when everything will be downloaded?

The solution is: Declare the method of retrofit's interface with annotation @Streaming and use Observable<ResponseBody> as return value. And then, by using flatMap(), convert ResponseBody to series of POJO (Observable<TYPE>).

Example:

1. Declare retrofit interface:

   public interface HugeJsonApi {
   
   String SERVICE_ENDPOINT = "https://raw.githubusercontent.com";
   
   @Streaming
   @GET("/zemirco/sf-city-lots-json/master/citylots.json")
   Observable<ResponseBody> get();
   }

2. Use it like that:

   public void playHugeJsonSample() {
   
       HugeJsonApi hugeJsonApi = RestUtils.createService(HugeJsonApi.class, HugeJsonApi.SERVICE_ENDPOINT);
   
       Handler handler = new Handler(Looper.getMainLooper());
       hugeJsonApi.get()
               .flatMap(responseBody -> convertObjectsStream(responseBody, gson, Feature.class))
               .subscribeOn(Schedulers.io())
               .observeOn(AndroidSchedulers.mainThread())
               .subscribe(new Subscriber<Feature>() {
   
                   @Override
                   public void onStart() {
                       super.onStart();
                       request(1);
                   }
   
                   @Override
                   public void onNext(Feature feature) {
                       Log.i(TAG, gson.toJson(feature));
                       counter[0]++;
                       request(1);
                   }
   
                   @Override
                   public void onCompleted() {
                       Log.i(TAG, "onCompleted() called. Fetched elements:" + counter[0]);
                   }
   
                   @Override
                   public void onError(Throwable e) {
                       Log.e(TAG, "something went wrong", e);
                   }
               });
   }
   
   @NonNull
   private static <TYPE> Observable<TYPE> convertObjectsStream(ResponseBody responseBody, Gson gson, Class<TYPE> clazz) {
       Type type = TypeToken.get(clazz).getType();
       return Observable.create(SyncOnSubscribe.<JsonReader, TYPE>createStateful(
               // initialize the reader
               () -> {
                   try {
                       JsonReader reader = gson.newJsonReader(responseBody.charStream());
                       reader.beginObject();
                       return reader;
                   } catch (IOException e) {
                       e.printStackTrace();
                       RxJavaHooks.onError(e);
                   }
                   return null;
               },
               // read elements one by one
               (reader, observer) -> {
   
                   if (reader == null) {
                       observer.onCompleted();
                       return null;
                   }
   
                   try {
                       if (reader.hasNext()) {
                           TYPE t = gson.fromJson(reader, type);
                           observer.onNext(t);
                       }
                       else {
                           observer.onCompleted();
                       }
   
                   } catch (IOException e) {
                       e.printStackTrace();
                       observer.onError(e);
                   }
   
                   return reader;
               },
               // close the reader
               reader -> {
                   if (reader != null) {
                       try {
                           reader.close();
                       } catch (Exception e) {
                           e.printStackTrace();
                           RxJavaHooks.onError(e);
                       }
                   }
               }
   
       ));
   }

Here is workable example:

https://github.com/allco/RetrofitAndRxJava

It takes 180+Mb Json and parses it as a real stream.

Answer 3

your service should look like this:

@GET("users/{username}/repos")
Observable<List<Repository>> publicRepositories(@Path("username") String username);

I mean, return a Observable<List<Something>>