Playwright源码之PipeTransport(四)

public interface Transport {
  //  发送消息
  void send(JsonObject message);
  // 拉取消息
  JsonObject poll(Duration timeout);
  // 关闭
  void close() throws IOException;
}

public class PipeTransport implements Transport {
	// **********************************************************
	// 1. 创建阻塞队列
	// 主要用于,当向浏览器发送消息以及浏览器回复消息时的载体.
	// **********************************************************
	private final BlockingQueue<JsonObject> incoming = new ArrayBlockingQueue<>(1000);
	private final BlockingQueue<String> outgoing= new ArrayBlockingQueue<>(1000);

	private final ReaderThread readerThread;
	private final WriterThread writerThread;

	private boolean isClosed;

	PipeTransport(InputStream input, OutputStream output) {
		
		DataInputStream in = new DataInputStream(new BufferedInputStream(input));
		
		// **********************************************************
		// 2. 创建读写线程,并启动.
		// **********************************************************
		readerThread = new ReaderThread(in, incoming);
		readerThread.start();
		writerThread = new WriterThread(output, outgoing);
		writerThread.start();
	} // end 
	
	public void send(JsonObject message) {
		if (isClosed) {
		  throw new PlaywrightException("Playwright connection closed");
		}
		
		try {
          // **************************************************************
		  // 3. 把对象消息转换成JSON字符串,并,插入到队列(BlockingQueue<String>)里.
		  // **************************************************************
		  outgoing.put(gson().toJson(message));
		} catch (InterruptedException e) {
		  throw new PlaywrightException("Failed to send message", e);
		}
	} // end send

  @Override
  public JsonObject poll(Duration timeout) {
    if (isClosed) {
      throw new PlaywrightException("Playwright connection closed");
    }
	
    try {
	   // **************************************************************
	   // 4. 读取队列(BlockingQueue<JsonObject>)里的数据
	   // **************************************************************
      JsonObject message = incoming.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
      if (message == null && readerThread.exception != null) {
        try {
          close();
        } catch (IOException e) {
          e.printStackTrace(System.err);
        }
        throw new PlaywrightException("Failed to read message from driver, pipe closed.", readerThread.exception);
      }
      return message;
    } catch (InterruptedException e) {
      throw new PlaywrightException("Failed to read message", e);
    }
  }// end 	poll
}