Golang Websocket 简易实现+Socket.io演

2019-12-16  本文已影响0人  坚果jimbowhy

造轮子 Websocket 现在就 Go

MD: 2019‎年‎12‎月17‎日，‏‎03:45:10
https://github.com/jimboyeah/demo

笔者坚果有幸从事软件开发，一直都是兴趣驱动的工作。第一次接触计算机是 1999 年后的事，我用来学习的电脑是大哥买来准备学 CAD 的 486 机，当时 CPU 还是威胜的 333MHz 主频，硬盘也只有 4GB，系统是 Windows 98 SE。那时所谓的学电脑纯属拆玩具模式，因为手上可用的资源不多，网络也不发达，也没有太丰富的参考资料，相关的图书也不是太丰富。所以翻查硬盘或系统光盘文件成了日常活动，除此之外，DOS 游戏也和红白机具有一样的可玩性。彼时，BAT 脚本和 Windows 系统光盘中 QBasic 脚本编程工具成了不错的好玩具。后来玩起了 Visual Studio 6.0，主要是 VB 和 VBA 脚本编程，C 语言也开始了解一些，C++ 几乎没有基础可言，所以 Visual C++ 一直玩不动 MFC 什么的更是不知云云。当然了，集成开发工具提供的最大的好处也就体现在这，即使你是个傻瓜也能毫不費力地运行配置好的模板程序，编译成完整的可运行程序。不知不觉，坚果从曾经的傻瓜程序员一路走到今天，没有兴趣带领还真不会有今天。

[TOC]

内容提要

这是我二进 GitChat 的创造，距离今年 ‎6 ‎月分‎第一次发布《从 JavaScript 入门到 Vue 组件实践》大谈前端技术全局观、30' JavaScript 入门课，还有 VSCode 和 Sublime 这些好用的开发工具。到如今已经有近半年时间，‎期间经历了较大的工作变动，技术上已经以脚本后端转到 Golang 为主，这是一种我一直期待的语言。期间也学到一些技术领域比较不容易学习到的知识，有项目管理层面的，有职业规划方面的，对知识付费时代也有了更深入的理解。

那么 Golang 作为一款以便利的并发编程的语言，用在后端的开发真的是不要太好。

Golang 虽然它已经有 10 岁大了，最早接触也是 2012 年左右，但是真正花心思学起来是今年的 7 月份。Golang 号称 21 世纪的 C 语言，这确实是对我最大的吸引力，它的特点可以总结为 C + OOP，以松散组合的方式去实现面向对象的编程思维。完全不像 C++ 把对象数据模型设计的异常复杂，把一种编程语言搞得自己发明人都不能完全掌握。当然每种语言都有它的适用领域及特点，免不了一堆人贬低 Golang 没有泛型之类，确实 Golang 1.x 就是没有提供实现。如日中天的 Python 就是个典型，作为奇慢的脚本解析型语言，慢这个缺点完全掩盖不了它中人工智能算法领域的应用，也完全阻挡不了爬虫一族赖以为生。这种取舍其实就是一种效益的体现，选择恰当的工具做适合的事！

我们将从网络协议层面来打开 Golang 编程大门，学习关于 Websocket 网络协议的相关知识点，在 TCP/IP 协议栈中，新加入的 Websocket 分量也是重量级的。WebSocket 作为实时性要求较高场合的应用协议，主要应用在在线网页聊天室、页游行业等等。掌握 Websocket 技能，你值得拥有！

在本轮学习中，你可以 Get 到技能：

从任天堂红白机时代接触单片机，尽管那时不懂却被深深吸引了；从 MS-DOS 时代结缘计算机，就这样一路披荆斩棘前行；很多人说 IT 人是吃青春饭的，对于我，一个 80 后，在乳臭未干的时候就闻到这饭香了，到现在也没觉得吃够吃厌倦了。我只当这是个兴趣，现在这个爱好还能给我带来一份收入而已。

by Jeangowhy 微信同名(jimboyeah◉gmail.com)
Tue Dec 17 2019 04:23:08 GMT+0800 (深圳宝安)

gitchat

websocket

The WebSocket Protocol https://tools.ietf.org/html/rfc6455

WebSocket vs Polling

先理解几个概念

建立 TCP 连接后，在数据传输完成时还保持 TCP 连接不断开，不发RST包、不进行四次握手断开，并等待对方继续用这个 TCP 通道传输数据，相反的就是短连接。通常 HTTP 连接就是短连接，浏览器建立 TCP 连接请求页面，服务器发送数据，然后关闭连接。下次再需要请求数据时又重新建立 TCP 连接，一问一答是短连接的一个特点。而新的 HTTP 2.0 规范中，为了提高性能则使用了长连接来复用同一个 TCP 连接来传送不同的文件数据。

参考 RFC 2616 HTTP 1.1 规范文档关于 Persistent connection 的部分 https://tools.ietf.org/html/rfc2616#page-44

HTTP 头信息 Connection: Keep-alive 是 HTTP 1.0 浏览器和服务器的实验性扩展，当前的 HTTP 1.1 RFC 2616 文档没有对它做说明，因为它所需要的功能已经默认开启，无须带着它，但是实践中可以发现，浏览器的报文请求都会带上它。如果不希望使用长连接，则要在请求报文首部加上 Connection: close。

《HTTP权威指南》提到，有部分古老的 HTTP 1.0 代理不理解 Keep-alive，而导致长连接失效：客户端-->代理-->服务端，客户端带有 Keep-alive，而代理不认识，于是将报文原封不动转给了服务端，服务端响应了 Keep-alive，也被代理转发给了客户端，于是保持了 客户端-->代理 连接和 代理-->服务端 连接不关闭，但是，当客户端第发送第二次请求时，代理会认为当前连接不会有请求了，于是忽略了它，长连接失效。书上也介绍了解决方案：遇到 HTTP 1.0 就忽略 Keep-alive，客户端就知道当前不该使用长连接。

使用了HTTP长连接（HTTP persistent connection ）之后的好处，包括可以使用HTTP 流水线技术（HTTP pipelining，也有翻译为管道化连接），它是指，在一个TCP连接内，多个HTTP请求可以并行，下一个HTTP请求在上一个HTTP请求的应答完成之前就发起。

Client 和 Server 间的实时数据传输是一个很重要的需求，但早期 HTTP 只能通过 AJAX 轮询 Pooling 方式实现，客户端定时向服务器发送 Ajax 请求，服务器接到请求后马上返回响应信息并关闭连接，这就时短连接的应用。轮询带来以下问题：

长轮询 LongPolling 是基于长连接实现的，是对 Polling 的一种改进。Client 发送请求，此时 Server 可以发送数据或等待数据准备好：

LongPolling 的方式虽然在某种程度上减少了网络带宽和 CPU 利用率等问题，但仍存在缺陷，因为 LongPolling 还是基于一问一答的 HTTP 协议模式。当 Server 的数据更新速度较快，Server 在传送一个数据包给 Client 后必须等待下一个 HTTP 请求到来，才能传递第二个更新的数据包给 Browser，这种场景在 HTTP 上实现的实时聊天几多人游戏是常见的。这样的话，Browser 显示实时数据最快的时间为 2 倍 RTT 往返时间。还不考虑网络拥堵的情况，这个应该是不能让用户接受的。另外，由于 HTTP 数据包的头部数据量很大，通常有 400 多个字节，但真正被服务器需要的数据却很少，可能只有 10个字节左右，这样的数据包在网络上周期性传输，难免对网络带宽是一种浪费。

WebSocket 正是基于支持客户端和服务端的双向通信、简化协议头这些需求下，基于 HTTP 和 TCP 的基础上登上了 Web 的舞台。由于 HTTP 协议的原设计不是用来做双向通讯的，它只是一问一答的执行，客户端发送请求，服务器进行答复，客服端需要什么文件，服务器就提供文件数据。

WebSocket 通信协议是一种双向通信协议，在建立连接后，WebSocket 服务器和 Client 都能主动的向对象发送或接收数据，就像 Socket 一样，所以建立在 Web 基础上的 WebSocket 需要通过升级 HTTP 连接来实现类似 TCP 那样的握手连接，连接成功后才能相互通信。相互沟通的 Header 很小，大概只有 2Bytes。服务器不再被动的接收到浏览器的请求之后才返回数据，而是在有新数据时就主动推送给浏览器。

WebSocket 协议本质上是一个基于 TCP 的协议。为了建立一个 WebSocket 连接，客户端浏览器首先要向服务器发起一个 HTTP 请求，这个请求和通常的 HTTP 请求不同，包含了一些附加头信息，其中附加头信息 Upgrade: WebSocket 表明这是一个申请协议升级的 HTTP 请求，服务器端解析这些附加的头信息然后产生应答信息返回给客户端，客户端和服务器端的 WebSocket 连接就建立起来了，双方就可以通过这个连接通道自由的传递信息，并且这个连接会持续存在直到客户端或者服务器端的某一方主动的关闭连接。

因为 WebSocket 是一种新的通信协议，目前还是草案，没有成为标准，市场上也有成熟的实现 WebSocket 协议开源 Library 可供使用。例如 PyWebSocket、 WebSocket-Node、 LibWebSockets 等。

本文介绍内容大致如下：

Websocket 协议通信分为两个部分，先是握手，再是数据传输。 主要是建立连接握手 Opening Handshake，断开连接握手 Closing Handshake 则简单地利用了 TCP closing handshake (FIN/ACK)。

如下就是一个基本的 Websocket 握手的请求与回包:

Open handshake 请求

GET /chat HTTP/1.1
Host: server.example.com
Origin: http://example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Version: 13

Open handshake 响应

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: chat

Websocket 需要使用到的附加信息头主要有以下几个：

Websocket协议中如何确保客户端与服务端接收到握手请求呢？ 这里就要说到HTTP的两个头部字段，Sec-Websocket-Key 与 Sec-Websocket-Accept。

更详细的说明可以看 RFC 6455 文档。

Data Framing 数据帧

根据 RFC 6455 定义，websocket 消息统称为 messages，可以由多个帧 frame 构成。有文本数据帧，二进制数据帧，控制帧三种，Websocket 官方定义有 6 种类型并预留了 10 种类型用于未来的扩展。

了解完 websocket 握手的大致过程后，这个部分介绍下 Websocket 数据帧与分片传输的方式，主要头部信息是前面的 2 byte。

  0                   1                   2                   3
  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 +-+-+-+-+-------+-+-------------+-------------------------------+
 |F|R|R|R| opcode|M| Payload len |    Extended payload length    |
 |I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
 |N|V|V|V|       |S|             |   (if payload len==126/127)   |
 | |1|2|3|       |K|             |                               |
 +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
 |     Extended payload length continued, if payload len == 127  |
 + - - - - - - - - - - - - - - - +-------------------------------+
 |     and more continued        |Masking-key, if MASK set to 1  |
 +-------------------------------+-------------------------------+
 | Masking-key (continued)       |          Payload Data         |
 +-------------------------------- - - - - - - - - - - - - - - - +
 :                     Payload Data continued ...                :
 + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
 |                     Payload Data continued ...                |
 +---------------------------------------------------------------+

当一个完整消息体大小不可知时，Websocket 支持分片传输 Fragmentation。这样可以方便服务端使用可控大小的 buffer 来传输分段数据，减少带宽压力，同时可以有效控制服务器内存。

同时在多路传输的场景下，可以利用分片技术使不同的 namespace 的数据能共享对外传输通道。不用等待某个大的 message 传输完成，进入等待状态。

对于控制数据帧 Control Frames 不能使用分片方式，并且 Playload 数据不大于 125 bytes，但可以在分片帧中插队传输。通过 opcodes 最高位置位来标记控制帧，0x8 (Close), 0x9 (Ping), 0xA (Pong)，Opcodes 0xB-0xF 保留。

fragmentation 分片规则:

扩展特性支持 Extensibility，Websocket 协议设计考虑了扩展特性支持，通过 Sec-WebSocket-Extensions 来协商需要支持的特性，客户端需要服务器支持的扩展特性添加到这个信息头中。服务端接收到后进行确认，如果支持某个特性，就通过这个信息头返回告诉客服端是支持的特性。常用的有信息压缩扩展有消息压缩 permessage-deflate，对消息进行压缩，deflate 就是给气球放气的意思，也是压缩算法名称，表示压缩很形象。如果这个消息需要分片发送，那么就对压缩过的数据进行分割发送，接收时先拼接在解压缩。

比如客户端查询时发送信息头如下，表示和服务器协商一下压缩扩展：

Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits

那么服务器如果支持消息压缩扩展功能，那么协商返回结果如下：

Sec-WebSocket-Extensions: permessage-deflate
Sec-WebSocket-Extensions: permessage-deflate; server_no_context_takeover; client_no_context_takeover

permessage-deflate 有四个配置参数，配置两端的两个工作方式 no_context_takeover、 max_window_bits。后者配置 DEFLATE 压缩算法的滑动窗口参数，参考 RFC 7692 关于术语 LZ77 sliding window：

以下是关于扩展特性的不规范也不完整可能的应用建议：
Below are some anticipated uses of extensions. This list is neither complete nor prescriptive.

关于压缩算法相关的参考材料：

完整的 Websocket 实现可以参考 Gorilla websocket 包原代码。

Data Frame Examples

telnet 工具的使用

TCP 网络世界，一切皆Socket！事实也是，现在的网络编程几乎都是用的socket。

Telnet 是一个标准的 TCP 协议应用程序，通过它可以建立 TCP 连接，然后发送制符串内容到服务器，这样就可以利用它来模块 HTTP 的数据传输。类似的邮件协议、 FTP 也能模拟的。

用 Telnet 来模拟 HTTP 请求，首先需要了解 HTTP 请求的数据结构，为了简化只采用最简单的 HTTP 协议头，只有 GET 动作及 URL 路径：

GET /chat/telnet HTTP/1.1
Host: localhost

那么现在就执行 telnet 连接到服务器，以连接 localhost 为例，一旦连接后所有敲到命令界面的字符都会实时发送到服务器的接收缓存区中，包括回车符也是。在命令界面中将以上内容敲进去，回车两次表示 HTTP 头部的结束标志，接着等待服务器回复，也可以先复制好这些 HTTP 协议头信息粘贴到 telnet 命令界面中：

telnet localhost 8000
GET /socket.io/ HTTP/1.1
Host: localhost
Connection: Upgrade
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits
Sec-WebSocket-Key: HRaJmux1hUnhnw4iNlYCYA==
Sec-WebSocket-Version: 13
Upgrade: websocket

telnet localhost 8000

GET /chat/telnet HTTP/1.1
Host: localhost

利用快捷键 Ctrl+] 打开命令控制界面，这样就可以通过工具主界面来操作各中选项，也可以看见输入的内容：

GET /IO HTTP/1.1ft Telnet Client
HOST: localhost
Escape 字符为 'CTRL+]'

Microsoft Telnet> ?

命令可能是缩写。支持的命令为:

c    - close                    关闭当前连接
d    - display                  显示操作参数
o    - open hostname [port]     连接到主机(默认端口 23)。
q    - quit                     退出 telnet
set  - set                      设置选项(键入 'set ?' 获得列表)
sen  - send                     将字符串发送到服务器
st   - status                   打印状态信息
u    - unset                    解除设置选项(键入 'set ?' 获得列表)
?/h  - help                     打印帮助信息

Microsoft Telnet> set ?
bsasdel         发送 Backspace 键作为删除
crlf            换行模式 - 引起 return 键发送 CR 和 LF
delasbs         发送 Delete 键作为退格
escape x        x 是进入 telnet 客户端提示的 escape 字符
localecho       打开 localecho
logfile x       x 是当前客户端的日志文件
logging         启用日志
mode x          x 是控制台或流
ntlm            启用 NTLM 身份验证
term x          x 是 ansi、vt100、vt52 或 vtnt

在 Telnet 控制界面直接回车回到交互界面，或者使用 send 命令发送数据，多按几次回车就好：

Microsoft Telnet> send apple
发送字符串 apple

Websocket Demo 握手及数据帧的收发

可以使用辅助调试工具 Fiddler 来抓取 Websocket 数据包用于调试分析，在主界面的连接列表中双击 WS 标记的连接即可在右侧数据面板中看到 Websocket 连接传输的数据。

为了简化，使用 MASK，约定载荷最大为 125 字节，不使用分包发送，不使用压缩扩展，即不返回 Sec-WebSocket-Extensions 的压缩相关扩展如 permessage-deflate。

以 Golang 为例，结合 engine.io-protocol、 socket.io-protocol 来实现一个简化版 Websocket 服务，参考 gorilla websocket 的实现：

go get github.com/gorilla/websocket

Gorilla WebSocket compared with other packages

Features github.com/gorilla golang.org/x/net
Passes Autobahn Test Suite Yes No
Receive fragmented message Yes No, see note 1
Send close message Yes No
Send pings and receive pongs Yes No
Get the type of a received data message Yes Yes, see note 2
Compression Extensions Experimental No
Read message using io.Reader Yes No, see note 3
Write message using io.WriteCloser Yes No, see note 3

Notes:

  1. Large messages are fragmented in Chrome's new WebSocket implementation.
  2. The application can get the type of a received data message by implementing
    a Codec marshal
    function.
  3. The go.net io.Reader and io.Writer operate across WebSocket frame boundaries.
    Read returns when the input buffer is full or a frame boundary is
    encountered. Each call to Write sends a single frame message. The Gorilla
    io.Reader and io.WriteCloser operate on a single WebSocket message.

简易服务器借用了 Gorilla Websocket 的前端示例，如果需要测试 Engine.io 或 Socket.io 通讯，请按《在 Go 中使用 Socket.IO》文章里的页面提供的代码进行修改使用
《在 Go 中使用 Socket.IO》 https://www.jianshu.com/p/566a0e2456a9

package main

import (
    _ "./example"
    "bufio"
    "crypto/sha1"
    "encoding/base64"
    "encoding/binary"
    "encoding/json"
    "errors"
    "fmt"
    "html/template"
    "io"
    stdLog "log"
    "net"
    "net/http"
    "os"
    "runtime"
    "time"
)

const (
    TOO_LONG    = "[payload too long]"
    PingTimeout = time.Second * 5

    // The message types are defined in RFC 6455, section 11.8.
    BrokenMessage     MessageType = 0xf0
    ContinuationFrame MessageType = 0
    TextMessage       MessageType = 1
    BinaryMessage     MessageType = 2
    CloseMessage      MessageType = 8
    PingMessage       MessageType = 9
    PongMessage       MessageType = 10

    // Close codes defined in RFC 6455, section 11.7.
    CloseNormalClosure           MessageCode = 1000
    CloseGoingAway               MessageCode = 1001
    CloseProtocolError           MessageCode = 1002
    CloseUnsupportedData         MessageCode = 1003
    CloseNoStatusReceived        MessageCode = 1005
    CloseAbnormalClosure         MessageCode = 1006
    CloseInvalidFramePayloadData MessageCode = 1007
    ClosePolicyViolation         MessageCode = 1008
    CloseMessageTooBig           MessageCode = 1009
    CloseMandatoryExtension      MessageCode = 1010
    CloseInternalServerErr       MessageCode = 1011
    CloseServiceRestart          MessageCode = 1012
    CloseTryAgainLater           MessageCode = 1013
    CloseTLSHandshake            MessageCode = 1015

    EIO_Close        EngineioPacket = "1"
    EIO_Ping         EngineioPacket = "2"
    EIO_Pond         EngineioPacket = "3"
    EIO_Message      EngineioPacket = "4"
    EIO_Upgrade      EngineioPacket = "5"
    EIO_Noop         EngineioPacket = "6"
    SIO_Connect      SocketioPacket = "0"
    SIO_Disconnect   SocketioPacket = "1"
    SIO_Event        SocketioPacket = "2"
    SIO_Ack          SocketioPacket = "3"
    SIO_Error        SocketioPacket = "4"
    SIO_Binary_Event SocketioPacket = "5"
    SIO_Binary_Ack   SocketioPacket = "6"

    // ErrorCode
    ErrorNil ErrorCode = iota
    ErrorLength
    ErrorKeyLength
    ErrorMessageType
)

type MessageType byte

func (s MessageType) String() string {
    cs := map[MessageType]string{
        0xf0: "BrokenMessage",
        0:    "ContinuationFrame",
        1:    "TextMessage",
        2:    "BinaryMessage",
        8:    "CloseMessage",
        9:    "PingMessage",
        10:   "PongMessage",
    }[s]
    if cs > "" {
        return cs
    } else {
        return fmt.Sprintf("<MessageType v:%d>", s)
    }
}

type EngineioPacket string

func (s EngineioPacket) String() string {
    cs := map[EngineioPacket]string{
        "1": "EIO_Close",
        "2": "EIO_Ping",
        "3": "EIO_Pond",
        "4": "EIO_Message",
        "5": "EIO_Upgrade",
        "6": "EIO_Noop",
    }[s]
    if cs > "" {
        return cs
    } else {
        return fmt.Sprintf("<EngineioPacket v:%q>", s)
    }
}

type SocketioPacket string

func (s SocketioPacket) String() string {
    cs := map[SocketioPacket]string{
        "0": "SIO_Connect",
        "1": "SIO_Disconnect",
        "2": "SIO_Event",
        "3": "SIO_Ack",
        "4": "SIO_Error",
        "5": "SIO_Binary_Event",
        "6": "SIO_Binary_Ack",
    }[s]
    if cs > "" {
        return cs
    } else {
        return fmt.Sprintf("<SocketioPacket v:%q>", s)
    }
}

type MessageCode uint

func (s MessageCode) String() string {
    cs := map[MessageCode]string{
        1000: "CloseNormalClosure",
        1001: "CloseGoingAway",
        1002: "CloseProtocolError",
        1003: "CloseUnsupportedData",
        1005: "CloseNoStatusReceived",
        1006: "CloseAbnormalClosure",
        1007: "CloseInvalidFramePayloadData",
        1008: "ClosePolicyViolation",
        1009: "CloseMessageTooBig",
        1010: "CloseMandatoryExtension",
        1011: "CloseInternalServerErr",
        1012: "CloseServiceRestart",
        1013: "CloseTryAgainLater",
        1015: "CloseTLSHandshake",
    }[s]
    if cs != "" {
        return cs
    } else {
        return fmt.Sprintf("<MessageCode v:%d>", s)
    }
}

type ErrorCode uint

func (s ErrorCode) String() string {
    cs := map[ErrorCode]string{
        ErrorNil:         "ErrorNil",
        ErrorLength:      "ErrorLength",
        ErrorKeyLength:   "ErrorKeyLength",
        ErrorMessageType: "ErrorMessageType",
    }[s]
    if cs != "" {
        return cs
    } else {
        return fmt.Sprintf("<ErrorCode v:%d>", s)
    }
}

var log = stdLog.New(os.Stderr, "[ws]", stdLog.Ltime)

func main() {
    // example.protobuf_test()
    // example.Socket_Run()

    staticweb := http.StripPrefix("/web/", http.FileServer(http.Dir("./")))

    http.Handle("/socket.io/", wsdemo{})
    http.Handle("/web/", staticweb)
    http.HandleFunc("/", home)

    log.SetPrefix("[wsDemo]")
    log.Println("Serving at localhost:8000...")
    log.Fatal(http.ListenAndServe(":8000", nil))
}

func home(w http.ResponseWriter, r *http.Request) {
    homeTemplate.Execute(w, "ws://"+r.Host+"/socket.io/")
}

func printStack() {
    var buf [4096]byte
    n := runtime.Stack(buf[:], false)
    os.Stdout.Write(buf[:n])
}

/*
    RFC 6455 Data Framing 数据帧

      0                   1                   2                   3
      0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
     +-+-+-+-+-------+-+-------------+-------------------------------+
     |F|R|R|R| opcode|M| Payload len |    Extended payload length    |
     |I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
     |N|V|V|V|       |S|             |   (if payload len==126/127)   |
     | |1|2|3|       |K|             |                               |
     +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
     |     Extended payload length continued, if payload len == 127  |
     + - - - - - - - - - - - - - - - +-------------------------------+
     |     and more continued        |Masking-key, if MASK set to 1  |
     +-------------------------------+-------------------------------+
     | Masking-key (continued)       |          Payload Data         |
     +-------------------------------- - - - - - - - - - - - - - - - +
     :                     Payload Data continued ...                :
     + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
     |                     Payload Data continued ...                |
     +---------------------------------------------------------------+

*/
type FrameParser struct {
    Raw          []byte
    parsed       bool
    Final        bool
    Rsv1         bool
    Rsv2         bool
    Rsv3         bool
    Opcode       byte
    Maskingkey   []byte
    Masking      bool
    Type         MessageType
    Code         MessageCode
    ExHeaderSize uint
    Length       uint64
    Header       []byte
    Data         []byte
}

func (c *FrameParser) Detect(header []byte) (msgtype MessageType, exsize uint, err ErrorCode) {
    c.Reset() // ready for a new frame
    if len(header) != 2 {
        return BrokenMessage, 0, ErrorLength
    }
    // 2bytes header
    c.Final = (header[0] | byte(1<<7)) > 0
    c.Rsv1 = (header[0] | byte(1<<6)) > 0
    c.Rsv2 = (header[0] | byte(1<<5)) > 0
    c.Rsv3 = (header[0] | byte(1<<4)) > 0
    c.Opcode = (header[0] & 0x0F)
    c.Type = MessageType(c.Opcode)
    c.Masking = (header[1] & byte(1<<7)) > 0
    c.Length = uint64(header[1] & 0x7f)

    c.ExHeaderSize = 0
    if c.Masking {
        c.ExHeaderSize += 4
    }
    if c.Length == 126 {
        c.ExHeaderSize += 2
    } else if c.Length == 127 {
        c.ExHeaderSize += 8
    }
    c.Raw = append([]byte{}, header...) // copy
    return c.Type, c.ExHeaderSize, ErrorNil
}

func (c *FrameParser) DecideLength(ex []byte) (payloadlen uint64, err ErrorCode) {
    if uint(len(ex)) != c.ExHeaderSize {
        return 0, ErrorLength
    }
    if c.Masking {
        c.Maskingkey = ex[len(ex)-4:]
    }
    if c.Length == 126 {
        c.Length = binary.BigEndian.Uint64(ex[0:2])
    } else if c.Length == 127 {
        c.Length = binary.BigEndian.Uint64(ex[0:8])
    }
    c.Raw = append(c.Raw, ex...) // copy
    return c.Length, ErrorNil
}

func (c *FrameParser) Load(data []byte) ErrorCode {
    if uint64(len(data)) != c.Length {
        return ErrorLength
    }
    c.Raw = append(c.Raw, data...)     // copy
    c.Data = append([]byte{}, data...) // copy
    c.Mask()
    return ErrorNil
}

func (c *FrameParser) Mask() ErrorCode {
    ldata := uint64(len(c.Data))
    lkey := len(c.Maskingkey)
    if !c.Masking && ldata == c.Length {
        return ErrorNil
    } else if !c.Masking && ldata != c.Length {
        return ErrorLength
    } else if c.Masking && lkey != 4 {
        return ErrorKeyLength
    } else if c.Masking {
        for i := uint64(0); i < c.Length; i++ {
            j := i % uint64(lkey)
            b := c.Maskingkey[j] ^ c.Data[i]
            c.Data[i] = b
        }
    }
    return ErrorNil
}

func (c *FrameParser) Reset() {
    c.Raw = []byte{}
    c.parsed = false
    c.Final = false
    c.Rsv1 = false
    c.Rsv2 = false
    c.Rsv3 = false
    c.Opcode = 0
    c.Maskingkey = []byte{}
    c.Masking = false
    c.Type = 0
    c.Code = 0
    c.Length = 0
    c.Header = []byte{}
    c.Data = []byte{}
}

type wsdemo struct {
    w       http.ResponseWriter
    r       *http.Request
    netconn net.Conn
    brw     *bufio.ReadWriter
    who     string
}

func (c wsdemo) read() {
    fm := FrameParser{}
    for {
        p, err := c.peek(2)
        if err != nil {
            break
        } else if len(p) != 2 {
            continue
        }
        log.Printf("Read a 2bytes header: %d [%x] [%x]\n", len(p), p, p)
        // onData...
        msgtype, exsize, ec := fm.Detect(p)
        if ec != ErrorNil {
            continue
        }
        ex, err := c.peek(int(exsize))
        if err != nil {
            break
        } else if uint(len(ex)) != exsize {
            continue
        }
        log.Printf("Read an ex-header [%d] %s: %d [%x] [%x]\n", exsize, msgtype.String(), len(ex), ex, ex)
        payloadlen, _ := fm.DecideLength(ex)
        payload, err := c.peek(int(payloadlen))
        if err != nil {
            break
        }
        log.Printf("Read payload [%d] %s: %d [%x]", payloadlen, msgtype.String(), len(payload), payload)
        fm.Load(payload)
        c.onMessage(fm)
    }
}

func (c wsdemo) ServeHTTP(w http.ResponseWriter, r *http.Request) {
    c.w = w
    c.r = r

    if r.URL.RequestURI() == "/socket.io/socket.io.js" {
        w.Write([]byte(`
        // socket.io.js not provide here, use CDN below:
        // https://cdnjs.com/libraries/socket.io
        // https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.3.0/socket.io.js
        `))
        return
    }

    netConn, brw, _ := c.Upgrade(w, r)
    c.netconn = netConn
    c.brw = brw
    c.who = c.netconn.RemoteAddr().String()

    log.Println("wsdemo.ServeHTTP()...", r.URL.RequestURI(), r.RequestURI)

    Upgrade := r.Header.Get("Upgrade") == "websocket"
    Connection := r.Header.Get("Connection") == "Upgrade"
    if isWebsocketUpgrade := Upgrade && Connection; isWebsocketUpgrade {
        c.OpenHandshake()
        c.EngineioOpen("Engine.io Open")
        go c.read()
    } else {
        // w.WriteHeader(200)
        // w.Write([]byte("Ok"))
        // brw.Write([]byte("Totally-Control: Yes\r\n"))
        // brw.Flush()
        netConn.Write([]byte("HTTP/1.1 200 OK\r\n\r\nThis is an websocket server, use websocket client to connect."))
        netConn.Close()
    }
}

func (c wsdemo) peek(size int) (p []byte, err error) {
    if size == 0 {
        return []byte{}, nil
    }
    c.netconn.SetDeadline(time.Now().Add(30000 * time.Millisecond))
    var brw *bufio.ReadWriter = c.brw
    buf, err := brw.Peek(size)
    brw.Discard(len(buf))
    // buf := []byte{}
    // size, err := brw.Read(buf)
    if err != nil && err == io.EOF { // client disconnect
        c.netconn.Close()
        return buf, err
    } else if err != nil {
        switch err.(type) {
        case *net.OpError:
            var e *net.OpError = err.(*net.OpError)
            log.Printf("Peeking net.OpError [%x] timeout:%t temporary:%t %#+v", p, e.Timeout(), e.Temporary(), e.Err)
            if !(e.Timeout() || e.Temporary()) {
                log.Printf("Client leave %s", c.who)
                c.netconn.Close()
                return buf, err
            }
        default:
            log.Printf("Peeking error %#+v", err)
        }
    }

    return buf, nil
}

func (c wsdemo) read_demo() {
    var brw *bufio.ReadWriter = c.brw
    for {
        c.netconn.SetDeadline(time.Now().Add(30000 * time.Millisecond))
        // size := 2
        // buf, err := brw.Peek(size)
        // brw.Discard(len(buf))

        // buf := []byte{}
        // size, err := brw.Read(buf)

        buf, ok, err := brw.ReadLine() // '\n' as delim and it not include in buf
        size := len(buf)

        // buf, err := brw.ReadBytes('B') // 'B' included in buf or error return
        // size := len(buf)

        msg := fmt.Sprintf("4recv %s [%d]: %x", c.who, size, string(buf))
        // c.TextFrame(msg)
        log.Println(msg, " ok:", ok)
        if err != nil && err == io.EOF { // client disconnect
            c.netconn.Close()
            break
        } else if err != nil {
            log.Printf("Error: %#+v", err)
        }
    }
}

func (c wsdemo) read_unworking() {
    input := bufio.NewScanner(c.netconn)
    buf := []byte{}
    input.Buffer(buf, 2)
    c.netconn.SetDeadline(time.Now().Add(100 * time.Millisecond))
    log.Printf("read by bufio.Scanner...")
    for input.Scan() {
        c.netconn.SetDeadline(time.Now().Add(100 * time.Millisecond))
        msg := fmt.Sprintf("4recv %s: 0x%x", c.who, input.Text())
        c.TextFrame(msg)
        log.Println(msg)
    }
}

func (c wsdemo) onSocketioMessage(data []byte, fm FrameParser) {
    msgtype := SocketioPacket(data[0:1])
    switch msgtype {
    case SIO_Binary_Event:
        packet := []string{}
        json.Unmarshal(data[3:], &packet)
        event := packet[0]
        c.SocketioEventDemo(event, packet[1])
        log.Printf("onSocketio BinaryEvent %s %s raw[%d]", event, packet[1], len(data))
    case SIO_Event:
        packet := []string{}
        json.Unmarshal(data[1:], &packet)
        event := packet[0]
        c.SocketioEventDemo(event, packet[1])
        log.Printf("onSocketio Event %s %s raw[%d]", event, packet[1], len(data))
    default:
        log.Printf("onSocketioPacet %s %s raw[%d]: %x", msgtype.String(), string(data), len(data), data)
    }
}
func (c wsdemo) onEngineioPacket(data []byte, fm FrameParser) {
    msgtype := EngineioPacket(data[0:1])
    log.Printf("onEnginioMessage %s %s raw[%d]: % x", msgtype.String(), string(data), len(data), data)
    if msgtype == EIO_Ping {
        c.EngineioPond("Pong...")
    } else if msgtype == EIO_Message {
        c.onSocketioMessage(data[1:], fm)
    }
}
func (c wsdemo) onMessage(fm FrameParser) {
    // msg := fmt.Sprintf("%s from %s Len:%d 0x%X", fm.Type.String(), c.who, fm.Length, fm.Data)
    // c.TextFrame(msg)
    // log.Println("onMessage and reply:", msg)
    if fm.Type == CloseMessage {
        log.Println("Close connection for " + fm.Type.String() + string(fm.Data))
        c.netconn.Close()
    } else if fm.Type == TextMessage {
        c.onEngineioPacket(fm.Data, fm)
    } else {
        log.Printf("onMessage %s data:%s raw[%d]: % x", fm.Type.String(), string(fm.Data), len(fm.Raw), fm.Raw)
    }
}

func (c wsdemo) computeAcceptKey(challengeKey string) string {
    var keyGUID = []byte("258EAFA5-E914-47DA-95CA-C5AB0DC85B11")
    sn := append([]byte(challengeKey), keyGUID...)
    h := sha1.New()
    h.Write(sn)
    // hash := sha1.Sum(sn)
    return base64.StdEncoding.EncodeToString(h.Sum(nil))
}

func (c wsdemo) TextFrame(payload string) {
    if len(payload) > 125 {
        payload = payload[0:124-len(TOO_LONG)] + TOO_LONG
    }
    len := byte(len(payload))
    frame := []byte{0x80 | byte(TextMessage), len}
    frame = append(frame, []byte(payload)...)
    c.send(frame)
}
func (c wsdemo) BinaryFrame(payload string) {
    if len(payload) > 125 {
        payload = payload[0:124-len(TOO_LONG)] + TOO_LONG
    }
    len := byte(len(payload))
    frame := []byte{0x80 | byte(BinaryMessage), len}
    frame = append(frame, []byte(payload)...)
    c.send(frame)
}
func (c wsdemo) CloseFrame(payload string) {
    if len(payload) > 125 {
        payload = payload[0:124-len(TOO_LONG)] + TOO_LONG
    }
    len := byte(len(payload))
    frame := []byte{0x80 | byte(CloseMessage), len}
    frame = append(frame, []byte(payload)...)
    c.send(frame)
}
func (c wsdemo) PingFrame(payload string) {
    if len(payload) > 125 {
        payload = payload[0:124-len(TOO_LONG)] + TOO_LONG
    }
    len := byte(len(payload))
    frame := []byte{0x80 | byte(PingMessage), len}
    frame = append(frame, []byte(payload)...)
    c.send(frame)
}
func (c wsdemo) PongFrame(payload string) {
    if len(payload) > 125 {
        payload = payload[0:124-len(TOO_LONG)] + TOO_LONG
    }
    len := byte(len(payload))
    frame := []byte{0x80 | byte(PongMessage), len}
    frame = append(frame, []byte(payload)...)
    c.send(frame)
}
func (c wsdemo) send(frame []byte) {
    c.netconn.SetWriteDeadline(time.Now().Add(30 * time.Second))
    c.brw.Write(frame)
    c.brw.Writer.Flush()
    // c.brw.Flush()
}

func (c wsdemo) OpenHandshake() {
    challengeKey := c.r.Header.Get("Sec-WebSocket-Key")
    Extensions := c.r.Header.Get("Sec-WebSocket-Extensions")
    Protocol := c.r.Header.Get("Sec-WebSocket-Protocol")

    p := []byte{}
    p = append(p, "HTTP/1.1 101 Switching Protocols\r\n"...)
    p = append(p, "Upgrade: websocket\r\n"...)
    p = append(p, "Connection: Upgrade\r\n"...)
    if Protocol > "" {
        p = append(p, ("Sec-WebSocket-Protocol: " + Protocol + "\r\n")...)
    }
    if false && Extensions > "" {
        p = append(p, ("Sec-WebSocket-Extensions: permessage-deflate\r\n")...)
    }
    p = append(p, ("Sec-WebSocket-Accept: " + c.computeAcceptKey(challengeKey) + "\r\n")...)

    p = append(p, "\r\n"...)

    // Clear deadlines set by HTTP server.
    c.netconn.SetDeadline(time.Time{})

    log.Printf("OpenHandshake %s...", c.who)
    c.netconn.Write(p)
}

func (c wsdemo) Upgrade(w http.ResponseWriter, r *http.Request) (net.Conn, *bufio.ReadWriter, error) {
    hj, ok := w.(http.Hijacker)
    if !ok {
        msg := "websocket: response does not implement http.Hijacker"
        http.Error(w, msg, http.StatusInternalServerError)
        return nil, nil, errors.New(msg)
    }
    var brw *bufio.ReadWriter
    netConn, brw, err := hj.Hijack()
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return nil, nil, err
    }

    if brw.Reader.Buffered() > 0 {
        netConn.Close()
        msg := "websocket: client sent data before handshake is complete"
        http.Error(w, msg, http.StatusInternalServerError)
        return nil, nil, errors.New(msg)
    }

    return netConn, brw, nil
}

/*
Engine.io API
packet format: <packet type id>[<data>]
for example, a ping message with text: 2probe
single-packet format: <packet1>
multi-packet format: <length1>:<packet1>[<length2>:<packet2>[...]]
message type:
- 0 `open` Sent from the server when a new transport is opened (recheck)
- 1 `close` Request the close of this transport but does not shutdown the connection itself.
- 2 `ping` Sent by the client. Server should answer with a pong packet containing the same data
- 3 `pong` Sent by the server to respond to ping packets.
- 4 `message` actual message, client and server should call their callbacks with the data.
- 5 `upgrade` Before engine.io switches a transport, it tests, if server and client can communicate over this transport.
     If this test succeed, the client sends an upgrade packets which requests the server to flush its cache
     on the old transport and switch to the new transport.
- 6 `noop` A noop packet. Used primarily to force a poll cycle when an incoming websocket connection is received.
*/

func (c wsdemo) EngineioOpen(text string) {
    // Engine.io Open Message
    packet := []byte(fmt.Sprintf(`0{"sid":"client_%s","upgrades":[],"pingInterval":15000,"pingTimeout":5000}`, c.who))
    c.TextFrame(string(packet))
    c.TextFrame("40")
}
func (c wsdemo) EngineioClose(text string) {
    c.TextFrame("1" + text)
}
func (c wsdemo) EngineioPing(text string) {
    c.TextFrame("2" + text)
}
func (c wsdemo) EngineioPond(text string) {
    c.TextFrame("3" + text)
}
func (c wsdemo) EngineioMessage(text string) {
    c.TextFrame("4" + text)
}
func (c wsdemo) EngineioUpgrade(text string) {
    c.TextFrame("5" + text) // ???
}
func (c wsdemo) EngineioNoop(text string) {
    c.TextFrame("6" + text)
}

/*
Socket.io API
- Packet#CONNECT (0)
- Packet#DISCONNECT (1)
- Packet#EVENT (2)
- Packet#ACK (3)
- Packet#ERROR (4)
- Packet#BINARY_EVENT (5)
    50-["protobuf", "CgxoZWxsbyBiaW5hcnkQYxoDQUJDIyoIZ29vZCBieWUk"]
- Packet#BINARY_ACK (6)
*/
func (c wsdemo) SocketioEventDemo(event string, data string) {
    res := fmt.Sprintf("%s received[%d]", event, len(data))
    jsontxt := fmt.Sprintf("{%q:%q,%q:%q}", "userId", c.who, "text", res)
    event = "res"
    c.SocketioEventEmit(event, jsontxt)
}
func (c wsdemo) SocketioEventEmit(event string, text string) {
    // enc := base64.StdEncoding.EncodeToString([]byte(text))
    msg := fmt.Sprintf("2[%q,%s]", event, text)
    c.EngineioMessage(msg)
}

var homeTemplate = template.Must(template.New("").Parse(`
    <!DOCTYPE html>
    <html>
    <head>
    <meta charset="utf-8">
    <script>  
    window.addEventListener("load", function(evt) {

        var output = document.getElementById("output");
        var input = document.getElementById("input");
        var ws;

        var print = function(message) {
            var d = document.createElement("div");
            d.innerHTML = message;
            output.appendChild(d);
        };

        document.getElementById("open").onclick = function(evt) {
            if (ws) {
                return false;
            }
            ws = new WebSocket("{{.}}");
            ws.onopen = function(evt) {
                print("OPEN");
            }
            ws.onclose = function(evt) {
                print("CLOSE");
                ws = null;
            }
            ws.onmessage = function(evt) {
                print("⇩⇧ " + evt.data);
            }
            ws.onerror = function(evt) {
                print("ERROR: " + evt.data);
            }
            return false;
        };

        document.getElementById("send").onclick = function(evt) {
            if (!ws) {
                return false;
            }
            print("☝ " + input.value);
            ws.send(input.value);
            return false;
        };

        document.getElementById("close").onclick = function(evt) {
            if (!ws) {
                return false;
            }
            ws.close();
            return false;
        };

    });
    </script>
    </head>
    <body>
    <table>
    <tr><td valign="top" width="50%">
    <p>1. <b>Open</b> a websocket connection to the server.
    <p>2. <b>Send</b> a message to the server.
    <p>3. <b>Change</b> the message below. 
    <p>4. <b>Close</b> the connection. 
    <p>
    <form>
    <button id="open">Open</button>
    <button id="close">Close</button>
    <p><textarea name="input" id="input" cols="30" rows="10">Hello world!</textarea>
    <finput id="finput" type="text" value="Hello world!">
    <button id="send">Send</button>
    </form>
    </td><td valign="top" width="50%">
    <div id="output"></div>
    </td></tr></table>
    </body>
    </html>
    `))
上一篇下一篇

猜你喜欢

热点阅读