优化websocket client

This commit is contained in:
lifeiyi
2019-02-20 10:55:19 +08:00
parent 7d1897992c
commit 6743a70794

View File

@@ -11,32 +11,38 @@ import (
"time" "time"
) )
//IWebsocketClient ...
type IWebsocketClient interface { type IWebsocketClient interface {
Init(slf IWebsocketClient, strurl string, bproxy bool, timeoutsec time.Duration) error Init(slf IWebsocketClient, strurl, strPath string, bproxy bool, timeoutsec time.Duration) error
Start() error Start() error
WriteMessage(msg []byte) error WriteMessage(msg []byte) error
OnDisconnect() error OnDisconnect() error
OnConnected() error OnConnected() error
OnReadMessage(msg []byte) error OnReadMessage(msg []byte) error
ReConnect()
} }
//WebsocketClient ...
type WebsocketClient struct { type WebsocketClient struct {
WsDailer *websocket.Dialer WsDailer *websocket.Dialer
conn *websocket.Conn conn *websocket.Conn
url string url string
state int //0未连接状态 1连接状态 state int //0未连接状态 1正在重连 2连接状态
bwritemsg chan []byte bwritemsg chan []byte
slf IWebsocketClient slf IWebsocketClient
timeoutsec time.Duration timeoutsec time.Duration
bRun bool
} }
func (ws *WebsocketClient) Init(slf IWebsocketClient, strurl string, bproxy bool, timeoutsec time.Duration) error { //Init ...
func (ws *WebsocketClient) Init(slf IWebsocketClient, strurl, strPath string, bproxy bool, timeoutsec time.Duration) error {
ws.timeoutsec = timeoutsec ws.timeoutsec = timeoutsec
ws.slf = slf ws.slf = slf
if bproxy == true { if bproxy == true {
proxy := func(_ *http.Request) (*url.URL, error) { proxy := func(_ *http.Request) (*url.URL, error) {
return url.Parse("http://127.0.0.1:1080") return url.Parse(strPath)
} }
if timeoutsec > 0 { if timeoutsec > 0 {
@@ -60,12 +66,22 @@ func (ws *WebsocketClient) Init(slf IWebsocketClient, strurl string, bproxy bool
return nil return nil
} }
//OnRun ...
func (ws *WebsocketClient) OnRun() error { func (ws *WebsocketClient) OnRun() error {
for { for {
if ws.bRun == false {
break
}
if ws.state == 0 { if ws.state == 0 {
time.Sleep(1 * time.Second) time.Sleep(1 * time.Second)
ws.StartConnect() ws.StartConnect()
} else { } else if ws.state == 1 {
log.Println("需要进行重连")
ws.conn.Close()
ws.state = 0
ws.slf.OnDisconnect()
} else if ws.state == 2 {
ws.conn.SetReadDeadline(time.Now().Add(ws.timeoutsec * time.Second)) ws.conn.SetReadDeadline(time.Now().Add(ws.timeoutsec * time.Second))
_, message, err := ws.conn.ReadMessage() _, message, err := ws.conn.ReadMessage()
@@ -84,6 +100,7 @@ func (ws *WebsocketClient) OnRun() error {
return nil return nil
} }
//StartConnect ...
func (ws *WebsocketClient) StartConnect() error { func (ws *WebsocketClient) StartConnect() error {
var err error var err error
@@ -93,13 +110,15 @@ func (ws *WebsocketClient) StartConnect() error {
return err return err
} }
ws.state = 1 ws.state = 2
ws.slf.OnConnected() ws.slf.OnConnected()
return nil return nil
} }
//Start ...
func (ws *WebsocketClient) Start() error { func (ws *WebsocketClient) Start() error {
ws.bRun = true
ws.state = 0 ws.state = 0
go ws.OnRun() go ws.OnRun()
go ws.writeMsg() go ws.writeMsg()
@@ -110,17 +129,21 @@ func (ws *WebsocketClient) Start() error {
func (ws *WebsocketClient) writeMsg() error { func (ws *WebsocketClient) writeMsg() error {
timerC := time.NewTicker(time.Second * 5).C timerC := time.NewTicker(time.Second * 5).C
for { for {
if ws.bRun == false {
break
}
if ws.state == 0 { if ws.state == 0 {
time.Sleep(1 * time.Second) time.Sleep(1 * time.Second)
continue continue
} }
select { select {
case <-timerC: case <-timerC:
if ws.state == 1 { if ws.state == 2 {
ws.WriteMessage([]byte(`ping`)) ws.WriteMessage([]byte(`ping`))
} }
case msg := <-ws.bwritemsg: case msg := <-ws.bwritemsg:
if ws.state == 1 { if ws.state == 2 {
ws.conn.SetWriteDeadline(time.Now().Add(ws.timeoutsec * time.Second)) ws.conn.SetWriteDeadline(time.Now().Add(ws.timeoutsec * time.Second))
err := ws.conn.WriteMessage(websocket.TextMessage, msg) err := ws.conn.WriteMessage(websocket.TextMessage, msg)
@@ -137,23 +160,36 @@ func (ws *WebsocketClient) writeMsg() error {
return nil return nil
} }
//ReConnect ...
func (ws *WebsocketClient) ReConnect() {
ws.state = 1
}
//WriteMessage ...
func (ws *WebsocketClient) WriteMessage(msg []byte) error { func (ws *WebsocketClient) WriteMessage(msg []byte) error {
ws.bwritemsg <- msg ws.bwritemsg <- msg
return nil return nil
} }
//OnDisconnect ...
func (ws *WebsocketClient) OnDisconnect() error { func (ws *WebsocketClient) OnDisconnect() error {
return nil return nil
} }
//OnConnected ...
func (ws *WebsocketClient) OnConnected() error { func (ws *WebsocketClient) OnConnected() error {
return nil return nil
} }
//触发 //OnReadMessage 触发
func (ws *WebsocketClient) OnReadMessage(msg []byte) error { func (ws *WebsocketClient) OnReadMessage(msg []byte) error {
return nil return nil
} }
//Stop ...
func (ws *WebsocketClient) Stop() {
ws.bRun = false
}