modelRT/handler/real_time_data_query.go

// Package handler provides HTTP handlers for various endpoints.
package handler

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"net/url"

	"modelRT/alert"
	"modelRT/constants"
	"modelRT/logger"
	"modelRT/network"

	"github.com/bitly/go-simplejson"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
)

// QueryRealTimeDataHandler define query real time data process API
// @Summary 获取实时测点数据
// @Description 根据用户输入的组件token,从 dataRT 服务中持续获取测点实时数据
// @Tags RealTime Component
// @Accept json
// @Produce json
// @Param token query string true "测量点唯一标识符 (e.g.grid_1:zone_1:station_1:transformfeeder1_220.I_A_rms)"
// @Param begin query int true "查询起始时间 (Unix时间戳, e.g., 1761008266)"
// @Param end query int true "查询结束时间 (Unix时间戳, e.g., 1761526675)"
// @Success 200 {object} network.SuccessResponse{payload=network.RealTimeDataPayload} "返回实时数据成功"
//
//	@Example 200 {
//	    "code": 200,
//	    "msg": "success",
//	    "payload": {
//	        "input": "grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_A_rms",
//	        "sub_pos": [
//			   {
//				"time": 1736305467506000000,
//				"value": 1
//		        }
//		   	]
//		}
//	}
//
// @Failure 400 {object} network.FailureResponse "返回实时数据失败"
//
//	@Example 400 {
//	    "code": 400,
//	    "msg": "failed to get real time data from dataRT",
//	}
//
// @Router /data/realtime [get]
func QueryRealTimeDataHandler(c *gin.Context) {
	token := c.Query("token")
	beginStr := c.Query("begin")
	endStr := c.Query("end")

	// TODO 启动一个goroutine用来开启与dataRT服务的websocket服务，并使用channel 将数据传递回来，本地api中启动并维持与前端UI的websocket连接
	var transportChannel chan network.RealTimeDataPoint
	var closeChannel chan struct{}

	params := url.Values{}
	params.Set("token", token)
	params.Set("begin", beginStr)
	params.Set("end", endStr)
	go receiveRealTimeDataByWebSocket(c, params, transportChannel, closeChannel)

	for {
		select {
		case data := <-transportChannel:
			fmt.Println("receive real time data:", data)
		case <-closeChannel:
			fmt.Println("websocket connection closed")
		}
	}

	var level int
	var targetLevel constants.AlertLevel
	alertManger := alert.GetAlertMangerInstance()
	targetLevel = constants.AlertLevel(level)
	events := alertManger.GetRangeEventsByLevel(targetLevel)

	resp := network.SuccessResponse{
		Code: http.StatusOK,
		Msg:  "success",
		PayLoad: map[string]interface{}{
			"events": events,
		},
	}
	c.JSON(http.StatusOK, resp)
}

func receiveRealTimeDataByWebSocket(ctx context.Context, params url.Values, transportChannel chan network.RealTimeDataPoint, closeChannel chan struct{}) {
	serverURL := "ws://127.0.0.1:8888/ws/points"
	u, err := url.Parse(serverURL)
	if err != nil {
		logger.Error(ctx, "parse url failed", "error", err)
	}

	q := u.Query()
	for key, values := range params {
		for _, value := range values {
			q.Add(key, value)
		}
	}
	u.RawQuery = q.Encode()
	finalServerURL := u.String()

	conn, resp, err := websocket.DefaultDialer.Dial(finalServerURL, nil)
	if err != nil {
		logger.Error(ctx, "dialing websocket server failed", "error", err)
		if resp != nil {
			// TODO 优化错误判断
			log.Printf("HTTP Response Status: %s", resp.Status)
		}
		return
	}
	defer conn.Close()

	for {
		msgType, message, err := conn.ReadMessage()
		if err != nil {
			// check if it is an expected shutdown error
			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
				logger.Info(ctx, "connection closed normally")
			} else {
				logger.Error(ctx, "abnormal disconnection from websocket server", "err", err)
			}
			close(closeChannel)
			break
		}
		logger.Info(ctx, "received info from dataRT server", "msg_type", messageTypeToString(msgType), "message", string(message))
		// parse message by xxx struct
		var point network.RealTimeDataPoint
		js, err := simplejson.NewJson(message)
		if err != nil {
			logger.Error(ctx, "parse real time data from message failed", "message", string(message), "err", err)
			continue
		}

		time, err := js.Get("time").Int64()
		if err != nil {
			logger.Error(ctx, "parse time data from message json info", "time", js.Get("time"), "err", err)
			continue
		}
		point.Time = time

		value, err := js.Get("value").Float64()
		if err != nil {
			logger.Error(ctx, "parse value data from message json info", "value", js.Get("value"), "err", err)
			continue
		}
		point.Value = value
		transportChannel <- point
	}
	return
}

// messageTypeToString define func of auxiliary to convert message type to string
func messageTypeToString(t int) string {
	switch t {
	case websocket.TextMessage:
		return "TEXT"
	case websocket.BinaryMessage:
		return "BINARY"
	case websocket.PingMessage:
		return "PING"
	case websocket.PongMessage:
		return "PONG"
	case websocket.CloseMessage:
		return "CLOSE"
	default:
		return "UNKNOWN"
	}
}