Transfer Encoding chunked with okhttp only delivers full result

Question

I am trying to get some insights on a chunked endpoint and therefore planned to print what the server sends me chunk by chunk. I failed to do so so I wrote a test to see if OkHttp/Retrofit are working as I expect it.

The following test should deliver some chunks to the console but all I get is the full response.

I am a bit lost what I am missing to even make the MockWebServer of OkHttp3 sending me chunks.

I found this retrofit issue entry but the answer is a bit ambiguous for me: Chunked Transfer Encoding Response

class ChunkTest {
    @Rule
    @JvmField
    val rule = RxImmediateSchedulerRule() // custom rule to run Rx synchronously

    @Test
    fun `test Chunked Response`() {
        val mockWebServer = MockWebServer()
        mockWebServer.enqueue(getMockChunkedResponse())

        val retrofit = Retrofit.Builder()
                .baseUrl(mockWebServer.url("/"))
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .client(OkHttpClient.Builder().build())
                .build()
        val chunkedApi = retrofit.create(ChunkedApi::class.java)

        chunkedApi.getChunked()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                    System.out.println(it.string())
                }, {
                    System.out.println(it.message)
                })

        mockWebServer.shutdown()
    }

    private fun getMockChunkedResponse(): MockResponse {
        val mockResponse = MockResponse()
        mockResponse.setHeader("Transfer-Encoding", "chunked")
        mockResponse.setChunkedBody("THIS IS A CHUNKED RESPONSE!", 5)
        return mockResponse
    }
}

interface ChunkedApi {
    @Streaming
    @GET("/")
    fun getChunked(): Flowable<ResponseBody>
}

Test console output:

Nov 06, 2018 4:08:15 PM okhttp3.mockwebserver.MockWebServer$2 execute
INFO: MockWebServer[49293] starting to accept connections
Nov 06, 2018 4:08:15 PM okhttp3.mockwebserver.MockWebServer$3 processOneRequest
INFO: MockWebServer[49293] received request: GET / HTTP/1.1 and responded: HTTP/1.1 200 OK
THIS IS A CHUNKED RESPONSE!
Nov 06, 2018 4:08:15 PM okhttp3.mockwebserver.MockWebServer$2 acceptConnections
INFO: MockWebServer[49293] done accepting connections: Socket closed

I expected to be more like (body "cut" every 5 bytes):

Nov 06, 2018 4:08:15 PM okhttp3.mockwebserver.MockWebServer$2 execute
INFO: MockWebServer[49293] starting to accept connections
Nov 06, 2018 4:08:15 PM okhttp3.mockwebserver.MockWebServer$3 processOneRequest
INFO: MockWebServer[49293] received request: GET / HTTP/1.1 and responded: HTTP/1.1 200 OK
THIS
IS A 
CHUNKE
D RESPO
NSE!
Nov 06, 2018 4:08:15 PM okhttp3.mockwebserver.MockWebServer$2 acceptConnections
INFO: MockWebServer[49293] done accepting connections: Socket closed

Mesh

Answer 1

The OkHttp Mockserver does chunk the data, however it looks like the LoggingInterceptor waits until the whole chunks buffer is full then it displays it.

From this nice summary about HTTP streaming:

The use of Transfer-Encoding: chunked is what allows streaming within a single request or response. This means that the data is transmitted in a chunked manner, and does not impact the representation of the content.

With that in mind, we are dealing with 1 "request / response", which means we'll have to do our chunks retrieval before getting the entire response. Then pushing each chunk in our own buffer, all that on an OkHttp network interceptor.

Here is an example of said NetworkInterceptor:

class ChunksInterceptor: Interceptor {

    val Utf8Charset = Charset.forName ("UTF-8")

    override fun intercept (chain: Interceptor.Chain): Response {
        val originalResponse = chain.proceed (chain.request ())
        val responseBody = originalResponse.body ()
        val source = responseBody!!.source ()

        val buffer = Buffer () // We create our own Buffer

        // Returns true if there are no more bytes in this source
        while (!source.exhausted ()) {
            val readBytes = source.read (buffer, Long.MAX_VALUE) // We read the whole buffer
            val data = buffer.readString (Utf8Charset)

            println ("Read: $readBytes bytes")
            println ("Content: \n $data \n")
        }

        return originalResponse
    }
}

Then of course we register this Network Interceptor on the OkHttp client.

Windshield