新增网络模块

network/websocketclient.go

@@ -0,0 +1,159 @@
+package network
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+	"net/url"
+
+	"github.com/gorilla/websocket"
+
+	"time"
+)
+
+type IWebsocketClient interface {
+	Init(slf IWebsocketClient, strurl string, bproxy bool, timeoutsec time.Duration) error
+	Start() error
+	WriteMessage(msg []byte) error
+	OnDisconnect() error
+	OnConnected() error
+	OnReadMessage(msg []byte) error
+}
+
+type WebsocketClient struct {
+	WsDailer   *websocket.Dialer
+	conn       *websocket.Conn
+	url        string
+	state      int //0未连接状态   1连接状态
+	bwritemsg  chan []byte
+	slf        IWebsocketClient
+	timeoutsec time.Duration
+}
+
+func (ws *WebsocketClient) Init(slf IWebsocketClient, strurl string, bproxy bool, timeoutsec time.Duration) error {
+
+	ws.timeoutsec = timeoutsec
+	ws.slf = slf
+	if bproxy == true {
+		proxy := func(_ *http.Request) (*url.URL, error) {
+			return url.Parse("http://127.0.0.1:1080")
+		}
+
+		if timeoutsec > 0 {
+			tosec := timeoutsec * time.Second
+			ws.WsDailer = &websocket.Dialer{Proxy: proxy, HandshakeTimeout: tosec}
+		} else {
+			ws.WsDailer = &websocket.Dialer{Proxy: proxy}
+		}
+	} else {
+		if timeoutsec > 0 {
+			tosec := timeoutsec * time.Second
+			ws.WsDailer = &websocket.Dialer{HandshakeTimeout: tosec}
+		} else {
+			ws.WsDailer = &websocket.Dialer{}
+		}
+	}
+
+	ws.url = strurl
+	ws.bwritemsg = make(chan []byte, 1000)
+
+	return nil
+}
+
+func (ws *WebsocketClient) OnRun() error {
+	for {
+		if ws.state == 0 {
+			time.Sleep(1 * time.Second)
+			ws.StartConnect()
+		} else {
+			ws.conn.SetReadDeadline(time.Now().Add(ws.timeoutsec * time.Second))
+			_, message, err := ws.conn.ReadMessage()
+
+			if err != nil {
+				log.Printf("到服务器的连接断开 %+v\n", err)
+				ws.conn.Close()
+				ws.state = 0
+				ws.slf.OnDisconnect()
+				continue
+			}
+
+			ws.slf.OnReadMessage(message)
+		}
+	}
+
+	return nil
+}
+
+func (ws *WebsocketClient) StartConnect() error {
+
+	var err error
+	ws.conn, _, err = ws.WsDailer.Dial(ws.url, nil)
+	fmt.Printf("connecting %s, %+v\n", ws.url, err)
+	if err != nil {
+		return err
+	}
+
+	ws.state = 1
+	ws.slf.OnConnected()
+
+	return nil
+}
+
+func (ws *WebsocketClient) Start() error {
+	ws.state = 0
+	go ws.OnRun()
+	go ws.writeMsg()
+	return nil
+}
+
+//触发
+func (ws *WebsocketClient) writeMsg() error {
+	timerC := time.NewTicker(time.Second * 5).C
+	for {
+		if ws.state == 0 {
+			time.Sleep(1 * time.Second)
+			continue
+		}
+		select {
+		case <-timerC:
+			if ws.state == 1 {
+				ws.WriteMessage([]byte(`ping`))
+			}
+		case msg := <-ws.bwritemsg:
+			if ws.state == 1 {
+				ws.conn.SetWriteDeadline(time.Now().Add(ws.timeoutsec * time.Second))
+				err := ws.conn.WriteMessage(websocket.TextMessage, msg)
+
+				if err != nil {
+					fmt.Print(err)
+					ws.state = 0
+					ws.conn.Close()
+					ws.slf.OnDisconnect()
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func (ws *WebsocketClient) WriteMessage(msg []byte) error {
+	ws.bwritemsg <- msg
+	return nil
+}
+
+func (ws *WebsocketClient) OnDisconnect() error {
+
+	return nil
+}
+
+func (ws *WebsocketClient) OnConnected() error {
+
+	return nil
+}
+
+//触发
+func (ws *WebsocketClient) OnReadMessage(msg []byte) error {
+
+	return nil
+}

network/websocketserver.go

@@ -0,0 +1,123 @@
+package network
+
+import (
+	"fmt"
+	"net/http"
+	"sync"
+
+	"github.com/gorilla/websocket"
+)
+
+type IWebsocketServer interface {
+	SendMsg(clientid uint64, messageType int, msg []byte) bool
+}
+
+type IMessageReceiver interface {
+	OnConnected(webServer IWebsocketServer, clientid uint64)
+	OnDisconnect(webServer IWebsocketServer, clientid uint64, err error)
+	OnRecvMsg(webServer IWebsocketServer, clientid uint64, msgtype int, data []byte)
+}
+
+type WSClient struct {
+	clientid  uint64
+	conn      *websocket.Conn
+	bwritemsg chan WSMessage
+}
+
+type WSMessage struct {
+	msgtype   int
+	bwritemsg []byte
+}
+
+type WebsocketServer struct {
+	wsUri       string
+	maxClientid uint64 //记录当前最新clientid
+	mapClient   map[uint64]*WSClient
+
+	pattern            string
+	port               uint16
+	bEnableCompression bool
+	locker             sync.Mutex
+	messageReciver     IMessageReceiver
+}
+
+func (slf *WebsocketServer) wsHandler(w http.ResponseWriter, r *http.Request) {
+	/*if r.Header.Get("Origin") != "http://"+r.Host {
+		http.Error(w, "Origin not allowed", 403)
+		return
+	}
+	*/
+	conn, err := websocket.Upgrade(w, r, w.Header(), 1024, 1024)
+	if err != nil {
+		http.Error(w, "Could not open websocket connection", http.StatusBadRequest)
+	}
+
+	slf.maxClientid++
+	pclient := &WSClient{slf.maxClientid, conn, make(chan WSMessage, 1024)}
+	slf.mapClient[pclient.clientid] = pclient
+	slf.messageReciver.OnConnected(slf, pclient.clientid)
+	go pclient.startSendMsg()
+	go slf.OnConnected(pclient)
+}
+
+func (slf *WebsocketServer) OnConnected(pclient *WSClient) {
+	for {
+		msgtype, message, err := pclient.conn.ReadMessage()
+		if err != nil {
+			pclient.conn.Close()
+			slf.locker.Lock()
+			defer slf.locker.Unlock()
+			delete(slf.mapClient, pclient.clientid)
+			slf.messageReciver.OnDisconnect(slf, pclient.clientid, err)
+			return
+		}
+
+		slf.messageReciver.OnRecvMsg(slf, pclient.clientid, msgtype, message)
+	}
+}
+
+func (slf *WebsocketServer) Init(pattern string, port uint16, messageReciver IMessageReceiver, bEnableCompression bool) {
+	slf.pattern = pattern
+	slf.port = port
+	slf.bEnableCompression = bEnableCompression
+	slf.mapClient = make(map[uint64]*WSClient)
+	slf.messageReciver = messageReciver
+
+	http.HandleFunc("/ws", slf.wsHandler)
+}
+
+func (slf *WebsocketServer) startListen() {
+	address := fmt.Sprintf(":%d", slf.port)
+	err := http.ListenAndServe(address, nil)
+	if err != nil {
+		fmt.Printf("%v", err)
+	}
+}
+
+func (slf *WSClient) startSendMsg() {
+	for {
+		msgbuf := <-slf.bwritemsg
+		slf.conn.WriteMessage(msgbuf.msgtype, msgbuf.bwritemsg)
+	}
+}
+
+func (slf *WebsocketServer) Start() {
+
+	go slf.startListen()
+}
+
+func (slf *WebsocketServer) SendMsg(clientid uint64, messageType int, msg []byte) bool {
+	slf.locker.Lock()
+	defer slf.locker.Unlock()
+	value, ok := slf.mapClient[clientid]
+	if ok == false {
+		return false
+	}
+
+	value.bwritemsg <- WSMessage{messageType, msg}
+	return true
+}
+
+func (slf *WebsocketServer) Stop() {
+
+}