update current project state

2026-07-03 20:55:15 +08:00
parent 6056b1fdc5
commit 13c4d03055
13 changed files with 1084 additions and 185 deletions
@@ -28,6 +28,12 @@ type PublishParams struct {
 	Content json.RawMessage `json:"content,omitempty"`
 }

+const (
+	wsFrameRequest byte = 0x01
+	wsFrameReply   byte = 0x02
+	wsFrameCast    byte = 0x03
+)
+
 type InferenceHeartbeat struct {
 	DeviceUUID  string  `json:"device_uuid"`
 	PersonCount int     `json:"person_count"`
@@ -47,8 +53,6 @@ func (c *WSClient) IsConnected() bool {
 	return c.connected
 }

-
-
 // DeviceData struct
 type DeviceData struct {
 	DeviceUUID           string      `json:"device_uuid"`
@@ -71,10 +75,10 @@ type DeviceData struct {
 	HostUUID             string `json:"host_uuid"`
 	Address              string `json:"address"`
 	// New fields for new configuration format
-	FireLeaveDetectArea  []Area        `json:"fire_leave_detect_area,omitempty"`
-	FireCheckDetectArea  []Area        `json:"fire_check_detect_area,omitempty"`
-	FireLeaveParam       FireLeaveParam `json:"fire_leave_param,omitempty"`
-	FireCheckParam       FireCheckParam `json:"fire_check_param,omitempty"`
+	FireLeaveDetectArea []Area         `json:"fire_leave_detect_area,omitempty"`
+	FireCheckDetectArea []Area         `json:"fire_check_detect_area,omitempty"`
+	FireLeaveParam      FireLeaveParam `json:"fire_leave_param,omitempty"`
+	FireCheckParam      FireCheckParam `json:"fire_check_param,omitempty"`
 }

 type InferenceParams struct {
@@ -100,6 +104,12 @@ type WSError struct {
 	Message string `json:"message"`
 }

+type ServiceReply struct {
+	PacketID uint32
+	Result   []byte
+	Error    *WSError
+}
+
 type RegisterParams struct {
 	ContainerName string                 `json:"container_name,omitempty"`
 	ServiceID     string                 `json:"service_id,omitempty"`
@@ -117,8 +127,8 @@ type WSClient struct {
 	serverID      string
 	connected     bool
 	mutex         sync.Mutex
-	messageID     int64
-	callbacks     map[int64]chan *WSMessage
+	messageID     uint32
+	callbacks     map[uint32]chan *ServiceReply
 	subscribers   map[string]func([]byte)
 	closeChan     chan struct{}
 	reconnectFlag bool
@@ -136,7 +146,7 @@ func NewWSClient(url string) (*WSClient, error) {
 		url:           url,
 		connected:     true,
 		messageID:     1,
-		callbacks:     make(map[int64]chan *WSMessage),
+		callbacks:     make(map[uint32]chan *ServiceReply),
 		subscribers:   make(map[string]func([]byte)),
 		closeChan:     make(chan struct{}),
 		reconnectFlag: true,
@@ -243,15 +253,8 @@ func RegisterService(containerName string) (string, error) {
 		return "", fmt.Errorf("Service registration error: %s", response.Error.Message)
 	}

-	// Parse response
-	var resultStr string
-	if err := json.Unmarshal(response.Result, &resultStr); err != nil {
-		logger.Logger.Printf("Failed to parse service registration response: %v, Raw response: %s", err, string(response.Result))
-		return "", fmt.Errorf("Failed to parse service registration response: %v", err)
-	}
-
-	if resultStr != "ok" {
-		logger.Logger.Printf("Service registration failed: Expected result 'ok', got '%s'", resultStr)
+	if string(response.Result) != "ok" {
+		logger.Logger.Printf("Service registration failed: Expected result 'ok', got '%s'", string(response.Result))
 		return "", fmt.Errorf("Service registration failed: Server returned abnormal response")
 	}

@@ -286,17 +289,58 @@ func SubscribeTopic(topic string, callback func([]byte)) error {
 }

 func (c *WSClient) SendRawMessage(message interface{}) error {
+	messageData, err := json.Marshal(message)
+	if err != nil {
+		logger.Logger.Printf("[WebSocket Send Full JSON] marshal failed: %v", err)
+	} else {
+		logger.Logger.Printf("[WebSocket Send Full JSON] %s", string(messageData))
+	}
+
+	if metric, ok := extractMetricData(message); ok {
+		return c.sendMetricData(metric.routeKey, metric.metric)
+	}
+	return fmt.Errorf("unsupported websocket message type for protobuf protocol")
+}
+
+type metricDataMessage struct {
+	routeKey []byte
+	metric   []byte
+}
+
+func extractMetricData(message interface{}) (metricDataMessage, bool) {
+	msg, ok := message.(map[string]interface{})
+	if !ok || msg["method"] != "metric_data" {
+		return metricDataMessage{}, false
+	}
+	params, ok := msg["params"].(map[string]interface{})
+	if !ok {
+		return metricDataMessage{}, false
+	}
+	routeKey, ok := params["route_key"].(string)
+	if !ok || routeKey == "" {
+		return metricDataMessage{}, false
+	}
+	metric, ok := params["metric"].(string)
+	if !ok {
+		return metricDataMessage{}, false
+	}
+	return metricDataMessage{
+		routeKey: []byte(routeKey),
+		metric:   []byte(metric),
+	}, true
+}
+
+func (c *WSClient) sendBinaryFrame(frameType byte, payload []byte) error {
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 	if !c.connected {
 		return fmt.Errorf("WebSocket connection disconnected")
 	}
-	messageData, err := json.Marshal(message)
-	if err != nil {
-		return fmt.Errorf("Message serialization failed: %v", err)
-	}
-	logger.Logger.Printf("Sending WebSocket message: %s", string(messageData))
-	err = c.Conn.WriteMessage(websocket.TextMessage, messageData)
+	frame := make([]byte, 1+len(payload))
+	frame[0] = frameType
+	copy(frame[1:], payload)
+	logger.Logger.Printf("Sending WebSocket binary frame: type=0x%02x, payload=%d bytes", frameType, len(payload))
+	err := c.Conn.WriteMessage(websocket.BinaryMessage, frame)
 	if err != nil {
 		logger.Logger.Printf("WebSocket message sending failed: %v", err)
 		c.connected = false
@@ -306,27 +350,38 @@ func (c *WSClient) SendRawMessage(message interface{}) error {
 	return nil
 }

+func (c *WSClient) sendMetricData(routeKey, metric []byte) error {
+	payload := encodeServiceCastMetricData(routeKey, metric)
+	return c.sendBinaryFrame(wsFrameCast, payload)
+}
+
 // sendRequest sends request
-func (c *WSClient) sendRequest(method string, params interface{}, waitResponse bool) (*WSMessage, error) {
+func (c *WSClient) sendRequest(method string, params interface{}, waitResponse bool) (*ServiceReply, error) {
 	c.mutex.Lock()
 	id := c.messageID
 	c.messageID++
 	c.mutex.Unlock()

-	message := WSMessage{
-		ID:     id,
-		Method: method,
-	}
-	if params != nil {
-		paramsData, err := json.Marshal(params)
-		if err != nil {
-			return nil, fmt.Errorf("Parameter serialization failed: %v", err)
+	var payload []byte
+	switch method {
+	case "register":
+		registerParams, ok := params.(RegisterParams)
+		if !ok {
+			return nil, fmt.Errorf("invalid register params")
 		}
-		message.Params = paramsData
+		payload = encodeServiceRequestRegister(id, registerParams.ServiceID)
+	case "subscribe":
+		subscribeParams, ok := params.(SubscribeParams)
+		if !ok {
+			return nil, fmt.Errorf("invalid subscribe params")
+		}
+		payload = encodeServiceRequestSubscribe(id, subscribeParams.Topic)
+	default:
+		return nil, fmt.Errorf("unsupported request method: %s", method)
 	}

 	if waitResponse {
-		callback := make(chan *WSMessage, 1)
+		callback := make(chan *ServiceReply, 1)
 		c.mutex.Lock()
 		c.callbacks[id] = callback
 		c.mutex.Unlock()
@@ -338,7 +393,7 @@ func (c *WSClient) sendRequest(method string, params interface{}, waitResponse b
 			logger.Logger.Printf("Cleaning up callback channel: ID=%d", id)
 		}()

-		if err := c.SendRawMessage(message); err != nil {
+		if err := c.sendBinaryFrame(wsFrameRequest, payload); err != nil {
 			return nil, err
 		}

@@ -355,7 +410,7 @@ func (c *WSClient) sendRequest(method string, params interface{}, waitResponse b
 			return nil, fmt.Errorf("Connection closed")
 		}
 	}
-	return nil, c.SendRawMessage(message)
+	return nil, c.sendBinaryFrame(wsFrameRequest, payload)
 }

 // readLoop message reading loop
@@ -370,7 +425,7 @@ func (c *WSClient) readLoop() {
 	}()

 	for {
-		_, messageData, err := c.Conn.ReadMessage()
+		messageType, messageData, err := c.Conn.ReadMessage()
 		if err != nil {
 			logger.Logger.Printf("WebSocket message reading failed: %v", err)
 			c.mutex.Lock()
@@ -380,34 +435,51 @@ func (c *WSClient) readLoop() {
 			return
 		}

+		if messageType != websocket.BinaryMessage {
+			logger.Logger.Printf("[WebSocket Received Raw] non-binary type=%d size=%d payload=%s", messageType, len(messageData), string(messageData))
+			continue
+		}
+		if len(messageData) == 0 {
+			logger.Logger.Printf("Ignoring empty WebSocket binary frame")
+			continue
+		}
+
 		if string(messageData) == "pong" {
 			logger.Logger.Printf("Received pong response")
 			continue
 		}

-		logger.Logger.Printf("Received WebSocket message: %s", string(messageData))
+		frameType := messageData[0]
+		payload := messageData[1:]
+		logger.Logger.Printf("Received WebSocket binary frame: type=0x%02x, payload=%d bytes", frameType, len(payload))

-		var message WSMessage
-		if err := json.Unmarshal(messageData, &message); err != nil {
-			logger.Logger.Printf("WebSocket message parsing failed: %v", err)
-			continue
-		}
-
-		if message.ID > 0 {
+		switch frameType {
+		case wsFrameReply:
+			reply, err := decodeServiceReply(payload)
+			if err != nil {
+				logger.Logger.Printf("WebSocket reply parsing failed: %v", err)
+				continue
+			}
+			logServiceReply(reply)
 			c.mutex.Lock()
-			if callback, exists := c.callbacks[message.ID]; exists {
+			if callback, exists := c.callbacks[reply.PacketID]; exists {
 				select {
-				case callback <- &message:
-					logger.Logger.Printf("Successfully sent message to callback channel: ID=%d", message.ID)
+				case callback <- reply:
+					logger.Logger.Printf("Successfully sent reply to callback channel: ID=%d", reply.PacketID)
 				default:
-					logger.Logger.Printf("Callback channel full, discarding message: ID=%d", message.ID)
+					logger.Logger.Printf("Callback channel full, discarding reply: ID=%d", reply.PacketID)
 				}
+			} else {
+				logger.Logger.Printf("[WebSocket Received Reply] no waiting callback for packet_id=%d", reply.PacketID)
 			}
 			c.mutex.Unlock()
-		}
-
-		if message.Method == "publish" {
-			c.handlePublishMessage(message)
+		case wsFrameCast:
+			if err := c.handleServiceCast(payload); err != nil {
+				logger.Logger.Printf("WebSocket cast handling failed: %v", err)
+			}
+		default:
+			logger.Logger.Printf("Ignoring unknown WebSocket frame type: 0x%02x", frameType)
+			continue
 		}
 	}
 }
@@ -433,42 +505,49 @@ func (c *WSClient) handlePublishMessage(message WSMessage) {
 	}

 	if publishParams.Data != nil {
-		var deviceData DeviceData
-		if err := json.Unmarshal(publishParams.Data, &deviceData); err != nil {
-			logger.Logger.Printf("Failed to parse DeviceData: %v", err)
-			return
-		}
-
-		// Check if DeviceUUID is valid
-		if deviceData.DeviceUUID == "" {
-			logger.Logger.Printf("Invalid DeviceData: DeviceUUID is empty, ignoring")
-			return
-		}
-
-		logger.Logger.Printf("Received valid DeviceData: DeviceUUID=%s", deviceData.DeviceUUID)
-
-		// 1. Save original configuration to service config file (using SaveSingleServiceConfig)
-		if err := SaveSingleServiceConfig(deviceData); err != nil {
-			logger.Logger.Printf("Failed to save service configuration: %v", err)
-		}
-
-		// 2. Process data and save to device data file
-		processedData := ProcessDeviceDataForInference(deviceData)
-		if err := UpdateDeviceData(processedData); err != nil {
-			logger.Logger.Printf("Failed to update DeviceData: %v", err)
-			return
-		}
-
-		select {
-		case ConfigUpdateChan <- processedData:
-			logger.Logger.Printf("New DeviceData sent to config update channel: DeviceUUID=%s", processedData.DeviceUUID)
-		default:
-			logger.Logger.Printf("Config update channel full, discarding message")
-		}
-
+		logger.Logger.Printf("Ignoring publish.data device configuration payload; service.conf is managed by startup config only")
 	}
 }

+func (c *WSClient) handleServiceCast(payload []byte) error {
+	topic, content, ok, err := decodeServiceCastTopicEvent(payload)
+	if err != nil {
+		return err
+	}
+	if !ok {
+		logger.Logger.Printf("Ignoring unsupported ServiceCast body")
+		return nil
+	}
+
+	logger.Logger.Printf("WebSocket received topic_event: Topic=%s | Content=%s", topic, string(content))
+	c.mutex.Lock()
+	callback, exists := c.subscribers[topic]
+	c.mutex.Unlock()
+	if !exists || callback == nil {
+		logger.Logger.Printf("No subscriber for topic_event: Topic=%s", topic)
+		return nil
+	}
+
+	wrapped, err := wrapTopicEvent(topic, content)
+	if err != nil {
+		return err
+	}
+	callback(wrapped)
+	return nil
+}
+
+func wrapTopicEvent(topic string, content []byte) ([]byte, error) {
+	msg := map[string]interface{}{
+		"topic": topic,
+	}
+	if json.Valid(content) {
+		msg["content"] = json.RawMessage(content)
+	} else {
+		msg["content"] = string(content)
+	}
+	return json.Marshal(msg)
+}
+
 // heartbeatLoop heartbeat loop - fixed version
 func (c *WSClient) heartbeatLoop() {
 	logger.Logger.Printf("Starting WebSocket heartbeat loop")
@@ -575,11 +654,238 @@ func (d *DeviceData) ToInferenceParams(port int) InferenceParams {
 	}
 }

+func appendVarint(dst []byte, v uint64) []byte {
+	for v >= 0x80 {
+		dst = append(dst, byte(v)|0x80)
+		v >>= 7
+	}
+	return append(dst, byte(v))
+}

+func appendBytesField(dst []byte, fieldNumber int, value []byte) []byte {
+	dst = appendVarint(dst, uint64(fieldNumber<<3|2))
+	dst = appendVarint(dst, uint64(len(value)))
+	return append(dst, value...)
+}

+func appendVarintField(dst []byte, fieldNumber int, value uint64) []byte {
+	dst = appendVarint(dst, uint64(fieldNumber<<3))
+	return appendVarint(dst, value)
+}

+func encodeServiceRequestRegister(packetID uint32, serviceID string) []byte {
+	register := appendBytesField(nil, 1, []byte(serviceID))
+	payload := appendVarintField(nil, 1, uint64(packetID))
+	payload = appendBytesField(payload, 10, register)
+	return payload
+}

+func encodeServiceRequestSubscribe(packetID uint32, topic string) []byte {
+	subscribe := appendBytesField(nil, 1, []byte(topic))
+	payload := appendVarintField(nil, 1, uint64(packetID))
+	payload = appendBytesField(payload, 11, subscribe)
+	return payload
+}

+func encodeServiceCastMetricData(routeKey, metric []byte) []byte {
+	metricData := appendBytesField(nil, 1, routeKey)
+	metricData = appendBytesField(metricData, 2, metric)
+	return appendBytesField(nil, 11, metricData)
+}
+
+func consumeVarint(data []byte, pos *int) (uint64, error) {
+	var value uint64
+	var shift uint
+	for *pos < len(data) {
+		b := data[*pos]
+		*pos += 1
+		value |= uint64(b&0x7f) << shift
+		if b < 0x80 {
+			return value, nil
+		}
+		shift += 7
+		if shift >= 64 {
+			return 0, fmt.Errorf("protobuf varint overflow")
+		}
+	}
+	return 0, fmt.Errorf("unexpected end of protobuf varint")
+}
+
+func consumeBytes(data []byte, pos *int) ([]byte, error) {
+	length, err := consumeVarint(data, pos)
+	if err != nil {
+		return nil, err
+	}
+	if length > uint64(len(data)-*pos) {
+		return nil, fmt.Errorf("protobuf bytes field length out of range")
+	}
+	value := data[*pos : *pos+int(length)]
+	*pos += int(length)
+	return value, nil
+}
+
+func skipField(data []byte, pos *int, wireType int) error {
+	switch wireType {
+	case 0:
+		_, err := consumeVarint(data, pos)
+		return err
+	case 2:
+		_, err := consumeBytes(data, pos)
+		return err
+	default:
+		return fmt.Errorf("unsupported protobuf wire type: %d", wireType)
+	}
+}
+
+func decodeServiceReply(payload []byte) (*ServiceReply, error) {
+	reply := &ServiceReply{}
+	for pos := 0; pos < len(payload); {
+		key, err := consumeVarint(payload, &pos)
+		if err != nil {
+			return nil, err
+		}
+		fieldNumber := int(key >> 3)
+		wireType := int(key & 0x7)
+		switch fieldNumber {
+		case 1:
+			if wireType != 0 {
+				return nil, fmt.Errorf("invalid packet_id wire type: %d", wireType)
+			}
+			packetID, err := consumeVarint(payload, &pos)
+			if err != nil {
+				return nil, err
+			}
+			reply.PacketID = uint32(packetID)
+		case 10:
+			if wireType != 2 {
+				return nil, fmt.Errorf("invalid result wire type: %d", wireType)
+			}
+			result, err := consumeBytes(payload, &pos)
+			if err != nil {
+				return nil, err
+			}
+			reply.Result = result
+		case 11:
+			if wireType != 2 {
+				return nil, fmt.Errorf("invalid error wire type: %d", wireType)
+			}
+			errorPayload, err := consumeBytes(payload, &pos)
+			if err != nil {
+				return nil, err
+			}
+			reply.Error = decodeServiceReplyError(errorPayload)
+		default:
+			if err := skipField(payload, &pos, wireType); err != nil {
+				return nil, err
+			}
+		}
+	}
+	return reply, nil
+}
+
+func logServiceReply(reply *ServiceReply) {
+	if reply == nil {
+		logger.Logger.Printf("[WebSocket Received Reply] <nil>")
+		return
+	}
+	errorJSON := "null"
+	if reply.Error != nil {
+		if data, err := json.Marshal(reply.Error); err == nil {
+			errorJSON = string(data)
+		} else {
+			errorJSON = fmt.Sprintf(`{"marshal_error":%q}`, err.Error())
+		}
+	}
+	logger.Logger.Printf("[WebSocket Received Reply] packet_id=%d result=%s error=%s", reply.PacketID, string(reply.Result), errorJSON)
+}
+
+func decodeServiceReplyError(payload []byte) *WSError {
+	wsErr := &WSError{}
+	for pos := 0; pos < len(payload); {
+		key, err := consumeVarint(payload, &pos)
+		if err != nil {
+			return wsErr
+		}
+		fieldNumber := int(key >> 3)
+		wireType := int(key & 0x7)
+		switch fieldNumber {
+		case 1:
+			code, err := consumeVarint(payload, &pos)
+			if err != nil {
+				return wsErr
+			}
+			wsErr.Code = int(int32(code))
+		case 2:
+			msg, err := consumeBytes(payload, &pos)
+			if err != nil {
+				return wsErr
+			}
+			wsErr.Message = string(msg)
+		default:
+			_ = skipField(payload, &pos, wireType)
+		}
+	}
+	return wsErr
+}
+
+func decodeServiceCastTopicEvent(payload []byte) (string, []byte, bool, error) {
+	for pos := 0; pos < len(payload); {
+		key, err := consumeVarint(payload, &pos)
+		if err != nil {
+			return "", nil, false, err
+		}
+		fieldNumber := int(key >> 3)
+		wireType := int(key & 0x7)
+		if fieldNumber != 10 {
+			if err := skipField(payload, &pos, wireType); err != nil {
+				return "", nil, false, err
+			}
+			continue
+		}
+		if wireType != 2 {
+			return "", nil, false, fmt.Errorf("invalid topic_event wire type: %d", wireType)
+		}
+		topicPayload, err := consumeBytes(payload, &pos)
+		if err != nil {
+			return "", nil, false, err
+		}
+		topic, content, err := decodeTopicEvent(topicPayload)
+		return topic, content, true, err
+	}
+	return "", nil, false, nil
+}
+
+func decodeTopicEvent(payload []byte) (string, []byte, error) {
+	var topic string
+	var content []byte
+	for pos := 0; pos < len(payload); {
+		key, err := consumeVarint(payload, &pos)
+		if err != nil {
+			return "", nil, err
+		}
+		fieldNumber := int(key >> 3)
+		wireType := int(key & 0x7)
+		switch fieldNumber {
+		case 1:
+			value, err := consumeBytes(payload, &pos)
+			if err != nil {
+				return "", nil, err
+			}
+			topic = string(value)
+		case 2:
+			value, err := consumeBytes(payload, &pos)
+			if err != nil {
+				return "", nil, err
+			}
+			content = value
+		default:
+			if err := skipField(payload, &pos, wireType); err != nil {
+				return "", nil, err
+			}
+		}
+	}
+	return topic, content, nil
+}

 func (c *WSClient) SendAsync(message interface{}) {
 	go func() {