Tengo una siguiente clase:

public class SessionStore {
    Subject<Session, Session> subject;

    public SessionStore() {
       subject = new SerializedSubject<>(BehaviorSubject.create(new Session());
    }

    public void set(Session session) {
        subject.onNext(session);
    }

    public Observable<UserSession> observe() {
        return subject.distinctUntilChanged();
    }
}

En la actividad, observo la sesión y realizo la operación de red en cada cambio:

private Subscription init() {
    return sessionStore
            .observe()
            .flatMap(new Func1<Session, Observable<Object>>() {
                @Override
                public Observable<Object> call(Session session) {
                    return (session.isValid() 
                       ? retrofitService.getThingForValid()
                       : retrofitService.getThingForInalid())
                       .subscribeOn(Schedulers.io());
                }
            })
            .subscribe(...);
}

Ahora tengo un interceptor de solicitud Okhttp, en el que configuro la instancia de sesión de válida a no válida cuando la respuesta de la red no es un código 200.

Esto es lo que pasa:

En la suscripción inicial a la sesión, almacene elgetThingForValid() se ejecuta y falla.OkHttp intercepta el error y establece una nueva sesión.El almacén de sesión emite una nueva sesión, ahora no válida.La nueva emisión ejecuta ungetThingForInvalid() método.

Lo importante es saber que esta ejecución ocurre en medio de la llamada de Retrofit anterior. Esto se debe a que el cliente OkHttp está envuelto por Retrofit y todos los interceptores se ejecutan antes de que Retrofit regrese.

Teniendo esto en cuenta, te das cuenta de que la segunda llamada ya está siendo ejecutada y procesada por Retrofit, mientras que la primera aún no ha terminado.

Cuando finaliza la primera llamada, arroja HttpException porque la respuesta fue código no 200.El xception mata el flujo de rx y con él la segunda llamada.

He tratado de ignorar esta excepción en la transmisión, pero la segunda llamada es cancelada por Retrofit de todos modos.

¿Tienes alguna idea de cómo hacer que mi concepto funcione, por favor?