import okhttp3.ws.WebSocketListener; //导入依赖的package包/类
public RealWebSocket(boolean isClient, BufferedSource source, BufferedSink sink, Random random,
    final Executor replyExecutor, final WebSocketListener listener, final String url) {
  this.listener = listener;

  writer = new WebSocketWriter(isClient, sink, random);
  reader = new WebSocketReader(isClient, source, new FrameCallback() {
    @Override public void onMessage(ResponseBody message) throws IOException {
      listener.onMessage(message);
    }

    @Override public void onPing(final Buffer buffer) {
      replyExecutor.execute(new NamedRunnable("OkHttp %s WebSocket Pong Reply", url) {
        @Override protected void execute() {
          try {
            writer.writePong(buffer);
          } catch (IOException ignored) {
          }
        }
      });
    }

    @Override public void onPong(Buffer buffer) {
      listener.onPong(buffer);
    }

    @Override public void onClose(final int code, final String reason) {
      readerSentClose = true;
      replyExecutor.execute(new NamedRunnable("OkHttp %s WebSocket Close Reply", url) {
        @Override protected void execute() {
          peerClose(code, reason);
        }
      });
    }
  });
}
 

import okhttp3.ws.WebSocketListener; //导入依赖的package包/类
/**
 * Attempts to perform a web socket upgrade on the connection. This will overwrite any previously
 * set status or body.
 */
public MockResponse withWebSocketUpgrade(WebSocketListener listener) {
  setStatus("HTTP/1.1 101 Switching Protocols");
  setHeader("Connection", "Upgrade");
  setHeader("Upgrade", "websocket");
  body = null;
  webSocketListener = listener;
  return this;
}
 

import okhttp3.ws.WebSocketListener; //导入依赖的package包/类
private void handleWebSocketUpgrade(Socket socket, BufferedSource source, BufferedSink sink,
    RecordedRequest request, MockResponse response) throws IOException {
  String key = request.getHeader("Sec-WebSocket-Key");
  String acceptKey = Util.shaBase64(key + WebSocketProtocol.ACCEPT_MAGIC);
  response.setHeader("Sec-WebSocket-Accept", acceptKey);

  writeHttpResponse(socket, sink, response);

  final WebSocketListener listener = response.getWebSocketListener();
  final CountDownLatch connectionClose = new CountDownLatch(1);

  ThreadPoolExecutor replyExecutor =
      new ThreadPoolExecutor(1, 1, 1, SECONDS, new LinkedBlockingDeque<Runnable>(),
          Util.threadFactory(String.format("MockWebServer %s WebSocket", request.getPath()),
              true));
  replyExecutor.allowCoreThreadTimeOut(true);
  final RealWebSocket webSocket =
      new RealWebSocket(false /* is server */, source, sink, new SecureRandom(), replyExecutor,
          listener, request.getPath()) {
        @Override protected void close() throws IOException {
          connectionClose.countDown();
        }
      };

  // Adapt the request and response into our Request and Response domain model.
  String scheme = request.getTlsVersion() != null ? "https" : "http";
  String authority = request.getHeader("Host"); // Has host and port.
  final Request fancyRequest = new Request.Builder()
      .url(scheme + "://" + authority + "/")
      .headers(request.getHeaders())
      .build();
  final Response fancyResponse = new Response.Builder()
      .code(Integer.parseInt(response.getStatus().split(" ")[1]))
      .message(response.getStatus().split(" ", 3)[2])
      .headers(response.getHeaders())
      .request(fancyRequest)
      .protocol(Protocol.HTTP_1_1)
      .build();

  listener.onOpen(webSocket, fancyResponse);

  while (webSocket.readMessage()) {
  }

  // Even if messages are no longer being read we need to wait for the connection close signal.
  try {
    connectionClose.await();
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }

  replyExecutor.shutdown();
  Util.closeQuietly(sink);
  Util.closeQuietly(source);
}
 

import okhttp3.ws.WebSocketListener; //导入依赖的package包/类
public WebSocketListener getWebSocketListener() {
  return webSocketListener;
}
 

import okhttp3.ws.WebSocketListener; //导入依赖的package包/类
private void handleWebSocketUpgrade(Socket socket, BufferedSource source, BufferedSink sink,
    RecordedRequest request, MockResponse response) throws IOException {
  String key = request.getHeader("Sec-WebSocket-Key");
  String acceptKey = Util.shaBase64(key + WebSocketProtocol.ACCEPT_MAGIC);
  response.setHeader("Sec-WebSocket-Accept", acceptKey);

  writeHttpResponse(socket, sink, response);

  final WebSocketListener listener = response.getWebSocketListener();
  final CountDownLatch connectionClose = new CountDownLatch(1);

  ThreadPoolExecutor replyExecutor =
      new ThreadPoolExecutor(1, 1, 1, SECONDS, new LinkedBlockingDeque<Runnable>(),
          Util.threadFactory(Util.format("MockWebServer %s WebSocket", request.getPath()),
              true));
  replyExecutor.allowCoreThreadTimeOut(true);
  final RealWebSocket webSocket =
      new RealWebSocket(false /* is server */, source, sink, new SecureRandom(), replyExecutor,
          listener, request.getPath()) {
        @Override protected void close() throws IOException {
          connectionClose.countDown();
        }
      };

  // Adapt the request and response into our Request and Response domain model.
  String scheme = request.getTlsVersion() != null ? "https" : "http";
  String authority = request.getHeader("Host"); // Has host and port.
  final Request fancyRequest = new Request.Builder()
      .url(scheme + "://" + authority + "/")
      .headers(request.getHeaders())
      .build();
  final Response fancyResponse = new Response.Builder()
      .code(Integer.parseInt(response.getStatus().split(" ")[1]))
      .message(response.getStatus().split(" ", 3)[2])
      .headers(response.getHeaders())
      .request(fancyRequest)
      .protocol(Protocol.HTTP_1_1)
      .build();

  listener.onOpen(webSocket, fancyResponse);

  while (webSocket.readMessage()) {
  }

  // Even if messages are no longer being read we need to wait for the connection close signal.
  try {
    connectionClose.await();
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }

  replyExecutor.shutdown();
  Util.closeQuietly(sink);
  Util.closeQuietly(source);
}