通过LiveGo学习WebSocket
一直对网络直播、在线互动的前端应用特比感兴趣，偶然间在Gitee上看到了livego的开源代码，是基于golang的直播服务实现。代码量不大，于是抽丝剥茧的分析了一下其设计思路，觉得还蛮有意思的。
所以，用户至少具备两个基本属性，用户的id和关联数据推送的WebSocket连接。除了这两个基本信息外，可根据实际业务需要增加其他信息，如用户名、头像等。上线用户的临时组管理无用户认证、用户信息管理等复杂业务逻辑，也没有数据信息的持久化处理。
对于消息的定义有如下字段：消息类型-MType、消息内容-Data和源用户的头像信息-Img。消息类型的串行化处理采用了json.Marshall。
-
1. 直播中的业务处理
-
直播是一端数据上传，多端数据分发的过程。数据的上传和转发都要借助长连接，所以在livego中采用了WebSocket。livego是一个简易的开放式聊天服务，在线群组管理做的比较简单：
-
1) 为请求连接的用户创建唯一标识，若连接成功则在群组中保留；否则将其移除；
-
2) 消息转发过程中，根据数据推送的状态更新群群组状态，如数据推送失败则断开连接，并将用户剔除。
所以，用户至少具备两个基本属性，用户的id和关联数据推送的WebSocket连接。除了这两个基本信息外，可根据实际业务需要增加其他信息，如用户名、头像等。上线用户的临时组管理无用户认证、用户信息管理等复杂业务逻辑，也没有数据信息的持久化处理。
对于消息的定义有如下字段：消息类型-MType、消息内容-Data和源用户的头像信息-Img。消息类型的串行化处理采用了json.Marshall。
2. WebSocket会话
WebSocket会话直播服务的核心基础构件。在livego中自带了一套websocket的封装，协议字段的定义、ws帧解析参考规范文档，draft-ietf-hybi-thewebsocketprotocol-10。这个文档是可以网上找到的，对照文档解读源码，思路会比较清晰易懂。
从WebSocket客户端请求说起。WebSocket请求处理的第一补是解析ws的url，其格式与http的URL格式类似，不同的是把scheme换成了ws或wss，该scheme的处理策略是创建ws连接。处理如下：
1） 根据URL及origin够构造DiagConfig，该配置参数会用于后续的TCP连接的建立和参数配置1
// Dial opens a new client connection to a WebSocket.
func Dial(url_, protocol, origin string) (ws *Conn, err error) {
config, err := NewConfig(url_, origin)
if err != nil {
return nil, err
}
if protocol != "" {
config.Protocol = []string{protocol}
}
return DialConfig(config)
}
Ø 借助golang的net库，创建TCP连接
func dialWithDialer(dialer *net.Dialer, config *Config) (conn net.Conn, err error) {
switch config.Location.Scheme {
case "ws":
conn, err = dialer.Dial("tcp", parseAuthority(config.Location))
case "wss":
conn, err = tls.DialWithDialer(dialer, "tcp", parseAuthority(config.Location), config.TlsConfig)
default:
err = ErrBadScheme
}
return
}
2）借助golang的net库，创建TCP连接
func dialWithDialer(dialer *net.Dialer, config *Config) (conn net.Conn, err error) {
switch config.Location.Scheme {
case "ws":
conn, err = dialer.Dial("tcp", parseAuthority(config.Location))
case "wss":
conn, err = tls.DialWithDialer(dialer, "tcp", parseAuthority(config.Location), config.TlsConfig)
default:
err = ErrBadScheme
}
return
}3）TCP连接创建成功后，构造WS Client
// DialConfig opens a new client connection to a WebSocket with a config.
func DialConfig(config *Config) (ws *Conn, err error) {
var client net.Conn
if config.Location == nil {
return nil, &DialError{config, ErrBadWebSocketLocation}
}
if config.Origin == nil {
return nil, &DialError{config, ErrBadWebSocketOrigin}
}
dialer := config.Dialer
if dialer == nil {
dialer = &net.Dialer{}
}
// client是TCP连接对象
client, err = dialWithDialer(dialer, config)
if err != nil {
goto Error
}
// 基于TCP连接创建ws连接
ws, err = NewClient(config, client)
if err != nil {
client.Close()
goto Error
}
return
Error:
return nil, &DialError{config, err}
}4） rwc是tcp连接对象，并由此构造缓冲读写的对象br，bw。并开始执行WS会话中的关键步骤-握手。握手完成后随即构造WS连接对象。hybiClientHandshake实现是纯HTTP的协议交互，按照HTTP的协议规范构造协议内容，如头部、请求体等，发送给对端，根据对端返回的状态码及必要字段的合法性判断握手过程是否成功。
// NewClient creates a new WebSocket client connection over rwc.
func NewClient(config *Config, rwc io.ReadWriteCloser) (ws *Conn, err error) {
br := bufio.NewReader(rwc)
bw := bufio.NewWriter(rwc)
err = hybiClientHandshake(config, br, bw)
if err != nil {
return
}
buf := bufio.NewReadWriter(br, bw)
ws = newHybiClientConn(config, buf, rwc)
return
}
Ø WS连接创建的过程即为Conn对象的构造和初始化过程，各自字段的含义及备注说明如下：
// newHybiConn creates a new WebSocket connection speaking hybi draft protocol.
func newHybiConn(config *Config, buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) *Conn {
if buf == nil {
br := bufio.NewReader(rwc)
bw := bufio.NewWriter(rwc)
buf = bufio.NewReadWriter(br, bw)
}
ws := &Conn{config: config, request: request, buf: buf, rwc: rwc,
frameReaderFactory: hybiFrameReaderFactory{buf.Reader},
frameWriterFactory: hybiFrameWriterFactory{
buf.Writer, request == nil},
PayloadType: TextFrame,
defaultCloseStatus: closeStatusNormal}
ws.frameHandler = &hybiFrameHandler{conn: ws}
return ws
}
// Conn represents a WebSocket connection.
//
// Multiple goroutines may invoke methods on a Conn simultaneously.
type Conn struct {
config *Config // 连接建立的基础参数配置
request *http.Request // 服务端WS处理用，保存对端WS连接的握手请求对象
buf *bufio.ReadWriter // TCP连接的缓存读写
rwc io.ReadWriteCloser // 关闭连接的接口对象
rio sync.Mutex
frameReaderFactory // 工厂类-构造帧读对象
frameReader // 读对象记录：每次的读都要构造新的读对象，用于处理连接中的数据。该对象记录用于读取上一次读操作后剩余的数据
wio sync.Mutex
frameWriterFactory // 工厂类-构造写对象
frameHandler // 读操作中的中继处理，如连续读中的状态码记录和转换
PayloadType byte // 与conn对象绑定，写数据时默认的数据类型
defaultCloseStatus int
// MaxPayloadBytes limits the size of frame payload received over Conn
// by Codec's Receive method. If zero, DefaultMaxPayloadBytes is used.
MaxPayloadBytes int // 读数据时最大的有效长度，超长直接报错
}
5） 连接建立之后，就是WS帧的读写操作。WS帧的结构为：
|------------- Frame Header -----------------|------------------- PlayLoad ----------------|
FrameHeader是对帧状态、长度等的基本信息描述，对PayLoad的读取起决定作用。
先看读操作
func (ws *Conn) Read(msg []byte) (n int, err error) {
ws.rio.Lock()
defer ws.rio.Unlock()
again:
if ws.frameReader == nil {
// 构造读对象。实现上是从连接中读数据，构造ws帧的描述信息，即Frame Header，包含状态位，帧长度等；
frame, err := ws.frameReaderFactory.NewFrameReader()
if err != nil {
return 0, err
}
// 可能是中间读，所以通过handler校正下PayloadType
ws.frameReader, err = ws.frameHandler.HandleFrame(frame)
if err != nil {
return 0, err
}
if ws.frameReader == nil {
goto again
}
}
// 根据FrameHeader描述，读取连接中的PayLoad数据。如果当前的OperateCode是FinishFrame则表示
// 整个帧读取结束，返回io.EOF
n, err = ws.frameReader.Read(msg)
if err == io.EOF {
if trailer := ws.frameReader.TrailerReader(); trailer != nil {
io.Copy(ioutil.Discard, trailer)
}
ws.frameReader = nil
goto again //再未遇到CloseFrame之前，要多次、连续读
}
return n, err
}
6） 写操作与读类似，读是分头解析和数据读取，那么写自然也是分两部分。值得注意是，写操作是一次性将所有数据单帧写入到连接，不做拆分，每一帧都结束帧，不存在中间连续帧的情况。
// Write implements the io.Writer interface:
// it writes data as a frame to the WebSocket connection.
func (ws *Conn) Write(msg []byte) (n int, err error) {
ws.wio.Lock()
defer ws.wio.Unlock()
// 构造FrameHeader对象，完成Header中描述字段的初始化赋值
w, err := ws.frameWriterFactory.NewFrameWriter(ws.PayloadType)
if err != nil {
return 0, err
}
// 写Header和PayLoad数据
n, err = w.Write(msg)
w.Close()
return n, err
}
至此，WS客户端的WS建立及数据读写过程就说完了。其实WS Client的建立和读写操作对服务同样适用，剩下的是服务端如何处理WS的握手请求并建立WS连接的。首先从http的请求分发开始
1） 服务启动时会注册如下的http请求处理，表示如果是chat请求则走WS的握手过程
func main() {
//fmt.Printf("LiveGoServer is ready...\n")
go func() {
http.Handle("/chat", websocket.Handler(pwint)) //这里是将函数对象强转为Handler类型
}()2） Handler是实现了http.Handler接口的类型，自然在接口的实现中开始处理对端的WS握手请求
// Handler是函数类型别名
type Handler func(*Conn)
// ServeHTTP implements the http.Handler interface for a WebSocket
func (h Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
s := Server{Handler: h, Handshake: checkOrigin}
s.serveWebSocket(w, req)
}3）serveWebSocket是完成握手过程，创建WS连接
func (s Server) serveWebSocket(w http.ResponseWriter, req *http.Request) {
rwc, buf, err := w.(http.Hijacker).Hijack() // 接管http的tcp连接对象
if err != nil {
panic("Hijack failed: " + err.Error())
}
// The server should abort the WebSocket connection if it finds
// the client did not send a handshake that matches with protocol
// specification.
defer rwc.Close() // 直播业务结束后，断开tcp连接
// 创建WS连接
conn, err := newServerConn(rwc, buf, req, &s.Config, s.Handshake)
if err != nil {
return
}
if conn == nil {
panic("unexpected nil conn")
}
// 这里类似接口回调，转发到上方的pwint接口中。pwint为阻塞接口，持续的连接读写。
s.Handler(conn)
}4） 最后看一下pwint，可以发现是最上层的业务处理：更新临时组、从当前的连接中读后转发给其他连接，完成直播中的多端分发处理。
好了，没写多少文字，大部分都是在分析代码，相信应该知道websocket是怎么回事了，这就是与new WebSocket而后Send、Receive调用相比有意思的地方的，就是可以看到背后的细节。