android rxjava repeating a request

Question

I am trying to achieve the following. I load a list of objects I want to get values to put in a list later.

First I gather all the values into an array (to mountain order) using flatmap and then when everything is done I populate an adapter.

The thing I am unable to do is to repeat the operation ever xxx seconds. I understand its done using an interval. Still I get no result at all, or only none repeating one result.

Here’s my code:

  Observable.fromIterable(URLList)
            .concatMap(url -> standartRequest(App.getInstance().getApi().getService().getData(currency.getUrl())))
            .retry(Constants.RETRY_COUNT)
            .timeout(Constants.TIMEOUT_IN_SECONDS, TimeUnit.SECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::success, this::error, this::valuesRetrieved); 



    recyclerView = ((CurrencyListFragment) controller).getRecyclerView();
    LinearLayoutManager linearManager = new LinearLayoutManager(controller.getContext());
    recyclerView.setLayoutManager(linearManager);
}


private void valuesRetrieved() {
    listAdapter adapter = new listAdapter(valuesFromResponse);
    recyclerView.setAdapter(adapter);
}

private void success(Object response) {
    valuesFromResponse.add(response);
}

Where do I put

.interval(5, TimeUnit.SECONDS).timeInterval()

Solution 1

Well actually, you do not put interval anywhere, for repeating the operation every x interval, you should use repeat operator variant called repeatWhen where you can provide your interval logic in this way:

.repeatWhen(completed -> completed.delay(5, TimeUnit.SECONDS))

repeatWhen() will hand to you an Observable that transform your source Observable onCompleted() events as onNext() (with void), you should return Observable that emits onNext() which signals to resubscribe to your source Observable - meaning repeat the operation. While onCompleted()/onError() will be delivered as onCompleted()/onError() on your source Observable. recommended reading regarding repeatWhen/retryWhen.

One thing to consider, as repeatWhen() will basically swallows all your onCompleted events (as you're repeating the operation there is no onCompleted(), your Observable will not stop from by itself!), then you should gather and update the adapter differently, I guess you can simply use toList() to gather all items to single onNext() (a replacement to your success() logic) and then on each onNext updates the list (what you're doing on onCompleted right now), to sum it up:

 Observable.fromIterable(URLList)
            .concatMap(url -> standartRequest(App.getInstance().getApi().getService().getData(currency.getUrl())))
            .retry(Constants.RETRY_COUNT)
            .timeout(Constants.TIMEOUT_IN_SECONDS, TimeUnit.SECONDS)
            .toList()
            .repeatWhen(completed -> completed.delay(5, TimeUnit.SECONDS))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::valuesRetrieved, this::error);

EDIT:

Your timeout and retry logic are applied to the entire operation chain, so if all the network requests together take more than Constants.TIMEOUT_IN_SECONDS you will get timeout exception, you probably just want to retry and time out each individual request. like this:

Observable.fromIterable(URLList)
            .concatMap(url ->  standartRequest(App.getInstance()
                    .getApi().getService().getData(currency.getUrl())
                    .retry(Constants.RETRY_COUNT)
                    .timeout(Constants.TIMEOUT_IN_SECONDS, TimeUnit.SECONDS)))                
            .toList()
            .repeatWhen(completed -> completed.delay(5, TimeUnit.SECONDS))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(this::valuesRetrieved, this::error);

Solution 2

I am repeating my retrofit call every 2 second after it is completed

//Retrofit Builder
        val retrofitBuilder = Retrofit.Builder()
            .addConverterFactory(GsonConverterFactory.create())
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
            .baseUrl("http://worldtimeapi.org/")
            .build()

        val timeApi = retrofitBuilder.create(TimeApi::class.java)


    val timeObservable = timeApi.getTime()

    timeObservable.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .repeatWhen { completed -> completed.delay(2, TimeUnit.SECONDS) }
        .subscribe(object : Observer<Time> {
            override fun onComplete() {
                Log.e("MainActivity", "It is completed")
            }

            override fun onSubscribe(d: Disposable) {
                Log.e("MainActivity", "you have successfully subscribed ")
            }

            override fun onNext(t: Time) {
                progress.visibility = View.INVISIBLE
                txtTime.text = t.unixtime
                Log.e("MainActivity", "OnNext Called" + t.unixtime)

            }

            override fun onError(e: Throwable) {
                Log.e("MainActivity", "ERROR")
            }
        })

}

See the Log Cat , onNext is called after every 2 second.

Comments

[Admin]

Dear @yosriz, that code doesn't work because repeatWhen is triggered each time a query returns a result, making it retry all the queries from the start. Mb cose of the concatMap?

[Admin]

p.s. now I get a java.util.concurrent.TimeoutException :)

[yosriz]

oh, that's because the retry and timeout, your'e time outing the entire operations (all the requests) so that's probably takes more than Constants.TIMEOUT_IN_SECONDS, you probably want this time out to each request not the whole operation

[yosriz]

p.s. why your'e doing concat? don't you want parallelism?

[Admin]

I want to maintain item order in the list without sorting. concat does that.

[yosriz]

Yes indeed, you can also use concatEager to have parallelism while maintaining items order

[Admin]

Thank you! I didn't know that. I still cannot achieve a simple repeat( my guess is I am confusing the order of the operations. Getting basically the same result for an interval and repeat when (going into a loop that adds another loop which adds another loop, and so on and forth)

[yosriz]

please accept the answer if that was helpful, thanks!