新增处理

tcpserver/handler.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"net"
+	"strings"
 	"time"
 )
 
@@ -34,21 +35,28 @@ func (h *TCPHandler) HandleClient(conn net.Conn) {
 	}
 
 	for {
-		message, err := reader.ReadBytes('\n')
+		//message, err := reader.ReadBytes('\n')
+		//if err != nil {
+		//	fmt.Println("客户端已断开连接:", err)
+		//	break
+		//}
+		message, err := readUntilDelimiter(reader, []byte("\r\n"))
 		if err != nil {
-			fmt.Println("客户端已断开连接:", err)
+			fmt.Println("Error reading message:", err)
 			break
 		}
-
 		// 去除末尾的换行符
 		message = bytes.TrimSpace(message)
-		fmt.Printf("收到消息来自 %s: %s\n", conn.RemoteAddr(), string(message))
 
-		if !json.Valid(message) {
-			fmt.Printf("来自客户端的数据非法 %s\n", conn.RemoteAddr())
-			conn.Close()
-			return
-		}
+		output := strings.ReplaceAll(string(message), "\t", "")
+		output = strings.ReplaceAll(output, "\n", "")
+		message = []byte(output)
+		fmt.Printf("收到消息来自 %s: %s\n", conn.RemoteAddr(), message)
+		//if !json.Valid(message) {
+		//	fmt.Printf("来自客户端的数据非法 %s\n", conn.RemoteAddr())
+		//	conn.Close()
+		//	return
+		//}
 
 		var fullMsg map[string]interface{}
 		if err := json.Unmarshal(message, &fullMsg); err != nil {
@@ -117,6 +125,34 @@ func (h *TCPHandler) HandleClient(conn net.Conn) {
 			// 广播 OTA 消息
 			broadcastMessage(message)
 
+		case "stop":
+			// 处理 客户端停止实时上报数据
+			if !client.IsAuth {
+				fmt.Printf("来自未授权客户端的消息 %s\n", conn.RemoteAddr())
+				conn.Close()
+				return
+			}
+			if err := h.Server.StopRealTimeReporting(client, message); err != nil {
+				fmt.Printf("客户端停止实时上报数据 错误: %v\n", err)
+				continue
+			}
+			// 广播 OTA 消息
+			broadcastMessage(message)
+
+		case "up":
+			// 处理 客户端定时上报数据
+			if !client.IsAuth {
+				fmt.Printf("来自未授权客户端的消息 %s\n", conn.RemoteAddr())
+				conn.Close()
+				return
+			}
+			if err := h.Server.TimingReporting(client, message); err != nil {
+				fmt.Printf("客户端定时上报数据 错误: %v\n", err)
+				continue
+			}
+			// 广播 OTA 消息
+			broadcastMessage(message)
+
 		default:
 			// 处理其他消息类型
 			if !client.IsAuth {
@@ -262,6 +298,31 @@ func (s *Server) RealTimeReporting(client *Client, message []byte) error {
 	if msg.Type != "start" {
 		return fmt.Errorf("unauthorized")
 	}
-	fmt.Printf("设备实时上报数据：%s\r\n", msg.Data)
+	fmt.Printf("设备实时上报数据：%v\r\n", msg)
+	return nil
+}
+
+func (s *Server) StopRealTimeReporting(client *Client, message []byte) error {
+	var msg Message
+	if err := json.Unmarshal(message, &msg); err != nil {
+		return err
+	}
+	if msg.Type != "stop" {
+		return fmt.Errorf("unauthorized")
+	}
+	fmt.Printf("设备停止实时上报数据：%s\r\n", msg.State)
+	return nil
+}
+
+// TimingReporting 定时上报数据
+func (s *Server) TimingReporting(client *Client, message []byte) error {
+	var msg Message
+	if err := json.Unmarshal(message, &msg); err != nil {
+		return err
+	}
+	if msg.Type != "up" {
+		return fmt.Errorf("unauthorized")
+	}
+	fmt.Printf("设备定时上报数据：%v\r\n", msg.MessageData)
 	return nil
 }

tcpserver/message.go

@@ -21,7 +21,13 @@ type MessageState struct {
 }
 
 type MessageData struct {
-	Data string `json:"Data,omitempty"`
+	UpDataStruct
+}
+
+type UpDataStruct struct {
+	Sum  int `json:"sum"`
+	Time int `json:"time"`
+	Mile int `json:"mile"`
 }
 
 type Message struct {
@@ -31,5 +37,5 @@ type Message struct {
 	MessageVer
 	MessageTime
 	MessageState
-	MessageData
+	MessageData `json:"Data"`
 }

tcpserver/tcpserver.go

@@ -1,6 +1,8 @@
 package tcpserver
 
 import (
+	"bufio"
+	"bytes"
 	"fmt"
 	"net"
 	"os"
@@ -223,3 +225,19 @@ func (s *Server) GetClient(id string) (*Client, bool) {
 	client, ok := s.clients[id]
 	return client, ok
 }
+
+func readUntilDelimiter(reader *bufio.Reader, delimiter []byte) ([]byte, error) {
+	var buffer bytes.Buffer
+	for {
+		chunk, err := reader.ReadBytes('\n')
+		if err != nil {
+			return nil, err
+		}
+		buffer.Write(chunk)
+		if bytes.HasSuffix(buffer.Bytes(), delimiter) {
+			break
+		}
+	}
+	data := buffer.Bytes()
+	return data[:len(data)-len(delimiter)], nil
+}