AIOT-coer/lib/SubClinets.go

package lib

import (
	"AIOTCOER/logs"
	"encoding/json"
	"fmt"
	"github.com/beego/beego/v2/server/web/context"
	"github.com/gorilla/websocket"

	"sync"

)

var WsSubClinets sync.Map // 订阅Map
var SseSubClinets sync.Map // 订阅Map
var SseLogSubClinets sync.Map // 订阅Map

// 清楚订阅
func ClearSubscription(client *websocket.Conn) {
	logs.Println("清楚订阅：", fmt.Sprintf("%p", client))
	// 编列所有订阅号
	WsSubClinets.Range(func(key, value interface{}) bool {
		ConnList := value.([]*websocket.Conn)
		Sn := key.(string)
		logs.Println("SubClinets_数量|", Sn, len(ConnList))
		for i, v := range ConnList {
			if v == client {
				fmt.Println("删除订阅：", fmt.Sprintf("%p", v), Sn)
				ConnList = append(ConnList[:i], ConnList[i+1:]...)
				WsSubClinets.Store(Sn, ConnList)
				break
			}
		}

		return true
	})
}

// 发送 数据
func WebsocketWriteJSON(client *websocket.Conn, msg interface{}) error {
	err := client.WriteJSON(msg)
	if err != nil {
		ClearSubscription(client) // 清楚订阅
		client.Close()
	}
	return nil
}

func WebsocketSubscribeSendAll(Sn string, json_r map[string]interface{}) {

	val, ok := WsSubClinets.Load(Sn)
	if ok {
		json_r_ws := CopyMap(json_r)
		json_r_ws["T_sn"] = Sn
		SubConn := val.([]*websocket.Conn)
		for _, v := range SubConn {
			WebsocketWriteJSON(v, json_r_ws)
		}
	}
}



// 发送 数据
func SseWriteJSON(client *context.Response, msg []byte) error {

	// Send data to the client
	client.Write(append(append([]byte("data: "), msg...), []byte("\n\n")...))
	//client.Write([]byte("\n"))

	// Flush the response writer to send the data immediately
	client.Flush()

	return nil
}
func SseSubscribeSendAll(Sn string, json_r map[string]interface{}) {
	val, ok := SseSubClinets.Load(Sn)
	if ok {
		json_r_ws_data, _ := json.Marshal(json_r)
		SubConn := val.([]*context.Response)
		for _, v := range SubConn {
			SseWriteJSON(v, json_r_ws_data)
		}
	}
}

func SseLogSubscribeSendAll(Sn string, json_r []string) {
	val, ok := SseLogSubClinets.Load(Sn)
	if ok {
		json_r_ws_data, _ := json.Marshal(json_r)
		SubConn := val.([]*context.Response)
		for _, v := range SubConn {
			SseWriteJSON(v, json_r_ws_data)
		}
	}
}