modelRT/pool/concurrency_anchor_parse.go

// Package pool define concurrency call function in ants
package pool

import (
	"fmt"
	"time"

	"modelRT/alert"
	"modelRT/config"
	"modelRT/constants"
	"modelRT/diagram"
	"modelRT/logger"

	"github.com/panjf2000/ants/v2"
)

// AnchorRealTimePool define anchor param pool of real time data
var AnchorRealTimePool *ants.PoolWithFunc

// AnchorPoolInit define anchor param pool init
func AnchorPoolInit(concurrentQuantity int) (pool *ants.PoolWithFunc, err error) {
	// init anchor param ants pool
	AnchorRealTimePool, err = ants.NewPoolWithFunc(concurrentQuantity, AnchorFunc)
	if err != nil {
		return nil, err
	}
	return AnchorRealTimePool, nil
}

// AnchorFunc defines func that process the real time data of component anchor params
var AnchorFunc = func(poolConfig interface{}) {
	var firstStart bool
	alertManager := alert.GetAlertMangerInstance()

	anchorChanConfig, ok := poolConfig.(config.AnchorChanConfig)
	if !ok {
		logger.Error(anchorChanConfig.Ctx, "conversion component anchor chan type failed")
		return
	}

	for {
		select {
		case <-anchorChanConfig.Ctx.Done():
			return
		case anchorParaConfig := <-anchorChanConfig.AnchorChan:
			if firstStart {
				close(anchorChanConfig.ReadyChan)
				firstStart = false
			}

			componentUUID := anchorParaConfig.ComponentUUID
			anchorRealTimeDatas := anchorParaConfig.AnchorRealTimeData

			for _, value := range anchorRealTimeDatas {
				anchorName, err := diagram.GetAnchorValue(componentUUID)
				if err != nil {
					logger.Error(anchorChanConfig.Ctx, "can not get anchor value from map by uuid", "component_uuid", componentUUID, "error", err)
					continue
				}

				if anchorName != anchorParaConfig.AnchorName {
					logger.Error(anchorChanConfig.Ctx, "anchor name not equal param config anchor value", "map_anchor_name", anchorName, "param_anchor_name", anchorParaConfig.AnchorName)
					continue
				}

				upperLimitVal := anchorParaConfig.CompareValUpperLimit
				lowerlimitVal := anchorParaConfig.CompareValLowerLimit
				compareValue := anchorParaConfig.CalculateFunc(value, anchorParaConfig.CalculateParams...)
				if compareValue > upperLimitVal || compareValue < lowerlimitVal {
					message := fmt.Sprintf("anchor param %s value out of range, value:%f, upper limit:%f, lower limit:%f", anchorName,
						compareValue, upperLimitVal, lowerlimitVal)
					event := alert.Event{
						ComponentUUID: componentUUID,
						AnchorName:    anchorName,
						Level:         constants.InfoAlertLevel,
						Message:       message,
						StartTime:     time.Now().Unix(),
					}
					alertManager.AddEvent(event)
				}
			}
		default:
		}
	}
}