okiobuffer.java

  public void readTwice()
  {
              Observable.fromCallable(() -> {
              RedditData inflatedModel = null;
              Response response = makeRequest();

              String diskValue = null;
              try {
                  File file = new File(getContext().getCacheDir(), "file");
                  BufferedSink cacheBody = Okio.buffer(Okio.sink(file));

                  Source cacheWritingSource = getCacheSource(response.body().source(), cacheBody);
                  response = response.newBuilder()
                          .body(new RealResponseBody(response.headers(), Okio.buffer(cacheWritingSource)))
                          .build();

                  InputStreamReader inputStreamReader = new InputStreamReader(response.body().source().inputStream());
                  inflatedModel = new GsonBuilder()
                           .create()
                          .fromJson(inputStreamReader, RedditData.class);
                  closeAllTheThings(response, cacheBody);

                  diskValue = Okio.buffer(Okio.source(file)).readUtf8Line();


              } catch (Exception e) {
                  e.printStackTrace();
              }
              assert diskValue != null;
              return inflatedModel;
          })
          .subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe();

  }

  private Response makeRequest() throws IOException {
      Request request = new Request.Builder()
              .url("https://www.reddit.com/r/aww/new/.json")
              .build();
      return new OkHttpClient().newCall(request).execute();
  }


  private void closeAllTheThings(Response response, BufferedSink cacheBody) throws IOException {
      response.body().source().close();
      cacheBody.close();
  }

  @NonNull
  private Source getCacheSource(final Source source, final BufferedSink cacheBody) {
      return new Source() {
          boolean cacheRequestClosed;

          @Override
          public long read(Buffer sink, long byteCount) throws IOException {
              long bytesRead;
              try {
                  bytesRead = source.read(sink, byteCount);
              } catch (IOException e) {
                  if (!cacheRequestClosed) {
                      cacheRequestClosed = true;
                  }
                  throw e;
              }

              if (bytesRead == -1) {
                  if (!cacheRequestClosed) {
                      cacheRequestClosed = true;
                      cacheBody.close(); // The cache response is complete!
                  }
                  return -1;
              }

              sink.copyTo(cacheBody.buffer(), sink.size() - bytesRead, bytesRead);
              cacheBody.emitCompleteSegments();
              return bytesRead;
          }

          @Override
          public Timeout timeout() {
              return source.timeout();
          }

          @Override
          public void close() throws IOException {
              if (!cacheRequestClosed
                      && !discard(this, HttpStream.DISCARD_STREAM_TIMEOUT_MILLIS, MILLISECONDS)) {
                  cacheRequestClosed = true;
              }
              source.close();
          }
      };
  }