Upgrade to Pro — share decks privately, control downloads, hide ads and more …

STEP BY STEP↑↑ - Okio & OkHttp

Shohei Kawano
February 17, 2018

STEP BY STEP↑↑ - Okio & OkHttp

At DroidKaigi Reject Conference, I have talked about Okio, specifically the internal implementation of Okio, with @stsn_jp talking about the internal implementation of OkHttp. To describe it (maybe too) simply, Okio is a library for I/O and OkHttp is a library for HTTP+HTTP/2 client.

Both libraries are for Java and Android and applications, and both are the ones of the most famous libraries, especially in Android development.

* square/okio: A modern I/O API for Java
https://github.com/square/okio

* square/okhttp: An HTTP+HTTP/2 client for Android and Java applications.
https://github.com/square/okhttp

* Droidcon Montreal Jake Wharton - A Few Ok Libraries - YouTube
https://www.youtube.com/watch?v=WvyScM_S88c

* DroidKaigi 2018 Reject Conference - connpass
https://connpass.com/event/73993/

Shohei Kawano

February 17, 2018
Tweet

More Decks by Shohei Kawano

Other Decks in Technology

Transcript

  1. private static final int MAX_SKIP_BUFFER_SIZE = 2048; public abstract int

    read() throws IOException public int read(byte b[]) throws IOException public int read(byte b[], int off, int len) throws IOException public long skip(long n) throws IOException public int available() throws IOException public void close() throws IOException public synchronized void mark(int readlimit) public synchronized void reset() throws IOException public boolean markSupported() InputStream.java
  2. APIの使い勝手 ▪ 用意されているAPIの中に、バッファ処理に関するものまで入っている (本来、 BufferedInputStream内に用意されているべき APIなど) ▪ 複数のread メソッド。独自拡張する場合には両方 overrideして処理を書く必要がある

    ▪ byte[]を渡す、どれくらい読み込んで、末尾までどれくらいか、自前で whileで都度チェック パフォーマンスへの意識 ▪ 渡したbyte[]が読み込んでいるデータ量より小さい場合は別途 byte[]を作成 ▪ 必要に応じてbyte[]のコピー処理 ▪ 不要になったbyte[]に対してGCが走ってしまう? InputStream
  3. public interface Source extends Closeable { long read(Buffer sink, long

    byteCount) throws IOException; Timeout timeout(); @Override void close() throws IOException; } Source.java(InputStreamの補完)
  4. Okio 概要 Okio.sink() Okio.source() といったstaticメソッドがある: • Source: InputStreamの補完 - Source

    of Bytes 読み込み先を指定 • Sink: OutputStreamの補完 - Sink for Bytes 書き込み先を指定 一般的なアプリで利用する場合にはBufferを利用する: • Okio.buffer(Source source) => return BufferedSource • Okio.buffer(Sink sink) => return BufferedSink
  5. Okio 概要 Okio.sink() Okio.source() といったstaticメソッドがある: • Source: InputStreamの補完 - Source

    of Bytes 読み込み先を指定 • Sink: OutputStreamの補完 - Sink for Bytes 書き込み先を指定 一般的なアプリで利用する場合にはBufferを利用する: • Okio.buffer(Source source) => return BufferedSource • Okio.buffer(Sink sink) => return BufferedSink
  6. public interface Source extends Closeable { long read(Buffer sink, long

    byteCount) throws IOException; Timeout timeout(); @Override void close() throws IOException; } Source.java
  7. Buffer • メモリ上のバイトの集まり(=バッファ)を表すクラス • BufferedSource, BufferedSink両方を実装(データの読み書きを行なう) • 読み書きしたデータをSegmentクラス(Bufferの一部)のbyte[] data内に保存 Segment

    • バッファの断片を表すクラス • byte[] dataを保持している(MAX: 8KB) • 自分の前(prev)のSegment, 次(next)のSegmentを知っている • SegmentPoolによって生成・最大限プールされる
  8. public static Source source(Socket socket) throws IOException { if (socket

    == null) throw new IllegalArgumentException("socket == null"); AsyncTimeout timeout = timeout(socket); Source source = source(socket.getInputStream(), timeout); return timeout.source(source); } Okio.source(Socket socket)
  9. Example: How Timeout is set? - OkHttpClient.Builder @Provides fun provideOkHttpClientBuilder(cache:

    Cache): OkHttpClient.Builder = OkHttpClient.Builder().cache(cache) .connectTimeout(10L, TimeUnit.SECONDS) .writeTimeout(10L, TimeUnit.SECONDS) .readTimeout(30L, TimeUnit.SECONDS)
  10. Example: How Timeout is set? - OkHttpCall private okhttp3.Call createRawCall()

    throws IOException { Request request = serviceMethod.toRequest(args); okhttp3.Call call = serviceMethod.callFactory.newCall(request); if (call == null) { throw new NullPointerException("Call.Factory returned null."); } return call; }
  11. Example: How Timeout is set? - OkHttpCall private okhttp3.Call createRawCall()

    throws IOException { Request request = serviceMethod.toRequest(args); okhttp3.Call call = serviceMethod.callFactory.newCall(request); if (call == null) { throw new NullPointerException("Call.Factory returned null."); } return call; }
  12. Example: How Timeout is set? - OkHttpClient @Override public Call

    newCall(Request request) { return RealCall.newRealCall(this, request, false /* for web socket */); }
  13. Example: How Timeout is set? - OkHttpClient @Override public Call

    newCall(Request request) { return RealCall.newRealCall(this, request, false /* for web socket */); } static RealCall newRealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) { // Safely publish the Call instance to the EventListener. RealCall call = new RealCall(client, originalRequest, forWebSocket); call.eventListener = client.eventListenerFactory().create(call); return call; }
  14. Example: How Timeout is set? - OkHttpClient @Override public Call

    newCall(Request request) { return RealCall.newRealCall(this, request, false /* for web socket */); } static RealCall newRealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) { // Safely publish the Call instance to the EventListener. RealCall call = new RealCall(client, originalRequest, forWebSocket); call.eventListener = client.eventListenerFactory().create(call); return call; }
  15. Example: How Timeout is set? - RealCall -> Chain final

    class RealCall implements Call { final OkHttpClient client; … Response getResponseWithInterceptorChain() throws IOException { Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0, originalRequest, this, eventListener, client.connectTimeoutMillis(), client.readTimeoutMillis(), client.writeTimeoutMillis()); return chain.proceed(originalRequest); } }
  16. Example: How Timeout is set? - RealConnection public HttpCodec newCodec(OkHttpClient

    client, Interceptor.Chain chain, StreamAllocation streamAllocation) throws SocketException { if (http2Connection != null) { return new Http2Codec(client, chain, streamAllocation, http2Connection); } else { socket.setSoTimeout(chain.readTimeoutMillis()); source.timeout().timeout(chain.readTimeoutMillis(), MILLISECONDS); sink.timeout().timeout(chain.writeTimeoutMillis(), MILLISECONDS); return new Http1Codec(client, streamAllocation, source, sink); } }
  17. Example: How Timeout is set? - RealConnection public HttpCodec newCodec(OkHttpClient

    client, Interceptor.Chain chain, StreamAllocation streamAllocation) throws SocketException { if (http2Connection != null) { return new Http2Codec(client, chain, streamAllocation, http2Connection); } else { socket.setSoTimeout(chain.readTimeoutMillis()); source.timeout().timeout(chain.readTimeoutMillis(), MILLISECONDS); sink.timeout().timeout(chain.writeTimeoutMillis(), MILLISECONDS); return new Http1Codec(client, streamAllocation, source, sink); } }
  18. Example: How Timeout is set? - RealConnection private void connectSocket(int

    connectTimeout, int readTimeout, Call call, EventListener eventListener) throws IOException { … try { source = Okio.buffer(Okio.source(rawSocket)); sink = Okio.buffer(Okio.sink(rawSocket)); } catch (NullPointerException npe) { if (NPE_THROW_WITH_NULL.equals(npe.getMessage())) { throw new IOException(npe); } }
  19. public static Source source(Socket socket) throws IOException { if (socket

    == null) throw new IllegalArgumentException("socket == null"); AsyncTimeout timeout = timeout(socket); Source source = source(socket.getInputStream(), timeout); return timeout.source(source); } Okio.source(Socket socket)
  20. public static Source source(Socket socket) throws IOException { if (socket

    == null) throw new IllegalArgumentException("socket == null"); AsyncTimeout timeout = timeout(socket); Source source = source(socket.getInputStream(), timeout); return timeout.source(source); } Okio.source(Socket socket)
  21. public static Source source(Socket socket) throws IOException { if (socket

    == null) throw new IllegalArgumentException("socket == null"); AsyncTimeout timeout = timeout(socket); Source source = source(socket.getInputStream(), timeout); return timeout.source(source); } Okio.source(Socket socket)
  22. public final Source source(final Source source) { return new Source()

    { @Override public long read(Buffer sink, long byteCount) throws IOException { boolean throwOnTimeout = false; enter(); try { long result = source.read(sink, byteCount); throwOnTimeout = true; return result; } catch (IOException e) { throw exit(e); } finally { exit(throwOnTimeout); } } AsyncTimeout.source
  23. public final Source source(final Source source) { return new Source()

    { @Override public long read(Buffer sink, long byteCount) throws IOException { boolean throwOnTimeout = false; enter(); try { long result = source.read(sink, byteCount); throwOnTimeout = true; return result; } catch (IOException e) { throw exit(e); } finally { exit(throwOnTimeout); } } AsyncTimeout.source
  24. public final Source source(final Source source) { return new Source()

    { @Override public long read(Buffer sink, long byteCount) throws IOException { boolean throwOnTimeout = false; enter(); try { long result = source.read(sink, byteCount); throwOnTimeout = true; return result; } catch (IOException e) { throw exit(e); } finally { exit(throwOnTimeout); } } AsyncTimeout.source
  25. OkHttp 概要 http://square.github.io/okhttp/ 効率的にHTTPなどのネットワーク通信をする - Connection Poolを使い、リクエストのレイテンシを減らす - GZIPなどの圧縮を透過的に行うことが出来る -

    Responseをキャッシュし、繰り返しのRequest時にネットワーク通信を避けることが 出来る InterceptorとCacheに絞って説明します。
  26. public interface Interceptor { Response intercept(Chain chain) throws IOException; }

    interface Chain { Response proceed(Request request) throws IOException; ... }
  27. Chain? Interceptor間のつなぎ込みをするクラス chain0(interceptor0, chain1) chain1(interceptor1, chain2) chain2… … chain2… chain1(interceptor1,

    chain2) chain0(interceptor0, chain1) 各chainは次へのchainを持っているので、chainが連鎖していく
  28. /** Chain.proceedメソッド内のコード */ // Call the next interceptor in the

    chain. RealInterceptorChain next = new RealInterceptorChain(interceptors, streamAllocation, httpCodec, connection, index + 1, request, call, eventListener, connectTimeout, readTimeout, writeTimeout); Interceptor interceptor = interceptors.get(index); Response response = interceptor.intercept(next);
  29. put レスポンス a put headers a hoge: fuga body a

    Journal DIRTY a in memory DIRTY a
  30. put レスポンス a オリジナルレ スポンス body a body a put

    Journal DIRTY a in memory DIRTY a headers a hoge: fuga
  31. put レスポンス a put Journal DIRTY a in memory DIRTY

    a オリジナルレ スポンス body a body a headers a hoge: fuga
  32. Source cacheWritingSource = new Source() { boolean cacheRequestClosed; @Override public

    long read(Buffer sink, long byteCount) { long bytesRead; try { bytesRead = source.read(sink, byteCount); // sourceはResponseBodyのstream } catch (IOException e) { // error handling throw e; } if (bytesRead == -1) { if (!cacheRequestClosed) { cacheRequestClosed = true; cacheBody.close(); } return -1; } sink.copyTo(cacheBody.buffer(), sink.size() - bytesRead, bytesRead); // ここでcacheに結果を読み込んだ結果をコピーする cacheBody.emitCompleteSegments(); return bytesRead; }
  33. put レスポンス a put source.read() Journal DIRTY a in memory

    DIRTY a headers a hoge: fuga body a オリジナルレ スポンス body a
  34. put body a Hello world レスポンス a source.read() 結果を返す put

    ついでに書き 込みも行う Journal DIRTY a in memory DIRTY a headers a hoge: fuga オリジナルレ スポンス body a
  35. put body a Hello world レスポンス a オリジナルレ スポンス body

    a source.read() 結果を返す put Journal DIRTY a CLEAN a in memory CLEAN a 完了通知 headers a hoge: fuga
  36. Journal DIRTY a CLEAN a in memory CLEAN a body

    a Hello world headers a hoge: fuga
  37. Journal DIRTY a CLEAN a ... in memory body a

    Hello world headers a hoge: fuga
  38. Journal DIRTY a CLEAN a ... in memory CLEAN a

    body a Hello world headers a hoge: fuga ファイルからin memoryに読み込み
  39. Journal DIRTY a CLEAN a ... in memory CLEAN a

    body a Hello world headers a hoge: fuga
  40. Journal DIRTY a CLEAN a in memory CLEAN a body

    a Hello world headers a hoge: fuga リクエスト a get key (url, vary) Candidate Response
  41. long now = System.currentTimeMillis(); CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(),

    cacheCandidate).get(); Request networkRequest = strategy.networkRequest; Response cacheResponse = strategy.cacheResponse;
  42. CacheControl requestCaching = request.cacheControl(); if (requestCaching.noCache() || hasConditions(request)) { return

    new CacheStrategy(request, null); } CacheControl responseCaching = cacheResponse.cacheControl(); if (responseCaching.immutable()) { return new CacheStrategy(null, cacheResponse); } long ageMillis = cacheResponseAge(); long freshMillis = computeFreshnessLifetime(); if (requestCaching.maxAgeSeconds() != -1) { freshMillis = Math.min(freshMillis, SECONDS.toMillis(requestCaching.maxAgeSeconds())); } ...
  43. String conditionName; String conditionValue; if (etag != null) { conditionName

    = "If-None-Match"; conditionValue = etag; } else if (lastModified != null) { conditionName = "If-Modified-Since"; conditionValue = lastModifiedString; } else if (servedDate != null) { conditionName = "If-Modified-Since"; conditionValue = servedDateString; } else { return new CacheStrategy(request, null); // No condition! Make a regular request. }
  44. Journal DIRTY a CLEAN a in memory CLEAN a body

    a Hello world headers a hoge: fuga リクエスト get key (url, method, vary) Candidate Response Candidate Responseが妥当かど うかをCacheStrategyから判定
  45. Journal DIRTY a CLEAN a in memory CLEAN a body

    a Hello world headers a hoge: fuga リクエスト get key (url, method, vary) Candidate Response Candidate Responseが妥当かど うかをCacheStrategyから判定 結果から、Network通信をするか、 Cache Responseを返すか決定す る Cacheが有効ならここから Responseを読み込む