OKHttp源码解析(五)--OKIO简介及FileSystem

    public static void main(String[] args) {
        File file = new File("hello.txt");
        try {
            readString(new FileInputStream(file));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void readString(InputStream in) throws IOException {
      BufferedSource source = Okio.buffer(Okio.source(in));  //创建BufferedSource
      String s = source.readUtf8();  //以UTF-8读
      System.out.println(s);     //打印
      source.close();
    }

public interface Sink extends Closeable, Flushable {
  /** Removes {@code byteCount} bytes from {@code source} and appends them to this. */
  // 定义基础的write操作,该方法将字节写入Buffer
  void write(Buffer source, long byteCount) throws IOException;

  /** Pushes all buffered bytes to their final destination. */
  @Override void flush() throws IOException;

  /** Returns the timeout for this sink. */
  Timeout timeout();

  /**
   * Pushes all buffered bytes to their final destination and releases the
   * resources held by this sink. It is an error to write a closed sink. It is
   * safe to close a sink more than once.
   */
  @Override void close() throws IOException;
}

public interface Source extends Closeable {
  /**
   * Removes at least 1, and up to {@code byteCount} bytes from this and appends
   * them to {@code sink}. Returns the number of bytes read, or -1 if this
   * source is exhausted.
   */
    // 定义基础的read操作,该方法将字节写入Buffer
  long read(Buffer sink, long byteCount) throws IOException;

  /** Returns the timeout for this source. */
  Timeout timeout();

  /**
   * Closes this source and releases the resources held by this source. It is an
   * error to read a closed source. It is safe to close a source more than once.
   */
  @Override void close() throws IOException;
}

final class RealBufferedSink implements BufferedSink {
    public final Buffer buffer = new Buffer();
    public final Sink sink;

    public RealBufferedSink(Sink sink){
        if (sink == null)
            throw new NullPointerException("sink == null");
        this.sink = sink;
    }
}

  @Override public BufferedSink write(byte[] source) throws IOException {
    if (closed) throw new IllegalStateException("closed");
    buffer.write(source);
    return emitCompleteSegments();
  }

  @Override public BufferedSink write(byte[] source, int offset, int byteCount) throws IOException {
    if (closed) throw new IllegalStateException("closed");
    buffer.write(source, offset, byteCount);
    return emitCompleteSegments();
  }

      // 每一个segment所含数据的大小，固定的
    static final int SIZE = 8192;
     // Segments 用分享的方式避免复制数组
    static final int SHARE_MINIMUM = 1024;
  
    final byte[] data;
     // data[]中第一个可读的位置
    int pos;
     // data[]中第一个可写的位 
     //所以一个Segment的可读数据数量为pos~limit-1=limit-pos；limit和pos的有效值为0~SIZE-1
    int limit;
    //当前存储的data数据是其它对象共享的则为真  
    boolean shared;
    //是否是自己是操作者 
    boolean owner;
    ///前一个Segment
    Segment pre;
    ///下一个Segment
    Segment next;

Segment() {  
  this.data = new byte[SIZE];  
  this.owner = true;   
  this.shared = false; 
}  
Segment(Segment shareFrom) {  
  this(shareFrom.data, shareFrom.pos, shareFrom.limit);  
  shareFrom.shared = true;
}  
Segment(byte[] data, int pos, int limit) {  
  this.data = data;  
  this.pos = pos;  
  this.limit = limit;  
  this.owner = false; 
  this.shared = true;  
}  

    public Segment pop(){
        Segment result = next != this ? next : null;
        pre.next = next;
        next.pre = pre;
        next = null;
        pre = null;
        return result;
    }

    public Segment push(Segment segment){
        segment.pre = this;
        segment.next = next;
        next.pre = segment;
        next = segment;
        return segment;
    }

  /** Moves {@code byteCount} bytes from this segment to {@code sink}. */
  public void writeTo(Segment sink, int byteCount) {
     //只能对自己操作
    if (!sink.owner) throw new IllegalArgumentException();
     //写的起始位置加上需要写的byteCount大于SIZE
    if (sink.limit + byteCount > SIZE) {
      // We can't fit byteCount bytes at the sink's current position. Shift sink first.
       //如果是共享内存，不能写，则抛异常
      if (sink.shared) throw new IllegalArgumentException();
      //如果不是共享内存，且写的起始位置加上byteCount减去头还大于SIZE则抛异常
      if (sink.limit + byteCount - sink.pos > SIZE) throw new IllegalArgumentException();
      //上述条件都不满足，我们需要先触动要写文件的地址，把sink.data从sink.pos这个位置移动到 0这个位置，移动的长度是limit - sink.pos，这样的好处就是sink初始位置是0
      System.arraycopy(sink.data, sink.pos, sink.data, 0, sink.limit - sink.pos);
      sink.limit -= sink.pos;
      sink.pos = 0;
    }
    //开始尾部写入 写完置limit地址
    System.arraycopy(data, pos, sink.data, sink.limit, byteCount);
    sink.limit += byteCount;
    pos += byteCount;//索引后移
  }

  public void compact() {
    //上一个节点就是自己，意味着就一个节点，无法压缩，抛异常
    if (prev == this) throw new IllegalStateException();
     //如果上一个节点不是自己的，所以你是没有权利压缩的
    if (!prev.owner) return; // Cannot compact: prev isn't writable.
    //能进来说明，存在上一个节点，且上一个节点是自己的，可以压缩
    //记录当前Segment具有的数据，数据大小为limit-pos
    int byteCount = limit - pos;
    统计前结点是否被共享，如果共享则只记录Size-limit大小，如果没有被共享，则加上pre.pos之前的空位置；
    //本行代码主要是获取前一个segment的可用空间。先判断prev是否是共享的，如果是共享的，则只记录SIZE-limit,如果没有共享则记录SIZE-limit加上prev.pos之前的空位置
    int availableByteCount = SIZE - prev.limit + (prev.shared ? 0 : prev.pos);
   //判断prev的空余空间是否能够容纳Segment的全部数据，不能容纳则返回
    if (byteCount > availableByteCount) return; // Cannot compact: not enough writable space.
    //能容纳则将自己的这个部分数据写入上一个Segment
    writeTo(prev, byteCount);
    //讲当前Segment从Segment链表中移除
    pop();
    //回收该Segment
    SegmentPool.recycle(this);
  }

 public Segment split(int byteCount) {
    if (byteCount <= 0 || byteCount > limit - pos) throw new IllegalArgumentException();
    Segment prefix;

    // We have two competing performance goals:
    //  - Avoid copying data. We accomplish this by sharing segments.
    //  - Avoid short shared segments. These are bad for performance because they are readonly and
    //    may lead to long chains of short segments.
    // To balance these goals we only share segments when the copy will be large.
    if (byteCount >= SHARE_MINIMUM) {
      prefix = new Segment(this);
    } else {
      prefix = SegmentPool.take();
      System.arraycopy(data, pos, prefix.data, 0, byteCount);
    }

    prefix.limit = prefix.pos + byteCount;
    pos += byteCount;
    prev.push(prefix);
    return prefix;
  }

//一个Segment记录的最大长度是8192，因此SegmentPool只能存储8个Segment
static final long MAX_SIZE = 64 * 1024;
//该SegmentPool存储了一个回收Segment的链表
static Segment next;
//该值记录了当前所有Segment的总大小，最大值是为MAX_SIZE
static long byteCount;  

  static Segment take() {
    synchronized (SegmentPool.class) {
      if (next != null) {
        Segment result = next;
        next = result.next;
        result.next = null;
        byteCount -= Segment.SIZE;
        return result;
      }
    }
    return new Segment(); // Pool is empty. Don't zero-fill while holding a lock.
  }

  static void recycle(Segment segment) {
    //如果这个要回收的Segment被前后引用，则无法回收
    if (segment.next != null || segment.prev != null) throw new IllegalArgumentException();
    //如果这个要回收的Segment的数据是分享的，则无法回收
    if (segment.shared) return; // This segment cannot be recycled.
    //加锁
    synchronized (SegmentPool.class) {
      //如果 这个空间已经不足以再放入一个空的Segment，则不回收
      if (byteCount + Segment.SIZE > MAX_SIZE) return; // Pool is full.
      //设置SegmentPool的池大小
      byteCount += Segment.SIZE;
      //segment的下一个指向头
      segment.next = next;
      //设置segment的可读写位置为0
      segment.pos = segment.limit = 0;
      //设置当前segment为头
      next = segment;
    }
  }

//明显给16进制数准备的
static final char[] HEX_DIGITS = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };
  //一个空的对象 
  /** A singleton empty {@code ByteString}. */
  public static final ByteString EMPTY = ByteString.of();
  
  final byte[] data;
  //hashcode的值
  transient int hashCode; // Lazily computed; 0 if unknown.
  transient String utf8; // Lazily computed.

  ByteString(byte[] data) {
    this.data = data; // Trusted internal constructor doesn't clone data.
  }

  /**
   * Returns a new byte string containing a clone of the bytes of {@code data}.
   */
  public static ByteString of(byte... data) {
    if (data == null) throw new IllegalArgumentException("data == null");
    return new ByteString(data.clone());
  }

  /**
   * Returns a new byte string containing a copy of {@code byteCount} bytes of {@code data} starting
   * at {@code offset}.
   */
  public static ByteString of(byte[] data, int offset, int byteCount) {
    if (data == null) throw new IllegalArgumentException("data == null");
    checkOffsetAndCount(data.length, offset, byteCount);

    byte[] copy = new byte[byteCount];
    System.arraycopy(data, offset, copy, 0, byteCount);
    return new ByteString(copy);
  }

  public static ByteString of(ByteBuffer data) {
    if (data == null) throw new IllegalArgumentException("data == null");

    byte[] copy = new byte[data.remaining()];
    data.get(copy);
    return new ByteString(copy);

/**
   * Returns a human-readable string that describes the contents of this byte string. Typically this
   * is a string like {@code [text=Hello]} or {@code [hex=0000ffff]}.
   */
  @Override public String toString() {
    if (data.length == 0) {
      return "[size=0]";
    }

    String text = utf8();
    int i = codePointIndexToCharIndex(text, 64);

    if (i == -1) {
      return data.length <= 64
          ? "[hex=" + hex() + "]"
          : "[size=" + data.length + " hex=" + substring(0, 64).hex() + "…]";
    }

    String safeText = text.substring(0, i)
        .replace("\\", "\\\\")
        .replace("\n", "\\n")
        .replace("\r", "\\r");
    return i < text.length()
        ? "[size=" + data.length + " text=" + safeText + "…]"
        : "[text=" + safeText + "]";
  }

  /** Constructs a new {@code String} by decoding the bytes as {@code UTF-8}. */
  public String utf8() {
    String result = utf8;
    // We don't care if we double-allocate in racy code.
    return result != null ? result : (utf8 = new String(data, Util.UTF_8));
  }

public final class Buffer implements BufferedSource, BufferedSink, Cloneable {
  //明显是给16进制准备的
  private static final byte[] DIGITS =
      { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };
  static final int REPLACEMENT_CHARACTER = '\ufffd';
  //Buffer存储了一个这样的head节点，这就是Buffer对数据的存储结构。字节数组都是交给Segment进行管理
  Segment head;
  //当前存储的数据大小
  long size;

  //看到注释的第一句话，我就知道了是深拷贝，哈哈！
  /** Returns a deep copy of this buffer. */
  @Override public Buffer clone() {
    //先new了一个Buffer对象
    Buffer result = new Buffer();
    //如果size==0，说明这个Buffer是空的，所以直接返回即可
    if (size == 0) return result;
    //如果size!=0，说明这个Buffer是有数据的，然后吧head指向这个copy的head,PS大家回想下Segment的这个构造函数，里面是怎么操作的？
    result.head = new Segment(head);
    //然后设置copy的head的next和prev的值
    result.head.next = result.head.prev = result.head;
    //开始遍历这个Buffer持有的Segment链了
    for (Segment s = head.next; s != head; s = s.next) {
      result.head.prev.push(new Segment(s));
    }
    result.size = size;
    return result;
  }

 @Override public int readInt() {
    //第一步
    if (size < 4) throw new IllegalStateException("size < 4: " + size);
    //第二步
    Segment segment = head;
    int pos = segment.pos;
    int limit = segment.limit;
    // If the int is split across multiple segments, delegate to readByte().
    if (limit - pos < 4) {
      return (readByte() & 0xff) << 24
          |  (readByte() & 0xff) << 16
          |  (readByte() & 0xff) <<  8
          |  (readByte() & 0xff);
    }
     //第三步
    byte[] data = segment.data;
    int i = (data[pos++] & 0xff) << 24
        |   (data[pos++] & 0xff) << 16
        |   (data[pos++] & 0xff) <<  8
        |   (data[pos++] & 0xff);
    size -= 4;
    //第四步
    if (pos == limit) {
      head = segment.pop();
      SegmentPool.recycle(segment);
    } else {
      segment.pos = pos;
    }
    //第五步
    return i;
  }

 @Override public byte readByte() {
     //第一步
    if (size == 0) throw new IllegalStateException("size == 0");
     //第二步
    Segment segment = head;
    int pos = segment.pos;
    int limit = segment.limit;
    //第三步
    byte[] data = segment.data;
    //第四步
    byte b = data[pos++];
    size -= 1;
    //第五步
    if (pos == limit) {
      head = segment.pop();
      SegmentPool.recycle(segment);
    } else {
      segment.pos = pos;
    }
    //第六步
    return b;
  }

  @Override public Buffer writeShort(int s) {
    //第一步
    Segment tail = writableSegment(2);
    //第二步
    byte[] data = tail.data;
    int limit = tail.limit;
    data[limit++] = (byte) ((s >>> 8) & 0xff);
    data[limit++] = (byte)  (s        & 0xff);
    tail.limit = limit;
    //第三步
    size += 2;
    //第四步
    return this;
  }

  /**
   * Returns a tail segment that we can write at least {@code minimumCapacity}
   * bytes to, creating it if necessary.
   */
  Segment writableSegment(int minimumCapacity) {
    if (minimumCapacity < 1 || minimumCapacity > Segment.SIZE) throw new IllegalArgumentException();
    //如果hea=null表明Buffer里面一个Segment都没有
    if (head == null) {
      //从SegmentPool取出一个Segment 
      head = SegmentPool.take(); // Acquire a first segment.
      return head.next = head.prev = head;
    }
    //如果head不为null.
    Segment tail = head.prev;
    //如果已经写入的数据+最小可以写入的空间超过限制，则在SegmentPool里面取一个
    if (tail.limit + minimumCapacity > Segment.SIZE || !tail.owner) {
      tail = tail.push(SegmentPool.take()); // Append a new empty segment to fill up.
    }
    return tail;
  }

//Okio.java

 //实际上调用的两个参数的sink方法，第二个参数是new的TimeOut对象，即同步超时
public static Sink sink(OutputStream out) {
    return sink(out, new Timeout());
  }

  private static Sink sink(final OutputStream out, final Timeout timeout) {
    if (out == null) throw new IllegalArgumentException("out == null");
    if (timeout == null) throw new IllegalArgumentException("timeout == null");

    return new Sink() {
        @Override 
        public void write(Buffer source, long byteCount) throws IOException {
        checkOffsetAndCount(source.size, 0, byteCount);
        while (byteCount > 0) {
          timeout.throwIfReached();
          Segment head = source.head;
          int toCopy = (int) Math.min(byteCount, head.limit - head.pos);
          out.write(head.data, head.pos, toCopy);

          head.pos += toCopy;
          byteCount -= toCopy;
          source.size -= toCopy;

          if (head.pos == head.limit) {
            source.head = head.pop();
            SegmentPool.recycle(head);
          }
        }
      }

      @Override public void flush() throws IOException {
        out.flush();
      }

      @Override public void close() throws IOException {
        out.close();
      }

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

      @Override public String toString() {
        return "sink(" + out + ")";
      }
    };
  }

  /**
   * Throws an {@link InterruptedIOException} if the deadline has been reached or if the current
   * thread has been interrupted. This method doesn't detect timeouts; that should be implemented to
   * asynchronously abort an in-progress operation.
   */
  public void throwIfReached() throws IOException {
    if (Thread.interrupted()) {
      throw new InterruptedIOException("thread interrupted");
    }

    if (hasDeadline && deadlineNanoTime - System.nanoTime() <= 0) {
      throw new InterruptedIOException("deadline reached");
    }
  }

  /**
   * Returns a sink that writes to {@code socket}. Prefer this over {@link
   * #sink(OutputStream)} because this method honors timeouts. When the socket
   * write times out, the socket is asynchronously closed by a watchdog thread.
   */
  public static Sink sink(Socket socket) throws IOException {
    if (socket == null) throw new IllegalArgumentException("socket == null");
    AsyncTimeout timeout = timeout(socket);
    Sink sink = sink(socket.getOutputStream(), timeout);
    return timeout.sink(sink);
  }

  private static final class Watchdog extends Thread {
    public Watchdog() {
      super("Okio Watchdog");
      setDaemon(true);
    }

    public void run() {
      while (true) {
        try {
          AsyncTimeout timedOut;
          synchronized (AsyncTimeout.class) {
            timedOut = awaitTimeout();

            // Didn't find a node to interrupt. Try again.
            if (timedOut == null) continue;

            // The queue is completely empty. Let this thread exit and let another watchdog thread
            // get created on the next call to scheduleTimeout().
            if (timedOut == head) {
              head = null;
              return;
            }
          }

          // Close the timed out node.
          timedOut.timedOut();
        } catch (InterruptedException ignored) {
        }
      }
    }
  }

    //不要一次写超过64k的数据否则可能会在慢连接中导致超时
    private static final int TIMEOUT_WRITE_SIZE = 64 * 1024;
    private static AsyncTimeout head;
    private boolean inQueue;
    private AsyncTimeout next;
    private long timeoutAt;

   public final void enter() {
    if (inQueue) throw new IllegalStateException("Unbalanced enter/exit");
    long timeoutNanos = timeoutNanos();
    boolean hasDeadline = hasDeadline();
    if (timeoutNanos == 0 && !hasDeadline) {
      return; // No timeout and no deadline? Don't bother with the queue.
    }
    inQueue = true;
    scheduleTimeout(this, timeoutNanos, hasDeadline);
  }

private static synchronized void scheduleTimeout(
      AsyncTimeout node, long timeoutNanos, boolean hasDeadline) {
    // Start the watchdog thread and create the head node when the first timeout is scheduled.
    //head==null，表明之前没有，本次是第一次操作，开启Watchdog守护线程
    if (head == null) {
      head = new AsyncTimeout();
      new Watchdog().start();
    }

    long now = System.nanoTime();
    //如果有最长限制(hasDeadline我翻译为最长限制)，并且超时时长不为0
    if (timeoutNanos != 0 && hasDeadline) {
      // Compute the earliest event; either timeout or deadline. Because nanoTime can wrap around,
      // Math.min() is undefined for absolute values, but meaningful for relative ones.
      //对比最长限制和超时时长，选择最小的那个值
      node.timeoutAt = now + Math.min(timeoutNanos, node.deadlineNanoTime() - now);
    } else if (timeoutNanos != 0) {
      //如果没有最长限制，但是超时时长不为0，则使用超时时长
      node.timeoutAt = now + timeoutNanos;
    } else if (hasDeadline) {
      //如果有最长限制，但是超时时长为0，则使用最长限制
      node.timeoutAt = node.deadlineNanoTime();
    } else {
     //如果既没有最长限制，和超时时长，则抛异常
      throw new AssertionError();
    }

    // Insert the node in sorted order.
    long remainingNanos = node.remainingNanos(now);
    for (AsyncTimeout prev = head; true; prev = prev.next) {
      //如果下一个为null或者剩余时间比下一个短 就插入node
      if (prev.next == null || remainingNanos < prev.next.remainingNanos(now)) {
        node.next = prev.next;
        prev.next = node;
        if (prev == head) {
          AsyncTimeout.class.notify(); // Wake up the watchdog when inserting at the front.
        }
        break;
      }
    }
  }

     @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);
             }
      }

    final void exit(boolean throwOnTimeout) throws IOException {
        boolean timeOut =  exit();
        if (timeOut && throwOnTimeout)
            throw newTimeoutException(null);
    }

    public final boolean exit(){
        if (!inQueue)
            return false;
        inQueue = false;
        return cancelScheduledTimeout(this);
    }

static AsyncTimeout awaitTimeout() throws InterruptedException {
        //拿到下一个节点
        AsyncTimeout node = head.next;
        //如果queue为空，等待直到有node进队，或者触发IDLE_TIMEOUT_MILLS
        if (node == null) {
            long startNanos = System.nanoTime();
            AsyncTimeout.class.wait(IDLE_TIMEOUT_MILLS);
            return head.next == null && (System.nanoTime() - startNanos) >= IDLE_TIMEOUT_NANOS ? head : null;
        }
        long waitNanos = node.remainingNanos(System.nanoTime());
        //这个head依然还没有超时，继续等待
        if (waitNanos > 0) {
            long waitMills = waitNanos / 1000000L;
            waitNanos -= (waitMills * 1000000L);
            AsyncTimeout.class.wait(waitMills, (int) waitNanos);
            return null;
        }
        head.next = node.next;
        node.next = null;
        return node;
    }

public interface FileSystem {
  /** The host machine's local file system. */
  FileSystem SYSTEM = new FileSystem() {
    @Override public Source source(File file) throws FileNotFoundException {
      return Okio.source(file);
    }

    @Override public Sink sink(File file) throws FileNotFoundException {
      try {
        return Okio.sink(file);
      } catch (FileNotFoundException e) {
        // Maybe the parent directory doesn't exist? Try creating it first.
        file.getParentFile().mkdirs();
        return Okio.sink(file);
      }
    }

    @Override public Sink appendingSink(File file) throws FileNotFoundException {
      try {
        return Okio.appendingSink(file);
      } catch (FileNotFoundException e) {
        // Maybe the parent directory doesn't exist? Try creating it first.
        file.getParentFile().mkdirs();
        return Okio.appendingSink(file);
      }
    }

    @Override public void delete(File file) throws IOException {
      // If delete() fails, make sure it's because the file didn't exist!
      if (!file.delete() && file.exists()) {
        throw new IOException("failed to delete " + file);
      }
    }

    @Override public boolean exists(File file) {
      return file.exists();
    }

    @Override public long size(File file) {
      return file.length();
    }

    @Override public void rename(File from, File to) throws IOException {
      delete(to);
      if (!from.renameTo(to)) {
        throw new IOException("failed to rename " + from + " to " + to);
      }
    }

    @Override public void deleteContents(File directory) throws IOException {
      File[] files = directory.listFiles();
      if (files == null) {
        throw new IOException("not a readable directory: " + directory);
      }
      for (File file : files) {
        if (file.isDirectory()) {
          deleteContents(file);
        }
        if (!file.delete()) {
          throw new IOException("failed to delete " + file);
        }
      }
    }
  };

  /** Reads from {@code file}. */
  Source source(File file) throws FileNotFoundException;

  /**
   * Writes to {@code file}, discarding any data already present. Creates parent directories if
   * necessary.
   */
  Sink sink(File file) throws FileNotFoundException;

  /**
   * Writes to {@code file}, appending if data is already present. Creates parent directories if
   * necessary.
   */
  Sink appendingSink(File file) throws FileNotFoundException;

  /** Deletes {@code file} if it exists. Throws if the file exists and cannot be deleted. */
  void delete(File file) throws IOException;

  /** Returns true if {@code file} exists on the file system. */
  boolean exists(File file);

  /** Returns the number of bytes stored in {@code file}, or 0 if it does not exist. */
  long size(File file);

  /** Renames {@code from} to {@code to}. Throws if the file cannot be renamed. */
  void rename(File from, File to) throws IOException;

  /**
   * Recursively delete the contents of {@code directory}. Throws an IOException if any file could
   * not be deleted, or if {@code dir} is not a readable directory.
   */
  void deleteContents(File directory) throws IOException;
}