add real time data measurement target update func

constants/subscription.go

@@ -8,17 +8,19 @@ const (
 	SubStopAction string = "stop"
 	// SubAppendAction define the real time subscription append action
 	SubAppendAction string = "append"
+	// SubUpdateAction define the real time subscription update action
+	SubUpdateAction string = "update"
 )
 
 // 定义状态常量
 const (
 	// SubSuccessCode define subscription success code
 	SubSuccessCode = "1001"
-	// SubSuccessCode define subscription failed code
+	// SubFailedCode define subscription failed code
 	SubFailedCode = "1002"
-	// RTDSuccessCode define real time data resturn success code
+	// RTDSuccessCode define real time data return success code
 	RTDSuccessCode = "1003"
-	// SubSuccessCode define real time data resturn failed code
+	// RTDFailedCode define real time data return failed code
 	RTDFailedCode = "1004"
 	// CancelSubSuccessCode define cancel subscription success code
 	CancelSubSuccessCode = "1005"
@@ -26,6 +28,10 @@ const (
 	CancelSubFailedCode = "1006"
 	// SubRepeatCode define subscription repeat code
 	SubRepeatCode = "1007"
+	// UpdateSubSuccessCode define update subscription success code
+	UpdateSubSuccessCode = "1008"
+	// UpdateSubFailedCode define update subscription failed code
+	UpdateSubFailedCode = "1009"
 )
 
 const (
@@ -33,9 +39,9 @@ const (
 	SubSuccessMsg = "subscription success"
 	// SubFailedMsg define subscription failed message
 	SubFailedMsg = "subscription failed"
-	// RTDSuccessMsg define real time data resturn success message
+	// RTDSuccessMsg define real time data return success message
 	RTDSuccessMsg = "real time data return success"
-	// RTDFailedMsg define real time data resturn failed message
+	// RTDFailedMsg define real time data return failed message
 	RTDFailedMsg = "real time data return failed"
 	// CancelSubSuccessMsg define cancel subscription success message
 	CancelSubSuccessMsg = "cancel subscription success"
@@ -43,6 +49,10 @@ const (
 	CancelSubFailedMsg = "cancel subscription failed"
 	// SubRepeatMsg define subscription repeat message
 	SubRepeatMsg = "subscription repeat in target interval"
+	// UpdateSubSuccessMsg define update subscription success message
+	UpdateSubSuccessMsg = "update subscription success"
+	// UpdateSubFailedMsg define update subscription failed message
+	UpdateSubFailedMsg = "update subscription failed"
 )
 
 // TargetOperationType define constant to the target operation type

handler/real_time_data_pull.go

@@ -218,6 +218,9 @@ func processTargetPolling(ctx context.Context, s *SharedSubState, clientID strin
 				appendTargets(ctx, config, stopChanMap, fanInChan, transportTargets.Targets)
 			case constants.OpRemove:
 				removeTargets(ctx, stopChanMap, transportTargets.Targets)
+			case constants.OpUpdate:
+				// TODO 处理更新操作
+				fmt.Println(11111)
 			}
 			config.mutex.Unlock()
 		case <-ctx.Done():

handler/real_time_data_subscription.go

@@ -41,12 +41,12 @@ func init() {
 //		    "payload": {
 //		        "targets": [
 //		            {
-//						"id": "grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_A_rms",
+//						"id": "grid1.zone1.station1.ns1.tag1.bay.I11_C_rms",
 //						"code": "1001",
 //	                    "msg": "subscription success"
 //	                },
 //	                {
-//	                    "id": "grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_B_rms",
+//	                    "id": "grid1.zone1.station1.ns1.tag1.bay.I11_B_rms",
 //	                    "code": "1002",
 //	                    "msg": "subscription failed"
 //	                }
@@ -62,12 +62,12 @@ func init() {
 //		"payload": {
 //		    "targets": [
 //		        {
-//					"id": "grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_A_rms",
+//					"id": "grid1.zone1.station1.ns1.tag1.bay.I11_A_rms",
 //					"code": "1002",
 //	                "msg": "subscription failed"
 //	            },
 //	            {
-//	                "id": "grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_B_rms",
+//	                "id": "grid1.zone1.station1.ns1.tag1.bay.I11_B_rms",
 //	                "code": "1002",
 //	                "msg": "subscription failed"
 //	            }
@@ -108,6 +108,9 @@ func RealTimeSubHandler(c *gin.Context) {
 	} else if request.Action == constants.SubStopAction && request.ClientID != "" {
 		subAction = request.Action
 		clientID = request.ClientID
+	} else if request.Action == constants.SubUpdateAction && request.ClientID != "" {
+		subAction = request.Action
+		clientID = request.ClientID
 	}
 
 	pgClient := database.GetPostgresDBClient()
@@ -179,6 +182,30 @@ func RealTimeSubHandler(c *gin.Context) {
 			return
 		}
 
+		c.JSON(http.StatusOK, network.SuccessResponse{
+			Code: http.StatusOK,
+			Msg:  "success",
+			Payload: network.RealTimeSubPayload{
+				ClientID:      clientID,
+				TargetResults: results,
+			},
+		})
+		return
+	case constants.SubUpdateAction:
+		results, err := globalSubState.UpdateTargets(c, tx, clientID, request.Measurements)
+		if err != nil {
+			logger.Error(c, "update target to real time data subscription config failed", "error", err)
+			c.JSON(http.StatusOK, network.FailureResponse{
+				Code: http.StatusBadRequest,
+				Msg:  err.Error(),
+				Payload: network.RealTimeSubPayload{
+					ClientID:      clientID,
+					TargetResults: results,
+				},
+			})
+			return
+		}
+
 		c.JSON(http.StatusOK, network.SuccessResponse{
 			Code: http.StatusOK,
 			Msg:  "success",
@@ -237,8 +264,8 @@ func NewSharedSubState() *SharedSubState {
 	}
 }
 
-// processAndValidateTargets define func to perform all database I/O operations in a lock-free state (eg,ParseDataIdentifierToken)
+// processAndValidateTargetsForStart define func to perform all database I/O operations in a lock-free state for start action
-func processAndValidateTargets(ctx context.Context, tx *gorm.DB, measurements []network.RealTimeMeasurementItem, allReqTargetNum int) (
+func processAndValidateTargetsForStart(ctx context.Context, tx *gorm.DB, measurements []network.RealTimeMeasurementItem, allReqTargetNum int) (
 	[]network.TargetResult, []string,
 	map[string][]string,
 	map[string]*TargetPollingContext,
@@ -282,8 +309,62 @@ func processAndValidateTargets(ctx context.Context, tx *gorm.DB, measurements []
 	return targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap
 }
 
-// mergeMeasurements define func to merge newMeasurementsMap into existingMeasurementsMap
+// processAndValidateTargetsForUpdate define func to perform all database I/O operations in a lock-free state for update action
-func mergeMeasurements(config *RealTimeSubConfig, newMeasurements map[string][]string, newMeasurementsContextMap map[string]*TargetPollingContext) []string {
+func processAndValidateTargetsForUpdate(ctx context.Context, tx *gorm.DB, config *RealTimeSubConfig, measurements []network.RealTimeMeasurementItem, allReqTargetNum int) (
+	[]network.TargetResult, []string,
+	map[string][]string,
+	map[string]*TargetPollingContext,
+) {
+	targetProcessResults := make([]network.TargetResult, 0, allReqTargetNum)
+	newMeasMap := make(map[string][]string)
+	successfulTargets := make([]string, 0, allReqTargetNum)
+	newMeasContextMap := make(map[string]*TargetPollingContext)
+
+	for _, measurementItem := range measurements {
+		interval := measurementItem.Interval
+		for _, target := range measurementItem.Targets {
+			targetResult := network.TargetResult{ID: target}
+			if _, exist := config.targetContext[target]; !exist {
+				err := fmt.Errorf("target %s does not exists in subscription list", target)
+				logger.Error(ctx, "update target does not exist in subscription list", "error", err, "target", target)
+				targetResult.Code = constants.UpdateSubFailedCode
+				targetResult.Msg = fmt.Sprintf("%s: %s", constants.UpdateSubFailedMsg, err.Error())
+				targetProcessResults = append(targetProcessResults, targetResult)
+				continue
+			}
+
+			targetModel, err := database.ParseDataIdentifierToken(ctx, tx, target)
+			if err != nil {
+				logger.Error(ctx, "parse data indentity token failed", "error", err, "identity_token", target)
+				targetResult.Code = constants.UpdateSubFailedCode
+				targetResult.Msg = fmt.Sprintf("%s: %s", constants.UpdateSubFailedMsg, err.Error())
+				targetProcessResults = append(targetProcessResults, targetResult)
+				continue
+			}
+
+			targetResult.Code = constants.UpdateSubSuccessCode
+			targetResult.Msg = constants.UpdateSubSuccessMsg
+			targetProcessResults = append(targetProcessResults, targetResult)
+			successfulTargets = append(successfulTargets, target)
+
+			if _, ok := newMeasMap[interval]; !ok {
+				newMeasMap[interval] = make([]string, 0, len(measurementItem.Targets))
+			}
+
+			meas := newMeasMap[interval]
+			meas = append(meas, target)
+			newMeasMap[interval] = meas
+			newMeasContextMap[target] = &TargetPollingContext{
+				interval:    interval,
+				measurement: targetModel.GetMeasurementInfo(),
+			}
+		}
+	}
+	return targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap
+}
+
+// mergeMeasurementsForStart define func to merge newMeasurementsMap into existingMeasurementsMap for start action
+func mergeMeasurementsForStart(config *RealTimeSubConfig, newMeasurements map[string][]string, newMeasurementsContextMap map[string]*TargetPollingContext) []string {
 	allDuplicates := make([]string, 0)
 	for interval, newMeas := range newMeasurements {
 		if existingMeas, ok := config.measurements[interval]; ok {
@@ -307,10 +388,59 @@ func mergeMeasurements(config *RealTimeSubConfig, newMeasurements map[string][]s
 	return allDuplicates
 }
 
+// mergeMeasurementsForUpdate define func to merge newMeasurementsMap into existingMeasurementsMap for update action
+func mergeMeasurementsForUpdate(config *RealTimeSubConfig, newMeasurements map[string][]string, newMeasurementsContextMap map[string]*TargetPollingContext) ([]string, error) {
+	allDuplicates := make([]string, 0)
+	delMeasMap := make(map[string][]string)
+	for _, newMeas := range newMeasurements {
+		for _, measurement := range newMeas {
+			oldInterval := config.targetContext[measurement].interval
+			if _, ok := delMeasMap[oldInterval]; !ok {
+				delMeasurements := []string{measurement}
+				delMeasMap[oldInterval] = delMeasurements
+			} else {
+				delMeasurements := delMeasMap[oldInterval]
+				delMeasurements = append(delMeasurements, measurement)
+				delMeasMap[oldInterval] = delMeasurements
+			}
+		}
+	}
+
+	for interval, delMeas := range delMeasMap {
+		existingMeas, exist := config.measurements[interval]
+		if !exist {
+			return nil, fmt.Errorf("can not find exist measurements in %s interval", interval)
+		}
+
+		measurements := util.RemoveTargetsFromSliceSimple(existingMeas, delMeas)
+		config.measurements[interval] = measurements
+	}
+
+	for interval, newMeas := range newMeasurements {
+		if existingMeas, ok := config.measurements[interval]; ok {
+			deduplicated, duplicates := util.DeduplicateAndReportDuplicates(existingMeas, newMeas)
+
+			if len(duplicates) > 0 {
+				for _, duplicate := range duplicates {
+					delete(newMeasurementsContextMap, duplicate)
+				}
+				allDuplicates = append(allDuplicates, duplicates...)
+			}
+
+			if len(deduplicated) > 0 {
+				existingMeas = append(existingMeas, deduplicated...)
+				config.measurements[interval] = existingMeas
+				maps.Copy(config.targetContext, newMeasurementsContextMap)
+			}
+		}
+	}
+	return allDuplicates, nil
+}
+
 // CreateConfig define function to create config in SharedSubState
 func (s *SharedSubState) CreateConfig(ctx context.Context, tx *gorm.DB, clientID string, measurements []network.RealTimeMeasurementItem) ([]network.TargetResult, error) {
 	requestTargetsCount := processRealTimeRequestCount(measurements)
-	targetProcessResults, _, newMeasurementsMap, measurementContexts := processAndValidateTargets(ctx, tx, measurements, requestTargetsCount)
+	targetProcessResults, _, newMeasurementsMap, measurementContexts := processAndValidateTargetsForStart(ctx, tx, measurements, requestTargetsCount)
 	s.globalMutex.Lock()
 	if _, exist := s.subMap[clientID]; exist {
 		s.globalMutex.Unlock()
@@ -343,10 +473,10 @@ func (s *SharedSubState) AppendTargets(ctx context.Context, tx *gorm.DB, clientI
 		return processRealTimeRequestTargets(measurements, requestTargetsCount, err), err
 	}
 
-	targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap := processAndValidateTargets(ctx, tx, measurements, requestTargetsCount)
+	targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap := processAndValidateTargetsForStart(ctx, tx, measurements, requestTargetsCount)
 
 	config.mutex.Lock()
-	allDuplicates := mergeMeasurements(config, newMeasMap, newMeasContextMap)
+	allDuplicates := mergeMeasurementsForStart(config, newMeasMap, newMeasContextMap)
 	if len(allDuplicates) > 0 {
 		logger.Warn(ctx, "some targets are duplicate and have been ignored in append operation", "clientID", clientID, "duplicates", allDuplicates)
 		// process repeat target in targetProcessResults and successfulTargets
@@ -391,7 +521,7 @@ func filterAndDeduplicateRepeatTargets(resultsSlice []network.TargetResult, idsS
 // UpsertTargets define function to upsert targets in SharedSubState
 func (s *SharedSubState) UpsertTargets(ctx context.Context, tx *gorm.DB, clientID string, measurements []network.RealTimeMeasurementItem) ([]network.TargetResult, error) {
 	requestTargetsCount := processRealTimeRequestCount(measurements)
-	targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap := processAndValidateTargets(ctx, tx, measurements, requestTargetsCount)
+	targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap := processAndValidateTargetsForStart(ctx, tx, measurements, requestTargetsCount)
 
 	s.globalMutex.RLock()
 	config, exist := s.subMap[clientID]
@@ -401,7 +531,7 @@ func (s *SharedSubState) UpsertTargets(ctx context.Context, tx *gorm.DB, clientI
 	if exist {
 		opType = constants.OpUpdate
 		config.mutex.Lock()
-		mergeMeasurements(config, newMeasMap, newMeasContextMap)
+		mergeMeasurementsForStart(config, newMeasMap, newMeasContextMap)
 		config.mutex.Unlock()
 	} else {
 		opType = constants.OpAppend
@@ -415,7 +545,7 @@ func (s *SharedSubState) UpsertTargets(ctx context.Context, tx *gorm.DB, clientI
 		} else {
 			s.globalMutex.Unlock()
 			config.mutex.Lock()
-			mergeMeasurements(config, newMeasMap, newMeasContextMap)
+			mergeMeasurementsForStart(config, newMeasMap, newMeasContextMap)
 			config.mutex.Unlock()
 		}
 		s.globalMutex.Unlock()
@@ -527,6 +657,48 @@ func (s *SharedSubState) RemoveTargets(ctx context.Context, clientID string, mea
 	return targetProcessResults, nil
 }
 
+// UpdateTargets define function to update targets in SharedSubState
+func (s *SharedSubState) UpdateTargets(ctx context.Context, tx *gorm.DB, clientID string, measurements []network.RealTimeMeasurementItem) ([]network.TargetResult, error) {
+	requestTargetsCount := processRealTimeRequestCount(measurements)
+	targetProcessResults := make([]network.TargetResult, 0, requestTargetsCount)
+
+	s.globalMutex.RLock()
+	config, exist := s.subMap[clientID]
+	s.globalMutex.RUnlock()
+
+	if !exist {
+		s.globalMutex.RUnlock()
+		err := fmt.Errorf("clientID %s not found", clientID)
+		logger.Error(ctx, "clientID not found in remove targets operation", "error", err)
+		return processRealTimeRequestTargets(measurements, requestTargetsCount, err), err
+	}
+
+	targetProcessResults, successfulTargets, newMeasMap, newMeasContextMap := processAndValidateTargetsForUpdate(ctx, tx, config, measurements, requestTargetsCount)
+
+	config.mutex.Lock()
+	allDuplicates, err := mergeMeasurementsForUpdate(config, newMeasMap, newMeasContextMap)
+	if err != nil {
+		logger.Warn(ctx, "can not find exist measurements in target interval", "clientID", clientID, "duplicates", allDuplicates, "error", err)
+	}
+
+	if len(allDuplicates) > 0 {
+		logger.Warn(ctx, "some targets are duplicate and have been ignored in append operation", "clientID", clientID, "duplicates", allDuplicates)
+		// process repeat target in targetProcessResults and successfulTargets
+		targetProcessResults, successfulTargets = filterAndDeduplicateRepeatTargets(targetProcessResults, successfulTargets, allDuplicates)
+	}
+	config.mutex.Unlock()
+
+	if len(successfulTargets) > 0 {
+		transportTargets := &transportTargets{
+			OperationType: constants.OpUpdate,
+			Targets:       successfulTargets,
+		}
+		config.noticeChan <- transportTargets
+	}
+
+	return targetProcessResults, nil
+}
+
 // Get define function to get subscriptions config from SharedSubState
 func (s *SharedSubState) Get(clientID string) (*RealTimeSubConfig, bool) {
 	s.globalMutex.RLock()
@@ -539,12 +711,10 @@ func (s *SharedSubState) Get(clientID string) (*RealTimeSubConfig, bool) {
 	return config, true
 }
 
-// TODO 增加一个update 函数用来更新 interval
-
 func processRealTimeRequestCount(measurements []network.RealTimeMeasurementItem) int {
 	totalTargetsCount := 0
-	for _, compItem := range measurements {
+	for _, measItem := range measurements {
-		totalTargetsCount += len(compItem.Targets)
+		totalTargetsCount += len(measItem.Targets)
 	}
 	return totalTargetsCount
 }

network/real_time_data_request.go

@@ -26,18 +26,18 @@ type RealTimeSubRequest struct {
 // RealTimeMeasurementItem define struct of real time measurement item
 type RealTimeMeasurementItem struct {
 	Interval string   `json:"interval" example:"1" description:"数据采集的时间间隔（秒）"`
-	Targets  []string `json:"targets" example:"[\"grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_A_rms\",\"grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_B_rms\"]" description:"需要采集数据的测点或标签名称列表"`
+	Targets  []string `json:"targets" example:"[\"grid1.zone1.station1.ns1.tag1.bay.I11_A_rms\",\"grid1.zone1.station1.ns1.tag1.tag1.bay.I11_B_rms\"]" description:"需要采集数据的测点或标签名称列表"`
 }
 
 // RealTimePullPayload define struct of pull real time data payload
 type RealTimePullPayload struct {
 	// required: true
-	Targets []RealTimePullTarget `json:"targets" example:"{\"targets\":[{\"id\":\"grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_A_rms\",\"datas\":[{\"time\":1736305467506000000,\"value\":1},{\"time\":1736305467506000000,\"value\":1}]},{\"id\":\"grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_B_rms\",\"datas\":[{\"time\":1736305467506000000,\"value\":1},{\"time\":1736305467506000000,\"value\":1}]}]}" description:"实时数据"`
+	Targets []RealTimePullTarget `json:"targets" example:"{\"targets\":[{\"id\":\"grid1.zone1.station1.ns1.tag1.bay.I11_A_rms\",\"datas\":[{\"time\":1736305467506000000,\"value\":1},{\"time\":1736305467506000000,\"value\":1}]},{\"id\":\"grid1.zone1.station1.ns1.tag1.bay.I11_B_rms\",\"datas\":[{\"time\":1736305467506000000,\"value\":1},{\"time\":1736305467506000000,\"value\":1}]}]}" description:"实时数据"`
 }
 
 // RealTimePullTarget define struct of pull real time data target
 type RealTimePullTarget struct {
-	ID    string             `json:"id" example:"grid1.zone1.station1.ns1.tag1.transformfeeder1_220.I_A_rms" description:"实时数据ID值"`
+	ID    string             `json:"id" example:"grid1.zone1.station1.ns1.tag1.bay.I11_A_rms" description:"实时数据ID值"`
 	Datas []RealTimePullData `json:"datas" example:"[{\"time\":1736305467506000000,\"value\":220},{\"time\":1736305467506000000,\"value\":220}]" description:"实时数据值数组"`
 }