title: Forcing request retry after custom API exceptions in RxJava
body: <p>I'm really new in RxJava, but I'm trying to implement an API using Retrofit framework and RxJava.
  On a server side there is an authorization service which handles user's session and in case of some delay in user's actions server breaks his session. After that user has to login again in order to perform new API call. Bad thing is - server always returns HTTP code 200 and for notification about expiration uses some custom JSON response with expiration code, so RxJava doesn't fire Exception during onNext operation because RxJava considers that request was passed successfully.</p>
  
  <p>And the question is: How to implement correct flow to handle custom API exceptions like expiration and retry failed request after some other request (in my case relogin)?</p>
  
  <p>Something like this:</p>
  
  <ul>
  <li>app -> login()</li>
  <li>server -> { code:0, ... }</li>
  <li>app -> getUsers()</li>
  <li>server -> { code:0, ... }</li>
  <li>------- in 30 minutes -------</li>
  <li>app -> getPicture()</li>
  <li>server -> { code:99, ... } // session expired, user unauthorized</li>
  <li>app -> login()</li>
  <li>server -> { code:0, ... }</li>
  <li>app -> getPicture()</li>
  <li>server -> { code:0, ... }</li>
  </ul>
  
  <p>I was thinking about something like this, but with no success:</p>
  
  <pre><code>        Observable.create(new Observable.OnSubscribe&lt;BackendResponse&lt;String&gt;&gt;() {
          @Override
          public void call(Subscriber&lt;? super Response&lt;String&gt;&gt; subscriber) {
              try {
                  Response&lt;String&gt; response;
                  subscriber.onNext(response = getInterface().getUsers());
  
                  if (response != null) {
                    response.checkData(); // throws ServerException in case of code != 0
                  }
                  subscriber.onCompleted();
              } catch (Exception e) {
                  subscriber.onError(e);
              }
          }
      }).subscribeOn(Schedulers.io()).retryWhen(new RetryWithSessionRefresh(new SessionService())).subscribe();
  </code></pre>
  
  <p>and RetryWithSessionRefresh is:</p>
  
  <pre><code>public class RetryWithSessionRefresh implements
      Func1&lt;Observable&lt;? extends Notification&lt;?&gt;&gt;, Observable&lt;?&gt;&gt; {
  
  private final SessionService sessionSerivce;
  
  public RetryWithSessionRefresh(SessionService sessionSerivce) {
      this.sessionSerivce = sessionSerivce;
  }
  
  @Override
  public Observable&lt;?&gt; call(Observable&lt;? extends Notification&lt;?&gt;&gt; attempts) {
      return attempts
              .flatMap(new Func1&lt;Notification&lt;?&gt;, Observable&lt;?&gt;&gt;() {
                  @Override
                  public Observable&lt;?&gt; call(final Notification notification) {
                      final Throwable throwable = notification.getThrowable();
                      if (throwable instanceof ServerException) {
                          final ServerException backendException = (ServerException) throwable;
                          if (backendException.getBackendErrorCode() == Response.AUTHORIZATION_FAILED) {
                              return sessionSerivce
                                      .observeSessionToken()
                                      .doOnNext(new Action1&lt;TokenCallback&gt;() {
                                          @Override
                                          public void call(TokenCallback token) {
                                              if (token != null) {
                                                  DataHolder.getInstance().setAuthToken(token.getToken());
                                              }
                                          }
                                      })
                                      .doOnError(new Action1&lt;Throwable&gt;() {
                                          @Override
                                          public void call(Throwable throwable) {
                                              DataHolder.getInstance().setAuthToken("");
                                          }
                                      });
                          }
                      }
                      return Observable.error(notification.getThrowable());
                  }
              });
  }
  </code></pre>