Android: server polling with retrofit

Question

I am creating a 2-player game on Android. The game works in turn, so player 1 waits until player 2 makes his entrance, and vice versa. I have a web server where I run the API with Slim Framework. On clients I use Retrofit. So for clients, I would like to poll my web server (I know that this is not the best approach) every X seconds to check if there was an input from player 2 or not, if so, change the interface (game plan).

Retrofitting I came across RxJava. My problem is to find out if I need to use RxJava or not? If so, are there really simple examples for polling with modification? (Since I only send a pair of key / value pairs) And if not, how can this be done with an upgrade?

I found this thread, but it didn’t help me either, because I still don’t know if I need Retrofit + RxJava at all, maybe ways are easier?

Answer 1

Say the interface you defined for Retrofit contains a method similar to this:

public Observable<GameState> loadGameState(@Query("id") String gameId); 

Retrofitting methods can be defined in one of three ways:

1.) simple synchronous:

 public GameState loadGameState(@Query("id") String gameId); 

2.) that accept a Callback for asynchronous processing:

 public void loadGameState(@Query("id") String gameId, Callback<GameState> callback); 

3.) and the one returned by rxjava Observable , see above. I think that if you are going to use Retrofit in combination with rxjava, it makes sense to use this version.

Thus, you can simply use Observable for a single query, directly as follows:

 mApiService.loadGameState(mGameId) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<GameState>() { @Override public void onNext(GameState gameState) { // use the current game state here } // onError and onCompleted are also here }); 

If you want to re-interrogate the server, you can provide a "pulse" using version timer() or interval() :

 Observable.timer(0, 2000, TimeUnit.MILLISECONDS) .flatMap(mApiService.loadGameState(mGameId)) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<GameState>() { @Override public void onNext(GameState gameState) { // use the current game state here } // onError and onCompleted are also here }). 

It is important to note that flatMap used instead of map , because the return value of loadGameState(mGameId) itself an Observable .

But the version that you use in your update should also work:

 Observable.interval(2, TimeUnit.SECONDS, Schedulers.io()) .map(tick -> Api.ReceiveGameTurn()) .doOnError(err -> Log.e("Polling", "Error retrieving messages" + err)) .retry() .observeOn(AndroidSchedulers.mainThread()) .subscribe(sub); 

That is, if ReceiveGameTurn() is defined synchronously, like mine 1.) above, you should use map instead of flatMap .

In both cases, the onNext your Subscriber will be called every two seconds with the last game status from the server. You can process them one by one to limit the radiation to one element by inserting take(1) to subscribe() .

However, with regard to the first version: a single network error will be delivered to onError , and then the Observable will stop emitting any elements, which will make your subscriber useless and without input (remember, onError can be called once). To get around this, you can use any of the onError* rxjava methods to “redirect” the onNext failure.

For instance:

 Observable.timer(0, 2000, TimeUnit.MILLISECONDS) .flatMap(new Func1<Long, Observable<GameState>>(){ @Override public Observable<GameState> call(Long tick) { return mApiService.loadGameState(mGameId) .doOnError(err -> Log.e("Polling", "Error retrieving messages" + err)) .onErrorResumeNext(new Func1<Throwable, Observable<GameState>(){ @Override public Observable<GameState> call(Throwable throwable) { return Observable.emtpy()); } }); } }) .filter(/* check if it is a valid new game state */) .take(1) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<GameState>() { @Override public void onNext(GameState gameState) { // use the current game state here } // onError and onCompleted are also here }). 

This will be every two seconds: * use Retrofit to get the current state of the game from the server * filter out invalid * take the first valid * and unsubscribe

In case of error: * it will display an error message in doOnNext * and otherwise ignore the error: onErrorResumeNext will "consume" onError -Event (i.e. your Subscriber onError will not be called) and replace it with nothing ( Observable.empty() )

And as for the second version: in the event of a network error, retry will immediately return to the interval - and since interval emits the first integer immediately after the subscription, the next request is also sent immediately - and not after 3 seconds, as you probably want ...

Final note. In addition, if your game state is quite large, you can also first poll the server to ask if a new state is available, and only if the answer is yes, restart the new state of the game.

If you need more complex examples, please ask.

UPDATE I rewrote portions of this post and added additional information between them.

UPDATE 2 . I have added a complete error handling example using onErrorResumeNext .

Answer 2

Thank you, I finally did it in a similar way, based on the post I mentioned in my question. Here is my code:

 Subscriber sub = new Subscriber<Long>() { @Override public void onNext(Long _EmittedNumber) { GameTurn Turn = Api.ReceiveGameTurn(mGameInfo.GetGameID(), mGameInfo.GetPlayerOneID()); Log.d("Polling", "onNext: GameID - " + Turn.GetGameID()); } @Override public void onCompleted() { Log.d("Polling", "Completed!"); } @Override public void onError(Throwable e) { Log.d("Polling", "Error: " + e); } }; Observable.interval(3, TimeUnit.SECONDS, Schedulers.io()) // .map(tick -> Api.ReceiveGameTurn()) // .doOnError(err -> Log.e("Polling", "Error retrieving messages" + err)) .retry() .subscribe(sub); 

Now the problem is that I need to stop emitting when getting a positive response (a GameTurn ). I read about the takeUntil method, where I would need to pass another Observable that would produce something that would cause my poll to stop. But I'm not sure how to implement this. According to your decision, your API method returns Observable , as shown on the Retrofit website. Maybe this is a solution? So how will this work?

UPDATE: I reviewed @ david.miholas tips and tried his suggestion with repetition and filtering. Below you can find the code to initialize the game. The poll should work the same way: Player1 launches a new game → polls for the opponent, Player2 joins the game → the server sends the opponent’s identifier Player1 → completion of the poll.

  Subscriber sub = new Subscriber<String>() { @Override public void onNext(String _SearchOpponentResult) {} @Override public void onCompleted() { Log.d("Polling", "Completed!"); } @Override public void onError(Throwable e) { Log.d("Polling", "Error: " + e); } }; Observable.interval(3, TimeUnit.SECONDS, Schedulers.io()) .map(tick -> mApiService.SearchForOpponent(mGameInfo.GetGameID())) .doOnError(err -> Log.e("Polling", "Error retrieving messages: " + err)) .retry() .filter(new Func1<String, Boolean>() { @Override public Boolean call(String _SearchOpponentResult) { Boolean OpponentExists; if (_SearchOpponentResult != "0") { Log.e("Polling", "Filter " + _SearchOpponentResult); OpponentExists = true; } else { OpponentExists = false; } return OpponentExists; } }) .take(1) .subscribe(sub); 

The outlier is correct, however I get this log message on every emission:

 E/Polling﹕ Error retrieving messages: java.lang.NullPointerException 

Each emission triggers a doOnError . I usually got some Retrofit debug logs on every emit, which means that mApiService.SearchForOpponent will not be called. What am I doing wrong?