confrontation-training/api/gateway/sse.go

package gateway

import (
	"confrontation-training/constant"
	errors "confrontation-training/err"
	"confrontation-training/global"
	"confrontation-training/models/gateway"
	"encoding/json"
	"fmt"
	"github.com/gorilla/websocket"
	"github.com/r3labs/sse/v2"
	"net/http"
	"strconv"
	"strings"
	"time"
)

// SseScanDevice 扫描设备
func SseScanDevice(paramMap map[string]string, filterType string) {
	var scanUrl = global.Config.Gateway.BaseUrl + global.Config.Gateway.ScanUrl
	deviceMap := make(map[string]gateway.DeviceScanned)
	if len(paramMap) > 0 {
		for key, value := range paramMap {
			scanUrl += "&" + key + "=" + value
		}
	}
	fmt.Println("创建新的sse")
	//events := make(chan *sse.Event)
	client := global.SseClientDevice
	client.URL = scanUrl
	//ws, _, err := websocket.DefaultDialer.Dial(global.Config.Websocket.WSUrl, nil)
	//if err != nil {
	//	response.Failed(errors.SocketInitError, c)
	//}
	ws, _, err := websocket.DefaultDialer.Dial(global.Config.Websocket.WSUrl, nil)
	if err != nil {
		panic(err)
		return
	}
	fmt.Println("开始SSE")

	err = client.SubscribeRaw(func(msg *sse.Event) {
		var data gateway.DeviceScannedFromGateway
		err := json.Unmarshal(msg.Data, &data)
		if err != nil {
			fmt.Printf("SSE数据转换异常:%s", err.Error())
			return
		}
		deviceScanned := gateway.DeviceScanned{}
		deviceScanned.MAC = data.Bdaddrs[0].Bdaddr
		//库中存在的设备
		if filterType == "1" {
			deviceService := GetDeviceService()
			d, i := deviceService.FindDeviceByMac(deviceScanned.MAC)
			if i > 0 {
				deviceScanned.Name = data.Name
				deviceScanned.Rssi = data.Rssi
				deviceScanned.BdadrType = data.Bdaddrs[0].BdaddrType
				deviceScanned.Chip = data.ChipId
				deviceScanned.DeviceType = d.Type
				deviceMap[deviceScanned.MAC] = deviceScanned
				fmt.Println(deviceScanned)
			}
		} else {
			deviceScanned.Name = data.Name
			deviceScanned.Rssi = data.Rssi
			deviceScanned.BdadrType = data.Bdaddrs[0].BdaddrType
			deviceScanned.Chip = data.ChipId
			deviceMap[deviceScanned.MAC] = deviceScanned
		}
		if len(deviceMap) > 0 {

			if err != nil {
				fmt.Println("Websocket client init error" + err.Error())
				return
			}
			messageMap := make(map[string]string)
			messageMap["msgType"] = constant.MessageTypeDeviceScanned
			marshal, err := json.Marshal(deviceMap)
			if err != nil {
				fmt.Println("data parse Error :" + err.Error())
				return
			}
			messageMap["content"] = string(marshal)
			messageMap["Sender"] = "server"
			messageMap["Recipient"] = "client"
			bytes, err := json.Marshal(messageMap)
			if err != nil {
				fmt.Printf("json异常：%s", err.Error())
				return
			}
			err = ws.WriteMessage(websocket.TextMessage, bytes)
			if err != nil {
				fmt.Println("消息发送异常：" + err.Error())
				return
			}
		} else {
			err := ws.WriteMessage(websocket.TextMessage, []byte("{}"))
			if err != nil {
				fmt.Println("消息发送异常：" + err.Error())
				return
			}
		}
		time.Sleep(200 * time.Millisecond)
	})
}
func StopScan(w http.ResponseWriter, r *http.Request) {
	go func() {
		<-r.Context().Done()
		println("The client is disconnected here")
		return
	}()
}

// SseOpenNotify 开启通知
func SseOpenNotify() {
	var notifyUrl = global.Config.Gateway.BaseUrl + global.Config.Gateway.NotifyUrl

	ws, _, err := websocket.DefaultDialer.Dial(global.Config.Websocket.WSUrl, nil)
	if err != nil {
		fmt.Println("Websocket client init error" + err.Error())
		return
	}

	//client := global.SseClientData(notifyUrl)
	client := global.SseClientData
	client.URL = notifyUrl
	err = client.SubscribeRaw(func(msg *sse.Event) {
		bytes := msg.Data
		//s := string(bytes)
		//fmt.Println("notify receive data :" + s)
		//脑电数据
		var receiveData gateway.DeviceDataReceived
		errJson := json.Unmarshal(msg.Data, &receiveData)
		if errJson != nil {
			fmt.Println("receive data parse error:" + errJson.Error())
			return
		}
		messageMap := make(map[string]string)
		if strings.Contains(receiveData.Value, "AAAA") {
			messageMap["msgType"] = constant.MessageTypeEEGData
			marshal, _ := json.Marshal(receiveData)
			messageMap["content"] = string(marshal)
		} else { //心电数据
			flag := strings.HasPrefix(receiveData.Value, "E840") || strings.HasPrefix(receiveData.Value, "E841") || strings.HasPrefix(receiveData.Value, "E823")
			if !flag {
				//var ecgData []int
				ecgData := [12]int{}
				fmt.Println("收到的心电数据：" + string(msg.Data))
				data := []byte(receiveData.Value[4:])
				fmt.Println(len(data))
				if len(data) == 36 {
					for i := 0; i < 36; i += 3 {
						ecgData[i/3] = int(16<<data[i]&0xFF | 8<<data[i+1]&0xFF | data[i+2]&0xFF)
					}
				} else {
					for i := len(data); i < 36; i++ {
						data = append(data, data[i])
					}
					for i := 0; i < 36; i += 3 {
						ecgData[i/3] = int(16<<data[i]&0xFF | 8<<data[i+1]&0xFF | data[i+2]&0xFF)
					}
				}
				messageMap["msgType"] = constant.MessageTypeECGData
				var dataStr []string
				for _, i := range ecgData {
					dataStr = append(dataStr, strconv.Itoa(i))
				}
				receiveData.Value = strings.Join(dataStr, ",")
				marshal, _ := json.Marshal(receiveData)
				messageMap["content"] = string(marshal)
			}
		}

		messageMap["Sender"] = "server"
		messageMap["Recipient"] = "client"
		bytes, err := json.Marshal(messageMap)
		err = ws.WriteMessage(websocket.TextMessage, bytes)
		if err != nil {
			fmt.Println(errors.SendMessageError + err.Error())
			return
		}
	})
	if err != nil {
		return
	}
}