ws分用户发数据

diff --git a/F/xinxin.go b/F/xinxin.go
index 62d7069c6d30698df66aec3726129556ea9f57e4..e8b30f63652fe1a225613ccf27db8865580c3405 100644 (file)
--- a/F/xinxin.go
+++ b/F/xinxin.go
@@ -1,14 +1,17 @@
 package F
 
 import (
+       "net"
        "net/http"
        "encoding/json"
     "time"
        "context"
+       "sync"
        "strconv"
        "github.com/gorilla/websocket"
        "github.com/skratchdot/open-golang/open"
        p "github.com/qydysky/part"
+       mq "github.com/qydysky/part/msgq"
        c "github.com/qydysky/bili_danmu/CV"
 )
 
@@ -37,12 +40,18 @@ type S struct {
        S string `json:"s"` //加密字符串
 }
 
+type Uinterface struct {
+       Id uint
+       Data interface{}
+       sync.Mutex
+}
+
 //全局对象
 var (
        xinxinboot = make(chan struct{},1) //调用标记，仅调用一次
        wslog = c.Log.Base(`api`).Base_add(`小心心加密`) //日志
-       send_chan = make(chan RT,1)//发通道
        rec_chan = make(chan S,1)//收通道
+       ws_mq = mq.New(200)//发通道
        port = p.Sys().GetFreePort()//随机端口
 )
 
@@ -77,6 +86,9 @@ func web() {
        var (
                server *http.Server
                upgrader = websocket.Upgrader{}
+               id = Uinterface{
+                       Id:1,//0表示全局广播
+               }
        )
 
        web.HandleFunc("/exit", func(w http.ResponseWriter, r *http.Request) {
@@ -91,46 +103,40 @@ func web() {
                }
                defer ws.Close()
 
+               //本会话id
+               Uid := id.Id
+               id.Lock()
+               id.Id += 1
+               id.Unlock()
+
                //测试 提示
-               go func(){
-                       time.Sleep(time.Second*time.Duration(3))
-                       if s := Wasm(RT{
-                               R:R{
-                               Id: "[9,371,1,22613059]",
-                               Device: "[\"AUTO8216117272375373\",\"77bee604-b591-4664-845b-b69603f8c71c\"]",
-                               Ets: 1611836581,
-                               Benchmark: "seacasdgyijfhofiuxoannn",
-                               Time: 60,
-                               Ts: 1611836642190,
-                               },
-                               T: []int{2, 5, 1, 4},
-                       });s == `e4249b7657c2d4a44955548eb814797d41ddd99bfdfa5974462b8c387d701b8c83898f6d7dde1772c67fad6a113d20c20e454be1d1627e7ea99617a8a1f99bd0` {
-                               wslog.L(`I: `,`测试通过`)
-                       } else {
-                               wslog.L(`E: `,`测试未通过`,s)
-                       }
-               }()
+               go test(Uid)
 
-               var close_chan = make(chan struct{})
                //发送
-               go func(){
-                       for {
-                               select {
-                               case <-close_chan:
-                                       return
-                               case tmp := <- send_chan:
-                                       b, e := json.Marshal(tmp)
-                                       if e != nil {
-                                               wslog.L(`E: `,e)
-                                       }
-
-                                       if e := ws.WriteMessage(websocket.TextMessage,b);e != nil {
-                                               wslog.L(`E: `,e)
-                                               return
+               ws_mq.Pull_tag(map[string]func(interface{})(bool){
+                       `send`:func(data interface{})(bool){
+                               if u,ok := data.(Uinterface);ok && u.Id == 0 || u.Id == Uid{
+                                       if t,ok := u.Data.(RT);ok {
+                                               b, e := json.Marshal(t)
+                                               if e != nil {
+                                                       wslog.L(`E: `,e)
+                                               }
+
+                                               if e := ws.WriteMessage(websocket.TextMessage,b);e != nil {
+                                                       wslog.L(`E: `,e)
+                                                       return true
+                                               }
                                        }
                                }
-                       }
-               }()
+                               return false
+                       },
+                       `close`:func(data interface{})(bool){
+                               if u,ok := data.(Uinterface);ok && u.Id == 0 || u.Id == Uid{
+                                       return true
+                               }
+                               return false
+                       },
+               })
 
                //接收
                for {
@@ -138,10 +144,14 @@ func web() {
                        if _, message, e := ws.ReadMessage();e != nil {
                                if websocket.IsCloseError(e,websocket.CloseGoingAway) {
                                        wslog.L(`I: `,e)
+                               } else if e,ok := e.(net.Error);ok && e.Timeout() {
+                                       //Timeout , js will reload html
                                } else {
                                        wslog.L(`E: `,e)
                                }
-                               close(close_chan)
+                               ws_mq.Push_tag(`close`,Uinterface{
+                                       Id:Uid,
+                               })
                                break
                        } else {
                                var s S
@@ -149,7 +159,11 @@ func web() {
                                if e != nil {
                                        wslog.L(`E: `, e, string(message))
                                }
-                               rec_chan <- s
+
+                               select{//现阶段暂不考虑多用户上传不同数据的情况
+                               case rec_chan <- s:
+                               default:
+                               }
                        }
                }
        })
@@ -177,17 +191,36 @@ func web() {
        server.ListenAndServe()
 }
 
-func Wasm(s RT) (o string) {
-       select{
-       case send_chan <- s:
-               select {
-               case r :=<- rec_chan:
-                       return r.S
-               case <- time.After(time.Second):
-                       wslog.L(`E: `,`超时！响应>1s，确认保持`,"http://127.0.0.1:"+strconv.Itoa(port),`开启`)
-                       return
-               }
-       default:
+func Wasm(uid uint,s RT) (o string) {
+       ws_mq.Push_tag(`send`,Uinterface{
+               Id:uid,
+               Data:s,
+       })
+
+       select {
+       case r :=<- rec_chan:
+               return r.S
+       case <- time.After(time.Second):
+               wslog.L(`E: `,`超时！响应>1s，确认保持`,"http://127.0.0.1:"+strconv.Itoa(port),`开启`)
                return
        }
+}
+
+func test(uid uint) bool {
+       time.Sleep(time.Second*time.Duration(3))
+       if s := Wasm(uid, RT{
+               R:R{
+               Id: "[9,371,1,22613059]",
+               Device: "[\"AUTO8216117272375373\",\"77bee604-b591-4664-845b-b69603f8c71c\"]",
+               Ets: 1611836581,
+               Benchmark: "seacasdgyijfhofiuxoannn",
+               Time: 60,
+               Ts: 1611836642190,
+               },
+               T: []int{2, 5, 1, 4},
+       });s != `e4249b7657c2d4a44955548eb814797d41ddd99bfdfa5974462b8c387d701b8c83898f6d7dde1772c67fad6a113d20c20e454be1d1627e7ea99617a8a1f99bd0` {
+               wslog.L(`E: `,`测试未通过`,s)
+               return false
+       }
+       return true
 }
\ No newline at end of file