optimize real time data pull api

diagram/redis_zset.go

@@ -3,6 +3,7 @@ package diagram
 
 import (
 	"context"
+	"fmt"
 	"iter"
 	"maps"
 
@@ -70,6 +71,76 @@ func (rs *RedisZSet) ZRANGE(setKey string, start, stop int64) ([]string, error)
 	return results, nil
 }
 
+func (rs *RedisZSet) FindEventsByTimeRange(ctx context.Context, key string, startTS, endTS string, withScores bool) ([]redis.Z, error) {
+	// ZRANGEBYLEX 的范围参数必须是字符串，并以 [ 或 ( 开头，或者使用特殊值 - 和 +
+	// 例如：[173...0000 [173...9999
+	min := fmt.Sprintf("[%s", startTS)
+	max := fmt.Sprintf("[%s", endTS)
+
+	// ZRangeByLexOptions 用于设置查询范围和限制
+	opts := redis.ZRangeBy{
+		Min:    min, // 范围的起始（包含）
+		Max:    max, // 范围的结束（包含）
+		Offset: 0,   // 分页偏移
+		Count:  -1,  // -1 表示不限制数量
+	}
+
+	// ZRangeByLex 会返回 Member 列表，它们将按字典序（即时间戳）升序排列
+	if withScores {
+		// return rs.storageClient.ZRangeByLexWithScores(ctx, key, &opts).Result()
+		var results []redis.Z
+		return results, nil
+	} else {
+		members, err := rs.storageClient.ZRangeByLex(ctx, key, &opts).Result()
+		if err != nil {
+			return nil, err
+		}
+		// ZRangeByLex 返回 []string，这里转换为 []redis.Z 结构体以保持一致性
+		var results []redis.Z
+		for _, member := range members {
+			results = append(results, redis.Z{Member: member})
+		}
+		return results, nil
+	}
+}
+
+// FindLatestEventsByTime 查找最新的 N 个事件（按时间降序）
+func (rs *RedisZSet) FindLatestEventsByTime(ctx context.Context, key string, limit int64, withScores bool) ([]redis.Z, error) {
+	// ZREVRANGEBYLEX 用于按字典序降序（即时间倒序）查找
+
+	// 对于 ZREVRANGEBYLEX，Min/Max 字段的语义与 ZRANGEBYLEX 相同，但排序是相反的。
+	// 为了获取全部，范围设置为 + (Max) 到 - (Min)，表示从最大的字符串开始往最小的字符串检索。
+	opts := redis.ZRangeBy{
+		Min:    "-", // 最小的字符串
+		Max:    "+", // 最大的字符串
+		Offset: 0,
+		Count:  limit, // 限制返回的数量
+	}
+
+	if withScores {
+		// ZRevRangeByLexWithScores
+		members, err := rs.storageClient.ZRangeByLex(ctx, key, &opts).Result()
+		fmt.Println(err)
+		var results []redis.Z
+		for _, member := range members {
+			results = append(results, redis.Z{Member: member})
+		}
+		return results, nil
+		// return rs.storageClient.ZRevRangeByLexWithScores(ctx, key, &opts).Result()
+	} else {
+		// ZRevRangeByLex
+		members, err := rs.storageClient.ZRevRangeByLex(ctx, key, &opts).Result()
+		if err != nil {
+			return nil, err
+		}
+		var results []redis.Z
+		for _, member := range members {
+			results = append(results, redis.Z{Member: member})
+		}
+		return results, nil
+	}
+}
+
 type Comparer[T any] interface {
 	Compare(T) int
 }

handler/real_time_data_pull.go

@@ -0,0 +1,206 @@
+// Package handler provides HTTP handlers for various endpoints.
+package handler
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"modelRT/logger"
+	"modelRT/network"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+)
+
+var pullUpgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024,
+	CheckOrigin: func(_ *http.Request) bool {
+		return true
+	},
+}
+
+// PullRealTimeDataHandler define real time data pull API
+// @Summary 实时数据拉取 websocket api
+// @Description 根据用户输入的clientID拉取对应的实时数据
+// @Tags RealTime Component Websocket
+// @Router /monitors/data/realtime/stream/:clientID [get]
+func PullRealTimeDataHandler(c *gin.Context) {
+	clientID := c.Param("clientID")
+	if clientID == "" {
+		err := fmt.Errorf("clientID is missing from the path")
+		logger.Error(c, "query clientID from path failed", "error", err, "url", c.Request.RequestURI)
+		c.JSON(http.StatusOK, network.FailureResponse{
+			Code: http.StatusBadRequest,
+			Msg:  err.Error(),
+		})
+		return
+	}
+
+	conn, err := pullUpgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		logger.Error(c, "upgrade http protocol to websocket protocal failed", "error", err)
+		c.JSON(http.StatusOK, network.FailureResponse{
+			Code: http.StatusBadRequest,
+			Msg:  err.Error(),
+		})
+		return
+	}
+	defer conn.Close()
+
+	ctx, cancel := context.WithCancel(c.Request.Context())
+	defer cancel()
+	fanInChan := make(chan []float64)
+	go processTargetPolling(ctx, globalMonitorState, clientID, fanInChan)
+
+	// 主循环：负责接收客户端消息（如心跳包、停止拉取命令等）
+	go readClientMessages(ctx, conn, clientID)
+
+	for {
+		select {
+		case value, ok := <-fanInChan:
+			// 从扇入通道拿去数据后，将数据写入websocket 返回流中
+			sendRealTimeDataStream(conn, clientID)
+			fmt.Println(value, ok)
+		default:
+			fmt.Println("default")
+		}
+	}
+}
+
+// readClientMessages 负责持续监听客户端发送的消息（例如 Ping/Pong, Close Frame, 或控制命令）
+func readClientMessages(ctx context.Context, conn *websocket.Conn, clientID string) {
+	// conn.SetReadLimit(512)
+	for {
+		msgType, _, err := conn.ReadMessage()
+		if err != nil {
+			// **【核心逻辑】判断是否为客户端主动正常关闭**
+			if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
+				fmt.Printf("客户端 %s 主动正常关闭连接 (Code 1000)。\n", clientID)
+				// 客户端主动发起的正常关闭，这是最清晰的主动退出信号。
+			} else if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
+				// 这是您代码中已有的逻辑，用于处理非正常或服务器离开的关闭（如网络中断、浏览器关闭 Tab 但未发送关闭帧）。
+				logger.Error(ctx, "clientID: %s 读取时发生错误: %v", clientID, err)
+			} else {
+				// 处理其他读取错误（如 I/O 错误）
+				logger.Error(ctx, "clientID: %s 读取时发生错误: %v", clientID, err)
+				fmt.Printf("客户端 %s 读取时发生未知错误: %v\n", clientID, err)
+			}
+
+			break // 退出循环，断开连接
+		}
+
+		// 客户端发送关闭帧时，msgType 可能是 websocket.CloseMessage，但通常在 ReadMessage 中被转换为错误。
+		// 如果客户端主动发送了数据（非关闭帧），在这里继续处理
+		if msgType == websocket.TextMessage || msgType == websocket.BinaryMessage {
+			fmt.Println(msgType)
+			// ... 处理正常接收到的消息 ...
+		}
+	}
+}
+
+// sendRealTimeDataStream 负责持续向客户端推送实时数据
+func sendRealTimeDataStream(conn *websocket.Conn, clientID string) {
+	fmt.Println(conn, clientID)
+	// ⚠️ 这是一个模拟的推送逻辑，您需要替换为您的实际数据源
+
+	// 模拟数据源
+	// dataChannel := globalMonitorState.GetDataChannel(clientID)
+
+	// 持续监听数据，并写入 WebSocket
+	// for data := range dataChannel {
+	//     err := conn.WriteJSON(data) // 使用 WriteJSON 发送结构化数据
+	//     if err != nil {
+	//         log.Printf("clientID: %s 写入数据失败: %v", clientID, err)
+	//         break // 写入失败，断开连接
+	//     }
+	// }
+}
+
+// processTargetPolling define function to process target in monitor map and data is continuously retrieved from redis based on the target
+func processTargetPolling(ctx context.Context, s *SharedMonitorState, clientID string, fanInChan chan []float64) {
+	stopChanMap := make(map[string]chan struct{})
+	s.mutex.RLock()
+
+	config, confExist := s.monitorMap[clientID]
+	if !confExist {
+		logger.Error(ctx, "can not found config into local stored map by clientID", "clientID", clientID)
+		return
+	}
+
+	for interval, componentItems := range config.components {
+		fmt.Println(componentItems)
+		for _, target := range componentItems.targets {
+			dataSize, exist := componentItems.targetParam[target]
+			if !exist {
+				logger.Error(ctx, "can not found subscription node param into param map", "target", target)
+				continue
+			}
+
+			stopChan := make(chan struct{})
+			// store stop channel with target into map
+			stopChanMap[target] = stopChan
+			go realTimeDataQueryFromRedis(ctx, interval, dataSize, fanInChan, stopChan)
+		}
+	}
+	s.mutex.RUnlock()
+
+	// TODO 持续监听noticeChannel
+	for {
+		select {
+		case noticeValue, ok := <-config.noticeChan:
+			if !ok {
+				logger.Error(ctx, "notice channel was closed unexpectedly", "clientID", clientID)
+				stopAllPolling(ctx, stopChanMap)
+				return
+			}
+			// TODO 判断传递数据类型，决定是调用新增函数或者移除函数
+			fmt.Println(noticeValue, ok)
+		case <-ctx.Done():
+			logger.Info(ctx, fmt.Sprintf("stop all data retrieval goroutines under this clientID:%s", clientID))
+			stopAllPolling(ctx, stopChanMap)
+			return
+		}
+	}
+}
+
+func stopAllPolling(ctx context.Context, stopChanMap map[string]chan struct{}) {
+	for target, stopChan := range stopChanMap {
+		logger.Info(ctx, fmt.Sprintf("stop the data fetching behavior for the corresponding target:%s", target))
+		close(stopChan)
+	}
+	clear(stopChanMap)
+	return
+}
+
+// TODO 根据Measure表 datasource 字段从 redis 查询信息
+func realTimeDataQueryFromRedis(ctx context.Context, interval string, dataSize int, fanInChan chan []float64, stopChan chan struct{}) {
+	duration, err := time.ParseDuration(interval)
+	if err != nil {
+		logger.Error(ctx, "failed to parse the time string", "interval", interval, "error", err)
+		return
+	}
+	ticker := time.NewTicker(duration * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			// TODO 添加 redis 查询逻辑
+			result := make([]float64, dataSize)
+			select {
+			case fanInChan <- result:
+			default:
+				// TODO 考虑如果 fanInChan 阻塞，避免阻塞查询循环，可以丢弃数据或记录日志
+				// logger.Warning("Fan-in channel is full, skipping data push.")
+			}
+
+		case <-stopChan:
+			// TODO 优化日志输出
+			logger.Info(ctx, "redis query goroutine have stopped")
+			return
+		}
+	}
+}

handler/real_time_data_query.go

@@ -3,7 +3,6 @@ package handler
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	"net/url"
 	"time"
@@ -79,8 +78,6 @@ func QueryRealTimeDataHandler(c *gin.Context) {
 	}
 	defer conn.Close()
 
-	fmt.Println(request)
-
 	// start a goroutine to open a websocket service with the dataRT service and use the channel to pass data back. Start and maintain the websocket connection with the front-end UI in the local api
 	transportChannel := make(chan []any, 100)
 	closeChannel := make(chan struct{})

handler/real_time_data_subscription.go

@@ -23,8 +23,8 @@ func init() {
 	globalMonitorState = NewSharedMonitorState()
 }
 
-// RealTimeMonitorHandler define real time data monitor process API
+// RealTimeSubHandler define real time data subscriptions process API
-// @Summary 开始或结束实时数据监控
+// @Summary 开始或结束订阅实时数据
 // @Description 根据用户输入的组件token,从 modelRT 服务中开始或结束对于实时数据的监控
 // @Tags RealTime Component
 // @Accept json
@@ -72,8 +72,8 @@ func init() {
 //		}
 //	}
 //
-// @Router /data/realtime [post]
+// @Router /monitors/data/subscriptions [post]
-func RealTimeMonitorHandler(c *gin.Context) {
+func RealTimeSubHandler(c *gin.Context) {
 	var request network.RealTimeSubRequest
 	var monitorAction string
 	var monitorID string
@@ -87,7 +87,7 @@ func RealTimeMonitorHandler(c *gin.Context) {
 		return
 	}
 
-	if request.Action == constants.MonitorStartAction && request.MonitorID == "" {
+	if request.Action == constants.MonitorStartAction && request.ClientID == "" {
 		monitorAction = request.Action
 		id, err := uuid.NewV4()
 		if err != nil {
@@ -99,12 +99,12 @@ func RealTimeMonitorHandler(c *gin.Context) {
 			return
 		}
 		monitorID = id.String()
-	} else if request.Action == constants.MonitorStartAction && request.MonitorID != "" {
+	} else if request.Action == constants.MonitorStartAction && request.ClientID != "" {
 		monitorAction = constants.MonitorAppendAction
-		monitorID = request.MonitorID
+		monitorID = request.ClientID
-	} else if request.Action == constants.MonitorStopAction && request.MonitorID != "" {
+	} else if request.Action == constants.MonitorStopAction && request.ClientID != "" {
 		monitorAction = request.Action
-		monitorID = request.MonitorID
+		monitorID = request.ClientID
 	}
 
 	pgClient := database.GetPostgresDBClient()
@@ -121,7 +121,7 @@ func RealTimeMonitorHandler(c *gin.Context) {
 				Code: http.StatusBadRequest,
 				Msg:  err.Error(),
 				PayLoad: network.RealTimeSubPayload{
-					MonitorID:     monitorID,
+					ClientID:      monitorID,
 					TargetResults: results,
 				},
 			})
@@ -132,7 +132,7 @@ func RealTimeMonitorHandler(c *gin.Context) {
 			Code: http.StatusOK,
 			Msg:  "success",
 			PayLoad: network.RealTimeSubPayload{
-				MonitorID:     monitorID,
+				ClientID:      monitorID,
 				TargetResults: results,
 			},
 		})
@@ -145,7 +145,7 @@ func RealTimeMonitorHandler(c *gin.Context) {
 				Code: http.StatusBadRequest,
 				Msg:  err.Error(),
 				PayLoad: network.RealTimeSubPayload{
-					MonitorID:     monitorID,
+					ClientID:      monitorID,
 					TargetResults: results,
 				},
 			})
@@ -156,7 +156,7 @@ func RealTimeMonitorHandler(c *gin.Context) {
 			Code: http.StatusOK,
 			Msg:  "success",
 			PayLoad: network.RealTimeSubPayload{
-				MonitorID:     monitorID,
+				ClientID:      monitorID,
 				TargetResults: results,
 			},
 		})
@@ -169,7 +169,7 @@ func RealTimeMonitorHandler(c *gin.Context) {
 				Code: http.StatusBadRequest,
 				Msg:  err.Error(),
 				PayLoad: network.RealTimeSubPayload{
-					MonitorID:     monitorID,
+					ClientID:      monitorID,
 					TargetResults: results,
 				},
 			})
@@ -180,7 +180,7 @@ func RealTimeMonitorHandler(c *gin.Context) {
 			Code: http.StatusOK,
 			Msg:  "success",
 			PayLoad: network.RealTimeSubPayload{
-				MonitorID:     monitorID,
+				ClientID:      monitorID,
 				TargetResults: results,
 			},
 		})
@@ -194,7 +194,7 @@ func RealTimeMonitorHandler(c *gin.Context) {
 			Code: http.StatusBadRequest,
 			Msg:  err.Error(),
 			PayLoad: network.RealTimeSubPayload{
-				MonitorID:     monitorID,
+				ClientID:      monitorID,
 				TargetResults: results,
 			},
 		})
@@ -329,6 +329,7 @@ func (s *SharedMonitorState) AppendTargets(ctx context.Context, tx *gorm.DB, mon
 			}
 		}
 	}
+	// TODO 将增加的订阅配置传递给channel
 	config.noticeChan <- struct{}{}
 	return targetProcessResults, nil
 }
@@ -417,21 +418,16 @@ func (s *SharedMonitorState) UpsertTargets(ctx context.Context, tx *gorm.DB, mon
 	return targetProcessResults, nil
 }
 
-// Get define function to get value from SharedMonitorState
+// Get define function to get subscriptions config from SharedMonitorState
-func (s *SharedMonitorState) Get(monitorID, interval string) ([]string, bool) {
+func (s *SharedMonitorState) Get(clientID string) (*RealTimeMonitorConfig, bool) {
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()
 
-	config, ok := s.monitorMap[monitorID]
+	config, ok := s.monitorMap[clientID]
 	if !ok {
 		return nil, false
 	}
-
+	return config, true
-	component, ok := config.components[interval]
-	if !ok {
-		return nil, false
-	}
-	return component.targets, true
 }
 
 // RemoveTargets define function to remove targets in SharedMonitorState
@@ -510,6 +506,7 @@ func (s *SharedMonitorState) RemoveTargets(ctx context.Context, monitorID string
 		}
 	}
 
+	// TODO 将移除的订阅配置传递给channel
 	config.noticeChan <- struct{}{}
 	return targetProcessResults, nil
 }

network/real_time_data_request.go

@@ -17,8 +17,8 @@ type RealTimeQueryRequest struct {
 type RealTimeSubRequest struct {
 	// required: true
 	// enum: [start, stop]
-	Action    string `json:"action" example:"start" description:"请求的操作，例如 start/stop"`
+	Action   string `json:"action" example:"start" description:"请求的操作，例如 start/stop"`
-	MonitorID string `json:"monitor_id" example:"5d72f2d9-e33a-4f1b-9c76-88a44b9a953e" description:"用于标识不同client的监控请求ID"`
+	ClientID string `json:"client_id" example:"5d72f2d9-e33a-4f1b-9c76-88a44b9a953e" description:"用于标识不同client的监控请求ID"`
 	// required: true
 	Components []RealTimeComponentItem `json:"components" description:"定义不同的数据采集策略和目标"`
 }

network/response.go

@@ -31,6 +31,6 @@ type TargetResult struct {
 
 // RealTimeSubPayload define struct of real time data subscription request
 type RealTimeSubPayload struct {
-	MonitorID     string         `json:"monitor_id" example:"5d72f2d9-e33a-4f1b-9c76-88a44b9a953e" description:"用于标识不同client的监控请求ID"`
+	ClientID      string         `json:"monitor_id" example:"5d72f2d9-e33a-4f1b-9c76-88a44b9a953e" description:"用于标识不同client的监控请求ID"`
 	TargetResults []TargetResult `json:"targets"`
 }

router/monitor.go

@@ -10,5 +10,6 @@ import (
 // registerMonitorRoutes define func of register monitordata routes
 func registerMonitorRoutes(rg *gin.RouterGroup) {
 	g := rg.Group("/monitors/")
-	g.POST("realtime", handler.RealTimeMonitorHandler)
+	g.POST("/data/subscriptions", handler.RealTimeSubHandler)
+	g.GET("/data/realtime/stream/:clientID", handler.PullRealTimeDataHandler)
 }