/
ws_new.go
281 lines (235 loc) · 6.51 KB
/
ws_new.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
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
package websocket
import (
"Futures-Go-demo/config"
"bytes"
"compress/gzip"
"encoding/binary" //"flag"
"fmt"
"io/ioutil"
"log"
"math/rand"
"net"
"time"
"github.com/gorilla/websocket"
"github.com/spf13/cast"
)
type Message struct {
Ts int `json:"ts"`
Status string `json:"status"`
ErrCode string `json:"err-code"`
ErrMsg string `json:"err-msg"`
Ping int `json:"ping"`
}
type Client struct {
Name string
Addr string
Ws *websocket.Conn
}
var (
localIP string
)
type Moniter struct {
clientNum int
addChan chan int
subChan chan int
lastUseSec int
}
var mon *Moniter
func InitMoniter() {
mon = &Moniter{}
mon.addChan = make(chan int, 1000)
mon.subChan = make(chan int, 1000)
go func() {
for {
select {
case <-mon.addChan:
mon.clientNum++
case <-mon.subChan:
mon.clientNum--
}
}
}()
}
func AddClientNum() {
mon.addChan <- 1
}
func SubClientNum() {
mon.subChan <- 1
}
func NewClient(addr string, name string) *Client {
return &Client{Name: name, Addr: addr}
}
func NowSec() int {
return int(time.Now().UnixNano() / 1000000000)
}
func (cli *Client) RunClient() {
AddClientNum()
dialer := websocket.DefaultDialer
dialer.NetDial = func(network, addr string) (net.Conn, error) {
addrs := []string{string(localIP)}
fmt.Println("addrs=", addrs)
localAddr := &net.TCPAddr{IP: net.ParseIP(addrs[rand.Int()%len(addrs)]), Port: 0}
d := net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
LocalAddr: localAddr,
DualStack: true,
}
c, err := d.Dial(network, addr)
return c, err
}
c, _, err := dialer.Dial(cli.Addr, nil)
if err != nil {
log.Println("Dial Erro:", err)
SubClientNum()
return
}
log.Println(c.LocalAddr().String())
// defer c.Close()
defer func() {
c.Close()
SubClientNum()
}()
//===============================================================================
//订阅websocket kline
/*
"market.$symbol.kline.$period"
symbol true string 交易对 如"BTC_CW"表示BTC当周合约,"BTC_NW"表示BTC次周合约,"BTC_CQ"表示BTC季度合约
period true string K线周期 1min, 5min, 15min, 30min, 60min,4hour,1day, 1mon
*/
message := []byte("{\"Sub\":\"market.BTC_CW.kline.60min\"}")
err = c.WriteMessage(websocket.TextMessage, message)
if err != nil {
log.Println("write err :", err)
}
//订阅websocket Market Detail 数据
/*
"market.$symbol.detail"
symbol true string 交易对 如"BTC_CW"表示BTC当周合约,"BTC_NW"表示BTC次周合约,"BTC_CQ"表示BTC季度合约
*/
message = []byte("{\"Sub\":\"market.BTC_CW.detail\"}")
err = c.WriteMessage(websocket.TextMessage, message)
if err != nil {
log.Println("write err :", err)
}
//订阅websocket Trade Detail 数据
/*
"market.$symbol.trade.detail"
symbol true string 交易对 如"BTC_CW"表示BTC当周合约,"BTC_NW"表示BTC次周合约,"BTC_CQ"表示BTC季度合约
*/
message = []byte("{\"Sub\":\"market.BTC_CW.trade.detail\"}")
err = c.WriteMessage(websocket.TextMessage, message)
if err != nil {
log.Println("write err :", err)
}
//订阅websocket Market Depth 数据
/*
"market.$symbol.depth.$type"
symbol true string 交易对 如"BTC_CW"表示BTC当周合约,"BTC_NW"表示BTC次周合约,"BTC_CQ"表示BTC季度合约.
type true string Depth 类型 (150档数据) step0, step1, step2, step3, step4, step5(合并深度1-5);step0时,不合并深度
(20档数据) step6, step7, step8, step9, step10, step11(合并深度7-11);step6时,不合并深度
*/
message = []byte("{\"Sub\":\"market.BTC_CW.depth.step0\"}")
err = c.WriteMessage(websocket.TextMessage, message)
if err != nil {
log.Println("write err :", err)
}
//请求websocket KLine 数据
/*
"market.$symbol.kline.$period"
symbol true string 交易对 如"BTC_CW"表示BTC当周合约,"BTC_NW"表示BTC次周合约,"BTC_CQ"表示BTC季度合约
period true string K线周期 1min, 5min, 15min, 30min, 60min,4hour,1day, 1mon
"from": "optional, type: long, 2017-07-28T00:00:00+08:00 至2050-01-01T00:00:00+08:00 之间的时间点,单位:秒",
"to": "optional, type: long, 2017-07-28T00:00:00+08:00 至2050-01-01T00:00:00+08:00 之间的时间点,单位:秒,必须比 from 大"}
[t1, t5] 假设有 t1 ~ t5 的K线:
from: t1, to: t5, return [t1, t5].
from: t5, to: t1, which t5 > t1, return [].
from: t5, return [t5].
from: t3, return [t3, t5].
to: t5, return [t1, t5].
from: t which t3 < t <t4, return [t4, t5].
to: t which t3 < t <t4, return [t1, t3].
*/
message = []byte("{\"req\":\"market.BTC_CQ.kline.1day\",\"from\":1544170607,\"to\":1544602608}")
err = c.WriteMessage(websocket.TextMessage, message)
if err != nil {
log.Println("write err :", err)
}
go func() {
pangTicker := time.NewTicker(time.Second * 5)
for {
select {
case <-pangTicker.C:
message = []byte(fmt.Sprintf("{\"pong\":%d}", time.Now().Unix()))
err = c.WriteMessage(websocket.TextMessage, message)
if err != nil {
log.Println("send msg err:", err)
return
}
}
}
}()
for {
_, zipmsg, err := c.ReadMessage()
if err != nil {
log.Println("Read Error : ", err, cli.Name)
c.Close()
return
}
msg, err := parseGzip(zipmsg)
if err != nil {
log.Println("gzip Error : ", err)
}
log.Println(string(msg[:]))
}
}
var address string
func WSRunWithIP(ip string) {
address = config.WS_URL
localIP = ip
InitMoniter()
createClient(address)
time.Sleep(1 * time.Millisecond)
reCreateClient()
for {
time.Sleep(10 * time.Second)
}
}
func reCreateClient() {
go func() {
time.Sleep(time.Second * 100)
checkTicker := time.NewTicker(time.Second * 20)
for {
select {
case <-checkTicker.C:
if mon.clientNum <= 0 {
createClient(address)
}
}
}
}()
}
var clientNameNum int
func createClient(addr string) {
clientNameNum++
c := NewClient(addr, cast.ToString(clientNameNum))
go c.RunClient()
}
func parseGzip(data []byte) ([]byte, error) {
b := new(bytes.Buffer)
binary.Write(b, binary.LittleEndian, data)
r, err := gzip.NewReader(b)
if err != nil {
fmt.Println("[ParseGzip] NewReader error: , maybe data is ungzip: ", err, string(data))
fmt.Println("[ParseGzip] NewReader error: , maybe data is ungzip: ", err, string(b.Bytes()))
return nil, err
} else {
defer r.Close()
undatas, err := ioutil.ReadAll(r)
if err != nil {
fmt.Println("[ParseGzip] ioutil.ReadAll error: :", err, string(data))
return nil, err
}
return undatas, nil
}
}