From: qydysky <qydysky@foxmail.com>
Date: Mon, 1 Feb 2021 02:47:59 +0000 (+0800)
Subject: ws分用户发数据
X-Git-Tag: v0.5.7~1^2~17
X-Git-Url: http://127.0.0.1:8081/?a=commitdiff_plain;h=a043004f325858bb595b325b40434f0757ab7a41;p=bili_danmu%2F.git

ws分用户发数据
---

diff --git a/F/xinxin.go b/F/xinxin.go
index 62d7069..e8b30f6 100644
--- 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