Recipe for tee

[digitalbuddha]

https://twitter.com/jessewilson/status/692326663061966848

Original issue:

I'm struggling with reading from a BufferedSource twice. What I'm trying to do is take an okhttp response body and stream it to both a parser and a file. First I created 2 Bufferinstances from same BufferedSource next I wrap a Buffer in an InputStreamReader and gracefully (or so I thought) pass it to gson. What I get is MalformedJsonException

If I however simply do body.source().readUtf8Line() I get the full raw json.

full example:

        Request request = new Request.Builder()
                .url("https://www.reddit.com/r/aww/new/.json")
                .build();
        response =new OkHttpClient().newCall(request).execute();

                        ResponseBody body = response.body();
                        BufferedSource sourceforFile = body.source();
                        sourceforFile.require(8014);
                        Buffer buffer = sourceforFile.buffer();
                        Buffer sourceForParser = buffer.clone();

                        InputStreamReader inputStreamReader = new InputStreamReader(sourceForParser.inputStream());
                        result= new GsonBuilder()
                                .registerTypeAdapterFactory(new GsonAdaptersModel()).create()
                                .fromJson(inputStreamReader, RedditData.class);

                        File file = new File(getContext().getCacheDir(), "file");
                        BufferedSink sink = Okio.buffer(Okio.sink(file));
                        sink.writeAll(sourceforFile.buffer());
                        diskValue = Okio.buffer(Okio.source(file)).readString(Charset.defaultCharset());

It seems like Buffers are perfectly suited for taking a single source and streaming in multiple direction, would love to finally bring it all together. Thank you for your help.

[digitalbuddha]

I rearranged to instead use the original BufferedSource to create the InputStream and the parsing step works, but the copied Buffer still only gives me a partial data saved to disk. Do I need to synchronize or wait for the originalSource to be exhausted prior to using a cloned Buffer?

ResponseBody body = response.body();
BufferedSource originalSource = body.source();
originalSource.require(8000);
Buffer copiedSource = originalSource.buffer().clone();

InputStreamReader inputStreamReader = new InputStreamReader(originalSource.inputStream());
result= new GsonBuilder()
        .fromJson(inputStreamReader, RedditData.class);

File file = new File(getContext().getCacheDir(), "file");
BufferedSink sink = Okio.buffer(Okio.sink(file));
sink.writeAll(copiedSource);
sink.close();
diskValue = Okio.buffer(Okio.source(file)).readString(Charset.defaultCharset());

[JakeWharton]

You can use Request.peekBody(Long.MAX_VALUE) to get a copy of the RequestBody, but this has a HUGE downside: it buffers the entire response in memory.

What you want is to write the stream to another sink while the application pulls bytes off the wire. OkHttp's cache behaves like this, and you should copy what it does. I can grab a pointer in the source if you can't find it when I get home.

[digitalbuddha]

Yeah definitely don't want to copy. A pointer would be appreciated.

[JakeWharton]

https://github.com/square/okhttp/blob/master/okhttp/src/main/java/okhttp3/internal/http/HttpEngine.java#L756-L803

cacheBody would be Okio.buffer(Okio.source(targetFile)) and everything else should be straightforward to modify.

[digitalbuddha]

thanks! I won't tell you how many times I read Cache.java today trying to figure it out

[digitalbuddha]

After 2 tweets and 2 issue comments I finally got it. Thanks @JakeWharton & @swankjesse in case anyone else needs a working example here's the relevant parts. https://gist.github.com/digitalbuddha/3c5bb15fa12a553c85ec

[JakeWharton]

By the way you can use ResponseBody.charStream() to get a Reader to pass to Gson that will use the appropriate charset defined in the Content-Type header automatically.

[digitalbuddha]

Reopening in preparation of pr.

Going to add @NonNull Source filter(final Source source, final BufferedSink cacheBody) to Okio.java. Any suggestions for method name, feels like a source filter to me

[swankjesse]

Tee ? https://en.wikipedia.org/wiki/Tee_%28command%29

[digitalbuddha]

Wasn't sure if that was a placeholder term first time you suggested it. Thanks. On Jan 27, 2016 10:42 AM, "Jesse Wilson" [email protected] wrote:

Tee ? https://en.wikipedia.org/wiki/Tee_%28command%29

— Reply to this email directly or view it on GitHub https://github.com/square/okio/issues/186#issuecomment-175694189.

[devcsrj]

bump

Hi! Is a recipe now documented somewhere for this (i.e.: single producer, multiple consumer)?

[swankjesse]

No recipe. The closest we have is this: https://github.com/square/okhttp/blob/master/okhttp/src/main/java/okhttp3/internal/cache2/Relay.java