引入Websocket，尚未完成

dtu/link.go

@@ -5,18 +5,24 @@ import (
 	"github.com/zgwit/dtu-admin/db"
 	"github.com/zgwit/dtu-admin/model"
 	"net"
+	"sync"
 	"time"
 )
 
 type Link struct {
 	model.Link
 
-	registerChecked bool
-
 	Rx int
 	Tx int
 
+	//设备连接
 	conn net.Conn
+	//透传链接
+	peer net.Conn
+
+	//监视器连接，
+	monitors sync.Map // <string, websocket>
+
 
 	lastTime time.Time
 }
@@ -25,15 +31,40 @@ func (l *Link) onData(buf []byte) {
 	l.Rx += len(buf)
 	l.lastTime = time.Now()
 
-	//TODO 内容转发，暂时直接回复
-	_, _ = l.Send(buf)
+	if l.peer != nil {
+		//TODO 协议封装 ChannelId + LinkId + recv + buf
+		_, _ = l.peer.Write(buf)
+	}
+
+	l.monitors.Range(func(key, value interface{}) bool {
+		//TODO Websocket && 协议封装
+		con := value.(net.Conn)
+		_, _ = con.Write(buf)
+		return true
+	})
+	//内容转发，暂时直接回复
+	//_, _ = l.Send(buf)
 }
 
 func (l *Link) Send(buf []byte) (int, error) {
 	l.Tx += len(buf)
 	l.lastTime = time.Now()
 
-	return l.conn.Write(buf)
+	n, e := l.conn.Write(buf)
+
+	if l.peer != nil {
+		//TODO 协议封装
+		_, _ = l.peer.Write(buf)
+	}
+
+	l.monitors.Range(func(key, value interface{}) bool {
+		//TODO Websocket && 协议封装
+		con := value.(net.Conn)
+		_, _ = con.Write(buf)
+		return true
+	})
+
+	return n, e
 }
 
 func (l *Link) Close() error {

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/go-openapi/swag v0.19.9 // indirect
 	github.com/go-sql-driver/mysql v1.5.0
 	github.com/golang/snappy v0.0.1 // indirect
+	github.com/gorilla/websocket v1.4.2
 	github.com/mailru/easyjson v0.7.6 // indirect
 	github.com/swaggo/gin-swagger v1.2.0
 	github.com/swaggo/swag v1.6.7

go.sum

@@ -118,6 +118,8 @@ github.com/gorilla/securecookie v1.1.1/go.mod h1:ra0sb63/xPlUeL+yeDciTfxMRAA+MP+
 github.com/gorilla/sessions v1.1.1/go.mod h1:8KCfur6+4Mqcc6S0FEfKuN15Vl5MgXW92AE8ovaJD0w=
 github.com/gorilla/sessions v1.1.3 h1:uXoZdcdA5XdXF3QzuSlheVRUvjl+1rKY7zBXL68L9RU=
 github.com/gorilla/sessions v1.1.3/go.mod h1:8KCfur6+4Mqcc6S0FEfKuN15Vl5MgXW92AE8ovaJD0w=
+github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
+github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=

web/api/link.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
 	"github.com/zgwit/dtu-admin/db"
 	"github.com/zgwit/dtu-admin/dtu"
 	"github.com/zgwit/dtu-admin/model"
@@ -127,8 +128,8 @@ func linkGet(ctx *gin.Context) {
 		return
 	}
 
-	var channel model.Channel
-	has, err := db.Engine.ID(pid.Id).Get(&channel)
+	var link model.Link
+	has, err := db.Engine.ID(pid.Id).Get(&link)
 	if err != nil {
 		replyError(ctx, err)
 		return
@@ -138,5 +139,66 @@ func linkGet(ctx *gin.Context) {
 		return
 	}
 
-	replyOk(ctx, channel)
+	replyOk(ctx, link)
+}
+
+
+var upGrader = websocket.Upgrader{
+	CheckOrigin: func (r *http.Request) bool {
+		return true
+	},
 }
+
+
+func linkMonitor(ctx *gin.Context){
+	var pid paramId
+	if err := ctx.BindUri(&pid); err != nil {
+		replyError(ctx, err)
+		return
+	}
+
+	var link model.Link
+	has, err := db.Engine.ID(pid.Id).Get(&link)
+	if err != nil {
+		replyError(ctx, err)
+		return
+	}
+	if !has {
+		replyFail(ctx, "找不到通道")
+		return
+	}
+
+	_, err = dtu.GetLink(link.ChannelId, link.Id)
+	if err != nil {
+		replyError(ctx, err)
+		return
+	}
+
+
+	ws, err := upGrader.Upgrade(ctx.Writer, ctx.Request, nil)
+	if err != nil {
+		return
+	}
+
+	//TODO 嵌入Link，监听数据
+	//lnk.Add
+
+	defer ws.Close()
+	for {
+		//读取ws中的数据
+		mt, message, err := ws.ReadMessage()
+		if err != nil {
+			break
+		}
+		if string(message) == "ping" {
+			message = []byte("pong")
+		}
+		//写入ws数据
+		err = ws.WriteMessage(mt, message)
+		if err != nil {
+			break
+		}
+	}
+
+	//replyOk(ctx, link)
+}
\ No newline at end of file