-
Notifications
You must be signed in to change notification settings - Fork 92
/
client.go
234 lines (213 loc) · 7.05 KB
/
client.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
package gws
import (
"bufio"
"context"
"crypto/tls"
"encoding/base64"
"encoding/binary"
"net"
"net/http"
"net/url"
"strings"
"time"
"github.com/lxzan/gws/internal"
)
// Dialer 拨号器接口
// Dialer interface
type Dialer interface {
// Dial 连接到指定网络上的地址
// Connects to the address on the named network
Dial(network, addr string) (c net.Conn, err error)
}
type connector struct {
option *ClientOption
conn net.Conn
eventHandler Event
secWebsocketKey string
}
// NewClient 创建一个新的 WebSocket 客户端连接
// Creates a new WebSocket client connection
func NewClient(handler Event, option *ClientOption) (*Conn, *http.Response, error) {
option = initClientOption(option)
c := &connector{option: option, eventHandler: handler}
URL, err := url.Parse(option.Addr)
if err != nil {
return nil, nil, err
}
if URL.Scheme != "ws" && URL.Scheme != "wss" {
return nil, nil, ErrUnsupportedProtocol
}
var tlsEnabled = URL.Scheme == "wss"
dialer, err := option.NewDialer()
if err != nil {
return nil, nil, err
}
c.conn, err = dialer.Dial("tcp", internal.GetAddrFromURL(URL, tlsEnabled))
if err != nil {
return nil, nil, err
}
if tlsEnabled {
if option.TlsConfig == nil {
option.TlsConfig = &tls.Config{}
}
if option.TlsConfig.ServerName == "" {
option.TlsConfig.ServerName = URL.Hostname()
}
c.conn = tls.Client(c.conn, option.TlsConfig)
}
client, resp, err := c.handshake()
if err != nil {
_ = c.conn.Close()
}
return client, resp, err
}
// NewClientFromConn 通过外部连接创建客户端, 支持 TCP/KCP/Unix Domain Socket
// Create new client via external connection, supports TCP/KCP/Unix Domain Socket.
func NewClientFromConn(handler Event, option *ClientOption, conn net.Conn) (*Conn, *http.Response, error) {
option = initClientOption(option)
c := &connector{option: option, conn: conn, eventHandler: handler}
client, resp, err := c.handshake()
if err != nil {
_ = c.conn.Close()
}
return client, resp, err
}
// 发送HTTP请求, 即WebSocket握手
// Sends an http request, i.e., websocket handshake
func (c *connector) request() (*http.Response, *bufio.Reader, error) {
_ = c.conn.SetDeadline(time.Now().Add(c.option.HandshakeTimeout))
ctx, cancel := context.WithTimeout(context.Background(), c.option.HandshakeTimeout)
defer cancel()
// 构建HTTP请求
// building a http request
r, err := http.NewRequestWithContext(ctx, http.MethodGet, c.option.Addr, nil)
if err != nil {
return nil, nil, err
}
for k, v := range c.option.RequestHeader {
r.Header[k] = v
}
r.Header.Set(internal.Connection.Key, internal.Connection.Val)
r.Header.Set(internal.Upgrade.Key, internal.Upgrade.Val)
r.Header.Set(internal.SecWebSocketVersion.Key, internal.SecWebSocketVersion.Val)
if c.option.PermessageDeflate.Enabled {
r.Header.Set(internal.SecWebSocketExtensions.Key, c.option.PermessageDeflate.genRequestHeader())
}
if c.secWebsocketKey == "" {
var key [16]byte
binary.BigEndian.PutUint64(key[0:8], internal.AlphabetNumeric.Uint64())
binary.BigEndian.PutUint64(key[8:16], internal.AlphabetNumeric.Uint64())
c.secWebsocketKey = base64.StdEncoding.EncodeToString(key[0:])
r.Header.Set(internal.SecWebSocketKey.Key, c.secWebsocketKey)
}
var ch = make(chan error)
// 发送http请求
// send http request
go func() { ch <- r.Write(c.conn) }()
// 同步等待请求是否发送成功
// Synchronized waiting for the request to be sent successfully
select {
case err = <-ch:
case <-ctx.Done():
err = ctx.Err()
}
if err != nil {
return nil, nil, err
}
// 读取响应结果
// Read the response result
br := bufio.NewReaderSize(c.conn, c.option.ReadBufferSize)
resp, err := http.ReadResponse(br, r)
return resp, br, err
}
// 获取压缩拓展结果
// Get compression expansion results
func (c *connector) getPermessageDeflate(extensions string) PermessageDeflate {
serverPD := permessageNegotiation(extensions)
clientPD := c.option.PermessageDeflate
pd := PermessageDeflate{
Enabled: clientPD.Enabled && strings.Contains(extensions, internal.PermessageDeflate),
Threshold: clientPD.Threshold,
Level: clientPD.Level,
PoolSize: clientPD.PoolSize,
ServerContextTakeover: serverPD.ServerContextTakeover,
ClientContextTakeover: serverPD.ClientContextTakeover,
ServerMaxWindowBits: serverPD.ServerMaxWindowBits,
ClientMaxWindowBits: serverPD.ClientMaxWindowBits,
}
pd.setThreshold(false)
return pd
}
// 执行 WebSocket 握手操作
// Performs the WebSocket handshake operation
func (c *connector) handshake() (*Conn, *http.Response, error) {
resp, br, err := c.request()
if err != nil {
return nil, resp, err
}
if err = c.checkHeaders(resp); err != nil {
return nil, resp, err
}
subprotocol, err := c.getSubProtocol(resp)
if err != nil {
return nil, resp, err
}
var extensions = resp.Header.Get(internal.SecWebSocketExtensions.Key)
var pd = c.getPermessageDeflate(extensions)
socket := &Conn{
ss: c.option.NewSession(),
isServer: false,
subprotocol: subprotocol,
pd: pd,
conn: c.conn,
config: c.option.getConfig(),
br: br,
continuationFrame: continuationFrame{},
fh: frameHeader{},
handler: c.eventHandler,
closed: 0,
deflater: new(deflater),
writeQueue: workerQueue{maxConcurrency: 1},
readQueue: make(channel, c.option.ParallelGolimit),
}
// 压缩字典和解压字典内存开销比较大, 故使用懒加载
// Compressing and decompressing dictionaries has a large memory overhead, so use lazy loading.
if pd.Enabled {
socket.deflater.initialize(false, pd, c.option.ReadMaxPayloadSize)
if pd.ServerContextTakeover {
socket.dpsWindow.initialize(nil, pd.ServerMaxWindowBits)
}
if pd.ClientContextTakeover {
socket.cpsWindow.initialize(nil, pd.ClientMaxWindowBits)
}
}
return socket, resp, c.conn.SetDeadline(time.Time{})
}
// 从响应中获取子协议
// Retrieves the subprotocol from the response
func (c *connector) getSubProtocol(resp *http.Response) (string, error) {
a := internal.Split(c.option.RequestHeader.Get(internal.SecWebSocketProtocol.Key), ",")
b := internal.Split(resp.Header.Get(internal.SecWebSocketProtocol.Key), ",")
subprotocol := internal.GetIntersectionElem(a, b)
if len(a) > 0 && subprotocol == "" {
return "", ErrSubprotocolNegotiation
}
return subprotocol, nil
}
// 检查响应头以验证握手是否成功
// Checks the response headers to verify if the handshake was successful
func (c *connector) checkHeaders(resp *http.Response) error {
if resp.StatusCode != http.StatusSwitchingProtocols {
return ErrHandshake
}
if !internal.HttpHeaderContains(resp.Header.Get(internal.Connection.Key), internal.Connection.Val) {
return ErrHandshake
}
if !strings.EqualFold(resp.Header.Get(internal.Upgrade.Key), internal.Upgrade.Val) {
return ErrHandshake
}
if resp.Header.Get(internal.SecWebSocketAccept.Key) != internal.ComputeAcceptKey(c.secWebsocketKey) {
return ErrHandshake
}
return nil
}