Robolectric + rxJava + retrofit Second call throws java.io.InterruptedIOException

Question

I am developing and android app. I am using retrofit (with OkClient) for api requests and Robolectric for testing. My api looks like this:

@GET("/v1/book/{bookId}") Observable<Book> getBook(@Path("bookId") int bookId);

Just for Robolectric I am forcing api calls to be synchronous. The restAdapter builder looks like this:

RestAdapter.Builder builder = new RestAdapter.Builder().setEndpoint(environment.getServerEndpoint())
                        .setClient(new OkClient(client))
                        .setExecutors(new ImmediateExecutor(), null)
                        .setErrorHandler(new ErrorHandler())
                        .setRequestInterceptor(new RequestInterceptor() {
                            @Override
                            public void intercept(RequestFacade request) {
                                // Always ask for JSON data
                                request.addHeader("Accept", "application/json");
                                request.addHeader("Content-Type", "application/json");               
                            }
                        });

public class ImmediateExecutor implements Executor {

        @Override public void execute(Runnable command) {
            command.run();
        }
    }

I have a simple test that looks like this:

API.getBook(1).subscribe();
API.getBook(2).subscribe();

Restadapter is created with the builder, and the API object with it (restadapter.create(...)). I have omitted it as it is trivial.

The first one works without problems, but the second one that should be the same throws an exception:

java.io.InterruptedIOException
    at okio.Timeout.throwIfReached(Timeout.java:146)
    at okio.Okio$1.write(Okio.java:75)
    at okio.AsyncTimeout$1.write(AsyncTimeout.java:155)
    at okio.RealBufferedSink.flush(RealBufferedSink.java:201)
    at com.squareup.okhttp.internal.http.HttpConnection.flush(HttpConnection.java:140)
    at com.squareup.okhttp.internal.http.HttpTransport.finishRequest(HttpTransport.java:52)
    at com.squareup.okhttp.internal.http.HttpEngine.readNetworkResponse(HttpEngine.java:828)
    at com.squareup.okhttp.internal.http.HttpEngine.access$200(HttpEngine.java:95)
    at com.squareup.okhttp.internal.http.HttpEngine$NetworkInterceptorChain.proceed(HttpEngine.java:823)
    at com.carmacarpool.carmaridepool.rest.CarmaHttpClientFactory$NetworkLoggingInterceptor.intercept(CarmaHttpClientFactory.java:77)
    at com.squareup.okhttp.internal.http.HttpEngine$NetworkInterceptorChain.proceed(HttpEngine.java:803)
    at com.squareup.okhttp.internal.http.HttpEngine.readResponse(HttpEngine.java:684)
    at com.squareup.okhttp.Call.getResponse(Call.java:272)
    at com.squareup.okhttp.Call$ApplicationInterceptorChain.proceed(Call.java:228)
    at com.carmacarpool.carmaridepool.rest.CarmaHttpClientFactory$LoggingInterceptor.intercept(CarmaHttpClientFactory.java:53)
    at com.squareup.okhttp.Call$ApplicationInterceptorChain.proceed(Call.java:225)
    at com.squareup.okhttp.Call.getResponseWithInterceptorChain(Call.java:199)
    at com.squareup.okhttp.Call.execute(Call.java:79)
    at retrofit.client.OkClient.execute(OkClient.java:53)
    at retrofit.RestAdapter$RestHandler.invokeRequest(RestAdapter.java:326)
    at retrofit.RestAdapter$RestHandler.access$100(RestAdapter.java:220)
    at retrofit.RestAdapter$RestHandler$1.invoke(RestAdapter.java:265)
    at retrofit.RxSupport$2.run(RxSupport.java:55)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
    at java.util.concurrent.FutureTask.run(FutureTask.java:262)
    at com.carmacarpool.carmaridepool.testutils.ShadowUICarmaRESTClassFactory$ImmediateExecutor.execute(ShadowUICarmaRESTClassFactory.java:91)
    at retrofit.RxSupport$1.call(RxSupport.java:42)
    at retrofit.RxSupport$1.call(RxSupport.java:32)
    at rx.Observable.subscribe(Observable.java:7393)
    at rx.Observable.subscribe(Observable.java:7083)
    at com.carmacarpool.carmaridepool.CorridorTest.test(CorridorTest.java:99)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
    at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
    at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
    at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
    at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
    at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
    at org.robolectric.RobolectricTestRunner$2.evaluate(RobolectricTestRunner.java:265)
    at org.robolectric.RobolectricTestRunner.runChild(RobolectricTestRunner.java:205)
    at org.robolectric.RobolectricTestRunner.runChild(RobolectricTestRunner.java:54)
    at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
    at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
    at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
    at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
    at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
    at org.robolectric.RobolectricTestRunner$1.evaluate(RobolectricTestRunner.java:173)
    at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
    at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecuter.runTestClass(JUnitTestClassExecuter.java:86)
    at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecuter.execute(JUnitTestClassExecuter.java:49)
    at org.gradle.api.internal.tasks.testing.junit.JUnitTestClassProcessor.processTestClass(JUnitTestClassProcessor.java:64)
    at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:50)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.gradle.messaging.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
    at org.gradle.messaging.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
    at org.gradle.messaging.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:32)
    at org.gradle.messaging.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:93)
    at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:106)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.gradle.messaging.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
    at org.gradle.messaging.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
    at org.gradle.messaging.remote.internal.hub.MessageHub$Handler.run(MessageHub.java:360)
    at org.gradle.internal.concurrent.DefaultExecutorFactory$StoppableExecutorImpl$1.run(DefaultExecutorFactory.java:64)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

I have one network interceptor just with a log and it works ok. I have access to the server logs, and the request is not even reaching the server.

Does anybody have any clue about what could be the problem? It seems that for some unknown reason the thread is being killed?

Thanks.

EDIT: If in the onNext function (first parameter of subscribe), I execute the second request, then it works. Everything is synchronous as expected.

SOLUTION After many tries I could find a solution. The problem seems to be coming from Okio. Apparently there is a buffer that writes the response (or something like this, I solved it a few weeks ago and I don't remember 100%). This buffer gets closed in the middle of the second request and that's what is causing the error.

To fix this what I do is wrap the request in a try/catch block. If an IOException happens then I retry. I am retrying a maximum of 5 times (to avoid infinite loops).

The code looks like:

    try {
        // Perform the request
        return chain.proceed(request);
    } catch (IOException e) {

        // Retry again if we haven't tried at least REQUEST_RETRIES times
        if (iteration < REQUEST_RETRIES) {
            return performRequest(chain, ++iteration);
        }

        // Otherwise, save the exception and throw it later
        exception = e;
    }

The chain object comes from an OkHttpClient interceptor:

OkHttpClient client = new OkHttpClient(); client.interceptors().add(new CustomInterceptor());

private class CustomInterceptor implements Interceptor { @Override public Response intercept(Chain chain) throws IOException { ...

Hope this is helpful.

Answer 1

I think you're doing a little too much to get observables to run synchronously during testing. Instead all you need to do is during your test

 Book book = retrofitAPI.getBook(someId).toBlocking.first()

This would execute the observable synchronously. One of my favorite parts of the Rx library.