打开APP
userphoto
未登录

开通VIP,畅享免费电子书等14项超值服

开通VIP
学习WebSocket协议

Android最佳实践——深入浅出WebSocket协议

http://blog.csdn.net/sbsujjbcy/article/details/52839540

首先明确一下概念,WebSocket协议是一种建立在TCP连接基础上的全双工通信的协议。概念强调了两点内容:

  • TCP基础上
  • 全双工通信

那么什么是全双工通信呢?

全双工就是指客户端和服务端可以同时进行双向通信,强调同时、双向通信

WebSocket可以应用于即时通信等场景,比如现在直播很火热,直播中的弹幕也可以使用WebSocket去实现。

WebSocket的协议内容可以见 The WebSocket Protocol,讲得最全面的官方说明。简单介绍可以见维基百科WebSocket

在Android客户端,一般可以使用下面的库完成WebSocket通信

  • okhttp,一般人我不告诉他okhttp还可以用来进行WebSocket通信
  • Java-WebSocket,纯java实现的WebSocket客户端和服务端实现

那么在没有服务端支持的情况下,我们客户端如何进行WebSocket的测试呢?一般人我也不告诉他!答案还是okhttp,这次是okhttp的扩展模块mockserver,不过最新版本的okhttp已经把WebSocket合入okhttp核心库中去了,如果你用的版本比较低,就可能需要依赖okhttp-ws模块。

先来看协议内容组成,先上一张神图

WebSocket按上面图中协议规则进行传输,上图称为一个数据帧。

  • FIN,共1位,标记消息是否是最后1帧,1个消息由1个或多个数据帧构成,若消息由1帧构成,起始帧就是结束帧。
  • RSV1,RSV2,RSV3,各1位,预留位,用于自定义扩展。如果没有扩展,各位值为0;如果定义了扩展,即为非0值。如果接收的帧中此处为非0,但是扩展中却没有该值的定义,那么关闭连接。
  • OPCODE,共4位,帧类型,分为控制帧和非控制帧。如果接收到未知帧,接收端必须关闭连接。已定义的帧类型如下图所示: 

除了上图中的0,1,2外(0x0,0x1,0x2),3-7(0x3-0x7)暂时没有进行定义,为以后的非控制帧保留。 
除了上图中的8,9,10(0x8,0x9,0xA)外,11-15(0xB-0xF)暂时没有进行定义,为以后的控制帧保留。

消息的分片,一般来说,对于一个长度较小的消息,可以使用1帧完成消息的发送, 比如说文本消息,Fin的值为1,表示结束,Opcode值不能为0,0表示后续还有数据帧会发送过来。 
而对于一些长度较长的消息,则需要将消息进行分片发送。比如语音消息,这时候起始帧的FIN值为0,Opcode为非0,接着是若干帧(FIN值都为0,Opcode值为0),结束帧FIN值为1,Opcode值为0。

WebSocket的控制帧有3种,关闭帧、Ping帧、Pong帧,关闭帧很好理解,客户端如果收到关闭帧直接关闭连接即可,当然客户端也可以发送关闭帧给服务器端。而Ping帧和Pong帧则是WebSocket的心跳检测,用于保证客户端是在线的,一般来说,只有服务端给客户端发送Ping帧,然后客户端发送Pong帧进行回应,表示自己还在线,可以进行后续通信。

  • MASK,共1位,掩码位,表示帧中的数据是否经过加密,客户端发出的数据帧需要经过掩码处理,这个值都是1。如果值是1,那么Masking-key域的数据就是掩码秘钥,用于解码PayloadData,否则Masking-key长度为0。

WebSocket协议规定数据通过帧序列传输。客户端必须对其发送到服务器的所有帧进行掩码处理。

服务器一旦收到无掩码帧,将关闭连接。服务器可能发送一个状态码是1002(表示协议错误)的Close帧。

而服务器发送客户端的数据帧不做掩码处理,一旦客户端发现经过掩码处理的帧,将关闭连接。客户端可能使用状态码1002。

更多状态码如下图所示:

  • Payload len,7位或者7+16位或者7+64位,表示数据帧中数据大小,这里有好几种情况。

    • 如果值为0-125,那么该值就是payload data的真实长度。
    • 如果值为126,那么该7位后面紧跟着的2个字节就是payload data的真实长度。
    • 如果值为127,那么该7位后面紧跟着的8个字节就是payload data的真实长度。
    • 长度遵循一个原则,就是用最少的字节表示长度,举个例子,当payload data的真实长度是124时,在0-125之间,必须用7位表示;不允许将这7位表示成126或者127,然后后面用2个字节或者8个字节表示124,这样做就违反了原则。
  • Masking-key ,0或者4个字节,当MASK位为1时,4个字节,否则0个字节。如果MASK值为1,则发出去的数据需要经过加密处理,加密流程如下:

  1. void mask(byte[] original, byte[] maskKey) {
  2. for (int i = 0; i < original.length; i++) {
  3. original[i] = (byte) (original[i] ^ maskKey[i % 4]);
  4. }
  5. }
  • 最后就是Payload data,其大小是(x+y)个字节,x是Extension data,即扩展数据,y是Application data,即程序数据,扩展数据可能为0。 如果扩展数据不为0,必须提前进行协商,规定其长度,否则是不合法的数据帧。

以上是WebSocket数据传输的帧内容,大致了解即可。除此之外,WebSocket协议还有一个握手的过程。握手通过发送一个http请求来完成,这里基本和http2.0有点类似,客户端发送一个请求协议升级的get请求给服务端,服务端如果支持的话会返回http code 为101,表示可以切换到对应的协议。大致流程如下:

  • 客户端发送get请求协议升级
  1. Upgrade: websocket
  2. Connection: Upgrade
  3. Sec-WebSocket-Key: RCfYMqhgCo4N4E+cIZ0iPg==
  4. Sec-WebSocket-Version: 13

该请求会在请求头上带上WebSocket的版本号,这里是13,以及客户端随机生成的Sec-WebSocket-Key,服务器端收到后根据这个key进行一些处理,返回一个Sec-WebSocket-Accept的值给客户端。

  • 服务端返回同意升级到WebSocket协议
  1. HTTP/1.1 101 Switching Protocols
  2. Upgrade: websocket
  3. Connection: upgrade
  4. Sec-WebSocket-Accept: b7RAFizjwDE9lWS46ZMPfmN35wc=

收到响应后,响应头中包含Sec-WebSocket-Accept值,该值表示服务器端同意握手,值的计算方式如下:

$(Sec-WebSocket-Accept)=BASE64(SHA1($(Sec-WebSocket-Key)+"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"))

客户端得到该值后,对本地的Sec-WebSocket-Key进行同样的编码,然后对比,如果相同则可以进行后续处理。

关于WebSocket协议,一般来说,如果是通过https协议开始升级而来的,那么一般是wss://开头,如果是http协议开始升级而来的,那么一般是ws://开头

讲完了概念性的东西,接下来就是最佳实践了。

那么客户端怎么进行WebSocket测试呢?这里我们使用OkHttp的扩展模块Mock Server来实现。

首先引入okhttp依赖和mockserver依赖,对maven来说,内容如下

  1. <dependency>
  2. <groupId>com.squareup.okhttp3</groupId>
  3. <artifactId>okhttp</artifactId>
  4. <version>3.4.1</version>
  5. </dependency>
  6. <dependency>
  7. <groupId>com.squareup.okhttp3</groupId>
  8. <artifactId>mockwebserver</artifactId>
  9. <version>3.4.1</version>
  10. </dependency>

对gradle来说,其内容如下

  1. compile 'com.squareup.okhttp3:okhttp:3.4.1'
  2. compile 'com.squareup.okhttp3:mockwebserver:3.4.1'

接下来我们实现一个功能,功能大致如下:

  • 客户端和服务端进行建连
  • 连接建立后客户端向服务端发送文本内容command 1
  • 服务器端收到文本内容command 1后返回给客户端内容replay command 1
  • 客户端收到了服务器端返回的replay command 1后再次向服务端发送command 2
  • 服务端收到文本内容command 2后,发送一个ping帧,客户端收到ping帧后会将ping帧内容原封不动的以pong帧返回给服务器端
  • 服务器端收到客户端返回的心跳pong帧后,发送一个关闭连接的控制帧
  • 客户端收到关闭控制帧后关闭连接
  • 服务器端检测到客户端关闭连接,关闭连接

这里需要注意一点,okhttp内部对线程做了检测,也就是收到消息的线程为read线程,那么回复消息不能再read线程中去回复,而要开一个write线程,具体可以看源码,不遵循的话就就会扔异常出来。

  1. if (Thread.currentThread() == looperThread) {
  2. throw new IllegalStateException("attempting to write from reader thread");
  3. }

looperThread就是read线程。

知道了这一点后,我们根据上面的步骤实现一下,首先是server端,使用MockWebServer构造一个mock server对象,顺便new一个线程池,用于write线程回写消息。

  1. private final MockWebServer mockWebServer = new MockWebServer();
  2. private final ExecutorService writeExecutor = Executors.newSingleThreadExecutor();

然后起一个webserver

  1. mockWebServer.enqueue(new MockResponse().withWebSocketUpgrade(new WebSocketListener() {
  2. WebSocket webSocket = null;
  3. @Override
  4. public void onOpen(final WebSocket webSocket, Response response) {
  5. //保存引用,用于后续操作
  6. this.webSocket = webSocket;
  7. //打印一些内容
  8. System.out.println("server onOpen");
  9. System.out.println("server request header:" + response.request().headers());
  10. System.out.println("server response header:" + response.headers());
  11. System.out.println("server response:" + response);
  12. }
  13. @Override
  14. public void onMessage(ResponseBody message) throws IOException {
  15. String string = message.string();
  16. System.out.println("server onMessage");
  17. System.out.println("message:" + string);
  18. //注意下面都是write线程回写给客户端
  19. //当收到客户端的command 1时回复replay command 1
  20. if ("command 1".equals(string)) {
  21. //replay it
  22. writeExecutor.execute(new Runnable() {
  23. @Override
  24. public void run() {
  25. try {
  26. webSocket.message(RequestBody.create(WebSocket.TEXT, "replay command 1"));
  27. } catch (IOException e) {
  28. e.printStackTrace();
  29. }
  30. }
  31. });
  32. } else if ("command 2".equals(string)) {
  33. //当收到客户端的command 2时,发送ping帧
  34. //ping it
  35. writeExecutor.execute(new Runnable() {
  36. @Override
  37. public void run() {
  38. try {
  39. webSocket.ping(ByteString.of("ping from server...".getBytes()));
  40. } catch (IOException e) {
  41. e.printStackTrace();
  42. }
  43. }
  44. });
  45. }
  46. }
  47. @Override
  48. public void onPong(ByteString payload) {
  49. //打印一些内容
  50. System.out.println("server onPong");
  51. //注意下面都是write线程回写给客户端
  52. //客户端收到ping帧后会回复pong帧,回调到这,收到pong帧后关闭连接
  53. //close it
  54. writeExecutor.execute(new Runnable() {
  55. @Override
  56. public void run() {
  57. try {
  58. webSocket.close(1000, "Normal Closure");
  59. } catch (IOException e) {
  60. e.printStackTrace();
  61. }
  62. }
  63. });
  64. }
  65. @Override
  66. public void onClose(int code, String reason) {
  67. //打印一些内容
  68. System.out.println("server onClose");
  69. System.out.println("code:" + code + " reason:" + reason);
  70. }
  71. @Override
  72. public void onFailure(Throwable t, Response response) {
  73. //出现异常会进入此回调
  74. System.out.println("server onFailure");
  75. System.out.println("throwable:" + t);
  76. System.out.println("response:" + response);
  77. }
  78. }));

然后是客户端的实现,也安装上面的步骤来即可。

不过这之前需要知道服务器端的Host和port,这两个值可以通过mockWebServer对象获得。

  1. String hostName = mockWebServer.getHostName();
  2. int port = mockWebServer.getPort();
  3. System.out.println("hostName:" + hostName);
  4. System.out.println("port:" + port);

然后通过host和port构造请求

  1. //新建client
  2. OkHttpClient client = new OkHttpClient.Builder()
  3. .build();
  4. //构造request对象
  5. Request request = new Request.Builder()
  6. .url("ws://" + hostName + ":" + port + "/")
  7. .build();
  8. //new 一个websocket调用对象并建立连接
  9. client.newWebSocketCall(request).enqueue(new WebSocketListener() {
  10. WebSocket webSocket = null;
  11. @Override
  12. public void onOpen(final WebSocket webSocket, Response response) {
  13. //保存引用,用于后续操作
  14. this.webSocket = webSocket;
  15. //打印一些内容
  16. System.out.println("client onOpen");
  17. System.out.println("client request header:" + response.request().headers());
  18. System.out.println("client response header:" + response.headers());
  19. System.out.println("client response:" + response);
  20. //注意下面都是write线程回写给客户端
  21. //建立连接成功后,发生command 1给服务器端
  22. writeExecutor.execute(new Runnable() {
  23. @Override
  24. public void run() {
  25. try {
  26. webSocket.message(RequestBody.create(WebSocket.TEXT, "command 1"));
  27. } catch (IOException e) {
  28. e.printStackTrace();
  29. }
  30. }
  31. });
  32. }
  33. @Override
  34. public void onMessage(ResponseBody message) throws IOException {
  35. //打印一些内容
  36. String string = message.string();
  37. System.out.println("client onMessage");
  38. System.out.println("message:" + string);
  39. //注意下面都是write线程回写给客户端
  40. if ("replay command 1".equals(string)) {
  41. //收到服务器返回的replay command 1后继续向服务器端发送command 2
  42. //replay it
  43. writeExecutor.execute(new Runnable() {
  44. @Override
  45. public void run() {
  46. try {
  47. webSocket.message(RequestBody.create(WebSocket.TEXT, "command 2"));
  48. } catch (IOException e) {
  49. e.printStackTrace();
  50. }
  51. }
  52. });
  53. }
  54. }
  55. @Override
  56. public void onPong(ByteString payload) {
  57. //打印一些内容
  58. System.out.println("client onPong");
  59. System.out.println("payload:" + payload);
  60. }
  61. @Override
  62. public void onClose(int code, String reason) {
  63. //打印一些内容
  64. System.out.println("client onClose");
  65. System.out.println("code:" + code + " reason:" + reason);
  66. }
  67. @Override
  68. public void onFailure(Throwable t, Response response) {
  69. //发生错误时会回调到这
  70. System.out.println("client onFailure");
  71. System.out.println("throwable:" + t);
  72. System.out.println("response:" + response);
  73. }
  74. });

最终输出如下图所示

除了文本内容外,也可以发送二进制内容,如图像,语音,视频等,所以我们完全可以自定义发送的内容。

webSocket.message(RequestBody.create(WebSocket.BINARY,bytes));

而除了okhttp外,我们也可以使用Java-Websocket库来实现,其maven依赖如下

  1. <dependency>
  2. <groupId>org.java-websocket</groupId>
  3. <artifactId>Java-WebSocket</artifactId>
  4. <version>1.3.0</version>
  5. </dependency>

gradle依赖如下

compile 'org.java-websocket:Java-WebSocket:1.3.0'

用法也和okhttp类似,具体细节不追究,大概给一个demo,开启一个mock server可以使用WebSocketServer对象,因为run了一个server只会会循环阻塞当前线程,所以我们在子线程中run。

  1. private final ExecutorService executorService = Executors.newSingleThreadExecutor();
  2. try {
  3. executorService.execute(new Runnable() {
  4. @Override
  5. public void run() {
  6. WebSocketServer webSocketServer = new WebSocketServer(new InetSocketAddress("localhost", 8080)) {
  7. @Override
  8. public void onOpen(WebSocket webSocket, ClientHandshake clientHandshake) {
  9. System.out.println("server onOpen");
  10. }
  11. @Override
  12. public void onClose(WebSocket webSocket, int i, String s, boolean b) {
  13. System.out.println("server onClose:" + i + " " + s + " " + b);
  14. }
  15. @Override
  16. public void onMessage(WebSocket webSocket, String s) {
  17. System.out.println("server onMessage:" + s);
  18. }
  19. @Override
  20. public void onError(WebSocket webSocket, Exception e) {
  21. System.out.println("server onMessage:" + e);
  22. }
  23. };
  24. webSocketServer.run();
  25. }
  26. });
  27. } catch (Exception e) {
  28. e.printStackTrace();
  29. }

然后客户端可以使用WebSocketClient对象

  1. private final ExecutorService executorService = Executors.newSingleThreadExecutor();
  2. executorService.execute(new Runnable() {
  3. @Override
  4. public void run() {
  5. Map<String, String> headers = new HashMap();
  6. WebSocketClient webSocketClient = new WebSocketClient(URI.create("ws://localhost:8080/"), new Draft_17(), headers, 10) {
  7. @Override
  8. public void onOpen(ServerHandshake serverHandshake) {
  9. System.out.println("client onOpen");
  10. }
  11. @Override
  12. public void onMessage(String s) {
  13. System.out.println("client onMessage:" + s);
  14. }
  15. @Override
  16. public void onClose(int i, String s, boolean b) {
  17. System.out.println("client onClose:" + i + " " + s + " " + b);
  18. }
  19. @Override
  20. public void onError(Exception e) {
  21. System.out.println("client onError:" + e);
  22. }
  23. };
  24. webSocketClient.connect();
  25. }
  26. });

知道了如何使用之后,我们来深究一些okhttp内部是怎么实现的WebSocket协议,其内部定义了三个接口,首先是WebSocket接口,用于实现发送消息帧,ping检测心跳,close关闭连接,其内部还定义了两个常量,用于发送不同类型的帧。

  1. public interface WebSocket {
  2. //文本帧时使用
  3. MediaType TEXT = MediaType.parse("application/vnd.okhttp.websocket+text; charset=utf-8");
  4. //二进制帧时使用
  5. MediaType BINARY = MediaType.parse("application/vnd.okhttp.websocket+binary");
  6. void message(RequestBody message) throws IOException;
  7. void ping(ByteString payload) throws IOException;
  8. void close(int code, String reason) throws IOException;
  9. }

接着是WebSocketListener接口,用于进行各种回调,如建立连接成功时的回调,收到消息帧时的回调,收到Pong帧时的回调,关闭连接时的回调,以及连接过程中发生任何错误的回调,其定义如下:

  1. public interface WebSocketListener {
  2. void onOpen(WebSocket webSocket, Response response);
  3. void onMessage(ResponseBody message) throws IOException;
  4. void onPong(ByteString payload);
  5. void onClose(int code, String reason);
  6. void onFailure(Throwable t, Response response);
  7. }

最后一个是类似Http请求时OkHttp返回的Call对象,定义了几个方法,如获取request对象,异步请求,取消连接,判断是否已经执行过,是否已经被取消了,以及一个clone方法,返回一个可被重新执行的WebSocketCall对象,此外,内部还定义了一个Factory接口,该接口被OkHttpClient所实现,用于返回一个WebSocketCall对象,从而建立WebSocket连接。

  1. public interface WebSocketCall extends Cloneable {
  2. Request request();
  3. void enqueue(WebSocketListener listener);
  4. void cancel();
  5. boolean isExecuted();
  6. boolean isCanceled();
  7. WebSocketCall clone();
  8. interface Factory {
  9. WebSocketCall newWebSocketCall(Request request);
  10. }
  11. }

OkHttpClient内部实现的Factory接口中的方法如下,返回了WebSocketCall的实现类RealWebSocketCall。

  1. public WebSocketCall newWebSocketCall(Request request) {
  2. return new RealWebSocketCall(this, request);
  3. }

在RealWebSocketCall构造函数中,主要做一件事情,就是构造请求协议升级的请求。必须是Get请求,然后生成一个随机数,进行base64编码,设置为请求头Sec-WebSocket-Key的值,OkHttp内部实现的WebSocket版本是13,所以添加请求头Sec-WebSocket-Version=13

  1. RealWebSocketCall(OkHttpClient client, Request request) {
  2. this(client, request, new SecureRandom());
  3. }
  4. RealWebSocketCall(OkHttpClient client, Request request, Random random) {
  5. if (!"GET".equals(request.method())) {
  6. throw new IllegalArgumentException("Request must be GET: " + request.method());
  7. }
  8. this.random = random;
  9. byte[] nonce = new byte[16];
  10. random.nextBytes(nonce);
  11. key = ByteString.of(nonce).base64();
  12. client = client.newBuilder()
  13. .readTimeout(0, SECONDS) // i.e., no timeout because this is a long-lived connection.
  14. .writeTimeout(0, SECONDS) // i.e., no timeout because this is a long-lived connection.
  15. .protocols(ONLY_HTTP1)
  16. .build();
  17. originalRequest = request;
  18. request = request.newBuilder()
  19. .header("Upgrade", "websocket")
  20. .header("Connection", "Upgrade")
  21. .header("Sec-WebSocket-Key", key)
  22. .header("Sec-WebSocket-Version", "13")
  23. .build();
  24. call = new RealCall(client, request, true /* for web socket */);
  25. }

当我们调用enqueue方法异步进行连接时,就会发送构造函数里构造的http升级协议请求,当服务器端返回响应体时,进行解析,获得StreamWebSocket对象。

  1. StreamWebSocket create(Response response, WebSocketListener listener) throws IOException {
  2. if (response.code() != 101) {
  3. throw new ProtocolException("Expected HTTP 101 response but was '"
  4. + response.code()
  5. + " "
  6. + response.message()
  7. + "'");
  8. }
  9. String headerConnection = response.header("Connection");
  10. if (!"Upgrade".equalsIgnoreCase(headerConnection)) {
  11. throw new ProtocolException(
  12. "Expected 'Connection' header value 'Upgrade' but was '" + headerConnection + "'");
  13. }
  14. String headerUpgrade = response.header("Upgrade");
  15. if (!"websocket".equalsIgnoreCase(headerUpgrade)) {
  16. throw new ProtocolException(
  17. "Expected 'Upgrade' header value 'websocket' but was '" + headerUpgrade + "'");
  18. }
  19. String headerAccept = response.header("Sec-WebSocket-Accept");
  20. String acceptExpected = Util.shaBase64(key + WebSocketProtocol.ACCEPT_MAGIC);
  21. if (!acceptExpected.equals(headerAccept)) {
  22. throw new ProtocolException("Expected 'Sec-WebSocket-Accept' header value '"
  23. + acceptExpected
  24. + "' but was '"
  25. + headerAccept
  26. + "'");
  27. }
  28. String name = response.request().url().redact().toString();
  29. ThreadPoolExecutor replyExecutor =
  30. new ThreadPoolExecutor(1, 1, 1, SECONDS, new LinkedBlockingDeque<Runnable>(),
  31. Util.threadFactory(Util.format("OkHttp %s WebSocket Replier", name), true));
  32. replyExecutor.allowCoreThreadTimeOut(true);
  33. StreamAllocation streamAllocation = call.streamAllocation();
  34. streamAllocation.noNewStreams(); // Web socket connections can't be re-used.
  35. return new StreamWebSocket(streamAllocation, random, replyExecutor, listener, response, name);
  36. }

如果服务器端返回的http code不是101,则表示升级协议失败,扔出异常,然后会检测响应头中是否包含Connection,且对应的值是否是Upgrade,再判断响应头中是否包含Upgrade,且其值为websocket,如果不满足条件,扔出异常,然后获取响应头中的Sec-WebSocket-Accept值,进行校验,是否和预期的值是一样。其计算方式就是构造函数中生成的随机数的base64的值加上WebSocket的魔数258EAFA5-E914-47DA-95CA-C5AB0DC85B11,进行sha1后的base64值。然后构造StreamWebSocket对象返回。

返回后调用 webSocket.loopReader();方法进行循环。该方法首先会调用回调接口中的onOpen方法告诉调用者建立连接成功了,然后不断读取消息帧。读取消息帧的流程就是解析文章中最开始贴的图中的协议内容。

  1. public final void loopReader() {
  2. looperThread = Thread.currentThread();
  3. try {
  4. try {
  5. readerListener.onOpen(this, response);
  6. } catch (Throwable t) {
  7. Util.throwIfFatal(t);
  8. replyToReaderError(t);
  9. readerListener.onFailure(t, null);
  10. return;
  11. }
  12. while (processNextFrame()) {
  13. }
  14. } finally {
  15. looperThread = null;
  16. }
  17. }

如读取到控制帧时会根据不同的opcode回调接口中的对应函数

  1. switch (opcode) {
  2. case OPCODE_CONTROL_PING:
  3. frameCallback.onReadPing(buffer.readByteString());
  4. break;
  5. case OPCODE_CONTROL_PONG:
  6. frameCallback.onReadPong(buffer.readByteString());
  7. break;
  8. case OPCODE_CONTROL_CLOSE:
  9. int code = CLOSE_NO_STATUS_CODE;
  10. String reason = "";
  11. long bufferSize = buffer.size();
  12. if (bufferSize == 1) {
  13. throw new ProtocolException("Malformed close payload length of 1.");
  14. } else if (bufferSize != 0) {
  15. code = buffer.readShort();
  16. reason = buffer.readUtf8();
  17. validateCloseCode(code, false);
  18. }
  19. frameCallback.onReadClose(code, reason);
  20. closed = true;
  21. break;
  22. default:
  23. throw new ProtocolException("Unknown control opcode: " + toHexString(opcode));
  24. }

当读到ping帧时,会将原数据以pong帧返回

  1. @Override public final void onReadPing(ByteString buffer) {
  2. replyToPeerPing(buffer);
  3. }
  4. /** Replies with a pong when a ping frame is read from the peer. */
  5. private void replyToPeerPing(final ByteString payload) {
  6. Runnable replierPong = new NamedRunnable("OkHttp %s WebSocket Pong Reply", name) {
  7. @Override protected void execute() {
  8. try {
  9. writer.writePong(payload);
  10. } catch (IOException t) {
  11. Platform.get().log(INFO, "Unable to send pong reply in response to peer ping.", t);
  12. }
  13. }
  14. };
  15. synchronized (replier) {
  16. if (!isShutdown) {
  17. replier.execute(replierPong);
  18. }
  19. }
  20. }

当读到pong帧时,直接回调

  1. public final void onReadPong(ByteString buffer) {
  2. readerListener.onPong(buffer);
  3. }

当读到close帧时,也是直接回调

  1. @Override public final void onReadClose(int code, String reason) {
  2. replyToPeerClose(code, reason);
  3. readerSawClose = true;
  4. readerListener.onClose(code, reason);
  5. }

再者读到消息帧的时候,就会读取payload data中的数据,回调frameCallback.onReadMessage方法,返回数据。

  1. private void readMessageFrame() throws IOException {
  2. final MediaType type;
  3. switch (opcode) {
  4. case OPCODE_TEXT:
  5. type = WebSocket.TEXT;
  6. break;
  7. case OPCODE_BINARY:
  8. type = WebSocket.BINARY;
  9. break;
  10. default:
  11. throw new ProtocolException("Unknown opcode: " + toHexString(opcode));
  12. }
  13. final BufferedSource source = Okio.buffer(framedMessageSource);
  14. ResponseBody body = new ResponseBody() {
  15. @Override public MediaType contentType() {
  16. return type;
  17. }
  18. @Override public long contentLength() {
  19. return -1;
  20. }
  21. @Override public BufferedSource source() {
  22. return source;
  23. }
  24. };
  25. messageClosed = false;
  26. frameCallback.onReadMessage(body);
  27. if (!messageClosed) {
  28. throw new IllegalStateException("Listener failed to call close on message payload.");
  29. }
  30. }

frameCallback.onReadMessage会回调到RealWebSocket中的onReadMessage,最终回调给监听器

  1. @Override public final void onReadMessage(ResponseBody message) throws IOException {
  2. readerListener.onMessage(message);
  3. }

同理,回复消息帧则是读取消息帧的逆过程,具体流程,有兴趣自己看源码把~

本站仅提供存储服务,所有内容均由用户发布,如发现有害或侵权内容,请点击举报
打开APP,阅读全文并永久保存 查看更多类似文章
猜你喜欢
类似文章
【热】打开小程序,算一算2024你的财运
Android之OkHttp详解(非原创)
java.lang.IllegalStateException: closed
Android Volley OkHttp3 Gson(Jackson)的封装过程
Android网络编程(六)OkHttp3用法全解析
RxJava Retrofit框架Demo(一)
HTTP客户端连接,选择HttpClient还是OkHttp?
更多类似文章 >>
生活服务
热点新闻
分享 收藏 导长图 关注 下载文章
绑定账号成功
后续可登录账号畅享VIP特权!
如果VIP功能使用有故障,
可点击这里联系客服!

联系客服