et-go/et_sink/sinkHandler.go

package et_sink

import (
	"gitea.anxinyun.cn/container/common_models"
	"gitea.anxinyun.cn/container/common_utils/configLoad"
	"gitea.anxinyun.cn/container/common_utils/storage/storageDBs"
	"log"
	"node/stages"
	"time"
)

type SinkHandler struct {
	stage            *stages.Stage
	storageConsumers []storageDBs.IStorageConsumer
	dataQueueRaw     []common_models.EsRaw
	dataQueueVib     []common_models.EsVbRaw
	dataQueueTheme   []common_models.EsTheme
	dataQueueGroup   []common_models.EsGroupTheme // 分组主题数据队列
	signBatch        chan bool
	batchCount       int
}

const defaultBatchCount = 1000

func NewSinkThemeHandler() *SinkHandler {
	the := &SinkHandler{
		stage:            stages.NewStage("Theme 数据存储"),
		storageConsumers: storageDBs.LoadIStorageConsumer(),
		dataQueueTheme:   make([]common_models.EsTheme, 0),
		signBatch:        make(chan bool, 1),
		batchCount:       defaultBatchCount,
	}
	go the.dumpThemeBatchMonitor()
	the.stage.AddProcess(the.sinkThemeToES)

	return the
}

func NewSinkGroupHandler() *SinkHandler {
	esAddresses := configLoad.LoadConfig().GetStringSlice("es.addresses")
	log.Printf("es addresses: %v", esAddresses)

	the := &SinkHandler{
		stage:            stages.NewStage("EsGroupTheme 数据存储"),
		storageConsumers: storageDBs.LoadIStorageConsumer(),
		dataQueueGroup:   make([]common_models.EsGroupTheme, 0),
		batchCount:       defaultBatchCount,
	}

	go the.onGroupData()
	the.stage.AddProcess(the.sinkGroupDataToES)

	return the
}

func NewSinkRawHandler() *SinkHandler {
	esAddresses := configLoad.LoadConfig().GetStringSlice("es.addresses")
	log.Printf("es addresses: %v", esAddresses)
	the := &SinkHandler{
		stage:            stages.NewStage("raws 数据存储"),
		storageConsumers: storageDBs.LoadIStorageConsumer(),
		dataQueueRaw:     make([]common_models.EsRaw, 0),
		batchCount:       defaultBatchCount,
	}
	go the.dumpRawBatchMonitor()
	the.stage.AddProcess(the.sinkRawData)
	return the
}

func (the *SinkHandler) GetStage() stages.Stage {
	return *the.stage
}

func (the *SinkHandler) sinkRawData(p *common_models.ProcessData) *common_models.ProcessData {
	go the.sinkRawDataToES(p.DeviceData)
	return p
}
func (the *SinkHandler) sinkRawDataToES(deviceData common_models.DeviceData) {

	switch deviceData.DataType {
	case common_models.RawTypeVib:
		vibData := deviceData.GetVibrationData()
		vbRaws := common_models.EsVbRaw{
			StructId:       deviceData.StructId,
			IotaDeviceName: deviceData.Name,
			Param:          vibData.FormatParams(),
			Data:           map[string]any{"raw": vibData.Data},
			CollectTime:    deviceData.AcqTime.Truncate(time.Millisecond),
			IotaDevice:     deviceData.DeviceId,
			CreateTime:     time.Now().Truncate(time.Millisecond),
		}
		the.dataQueueVib = append(the.dataQueueVib, vbRaws)
	case common_models.RawTypeDiag:
	default:
		esRaws := common_models.EsRaw{
			StructId:       deviceData.StructId,
			IotaDeviceName: deviceData.Name,
			Data:           deviceData.Raw,
			CollectTime:    deviceData.AcqTime.Truncate(time.Millisecond),
			Meta:           deviceData.DeviceInfo.DeviceMeta.GetOutputProps(),
			IotaDevice:     deviceData.DeviceId,
			CreateTime:     time.Now().Truncate(time.Millisecond),
		}
		the.dataQueueRaw = append(the.dataQueueRaw, esRaws)
	}

	if len(the.dataQueueRaw) >= the.batchCount || len(the.dataQueueVib) >= the.batchCount {
		the.signBatch <- true
	}
}

func (the *SinkHandler) dumpRawBatchMonitor() {
	for {
		select {
		case <-the.signBatch:
			log.Printf("批存储信号raw,监控器收到")
		case <-time.After(200 * time.Millisecond):
		}
		if len(the.dataQueueRaw) > 0 {
			count := len(the.dataQueueRaw)
			log.Printf("es写入dataQueueRaw数据 count====> %d", count)
			needDump := the.dataQueueRaw[:count]
			the.dataQueueRaw = the.dataQueueRaw[count:]
			the.dumpRaws(needDump)
		}

		if len(the.dataQueueVib) > 0 {
			count := len(the.dataQueueVib)
			log.Printf("es写入dataQueueVib数据 count====> %d", count)
			needDump := the.dataQueueVib[:count]
			the.dataQueueVib = the.dataQueueVib[count:]
			the.dumpVibRaws(needDump)
		}
	}

}

func (the *SinkHandler) dumpRaws(esRaws []common_models.EsRaw) {
	for i, consumer := range the.storageConsumers {
		log.Printf("[consumer-%d]存储raw数据 %d 条", i, len(esRaws))
		go consumer.SaveRaw(esRaws)
	}
}
func (the *SinkHandler) dumpVibRaws(esVbRaws []common_models.EsVbRaw) {
	for i, consumer := range the.storageConsumers {
		log.Printf("[consumer-%d]存储VbRaw数据 %d 条", i, len(esVbRaws))
		go consumer.SaveVib(esVbRaws)
	}
}

func (the *SinkHandler) dumpThemeBatchMonitor() {
	for {
		select {
		case <-the.signBatch:
			log.Printf("批存储信号Theme,监控器收到")
		case <-time.After(200 * time.Millisecond):
		}
		if len(the.dataQueueTheme) > 0 {
			count := len(the.dataQueueTheme)
			needDump := the.dataQueueTheme[:count]
			the.dataQueueTheme = the.dataQueueTheme[count:]
			the.dumpThemes(needDump)
		}
	}

}

func (the *SinkHandler) sinkThemeToES(p *common_models.ProcessData) *common_models.ProcessData {
	go the.sinkThemeData(p.Stations)
	return p
}
func (the *SinkHandler) sinkThemeData(stations []common_models.Station) {
	var EsThemes []common_models.EsTheme
	for _, station := range stations {
		esTheme := common_models.EsTheme{
			SensorName:      station.Info.Name,
			FactorName:      station.Info.Factor.Name,
			FactorProtoCode: station.Info.Proto.Code,
			Data:            station.Data.ThemeData,
			FactorProtoName: station.Info.Proto.Name,
			Factor:          station.Info.FactorId,
			CollectTime:     station.Data.CollectTime.Truncate(time.Millisecond),
			Sensor:          station.Info.Id,
			Structure:       station.Info.StructureId,
			IotaDevice:      station.Info.GetDeviceIdArray(),
			CreateTime:      time.Now().Truncate(time.Millisecond),
		}
		EsThemes = append(EsThemes, esTheme)
	}
	the.dataQueueTheme = append(the.dataQueueTheme, EsThemes...)
	if len(the.dataQueueTheme) >= the.batchCount {
		the.signBatch <- true
	}
}
func (the *SinkHandler) dumpThemes(esThemes []common_models.EsTheme) {
	for i, consumer := range the.storageConsumers {
		log.Printf("[consumer-%d]存储Theme数据 %d 条", i, len(esThemes))
		go consumer.SaveTheme(esThemes)
	}
}

// ******************* 分组主题数据存储 *********************
// onGroupData 监听分组主题数据
func (the *SinkHandler) onGroupData() {
	for {
		select {
		case <-the.signBatch:
		case <-time.After(200 * time.Millisecond):
		}

		if len(the.dataQueueGroup) > 0 {
			count := len(the.dataQueueGroup)
			needDump := the.dataQueueGroup[:count]
			the.dataQueueGroup = the.dataQueueGroup[count:]
			the.dumpGroupThemes(needDump)
		}
	}
}

func (the *SinkHandler) sinkGroupDataToES(p *common_models.ProcessData) *common_models.ProcessData {
	go the.sinkGroupData(p.Stations)
	return p
}

func (the *SinkHandler) sinkGroupData(stations []common_models.Station) {
	var EsThemes []common_models.EsGroupTheme
	for _, station := range stations {
		esTheme := common_models.EsGroupTheme{
			Structure:       station.Info.StructureId,
			GroupId:         station.Info.Group.Id,
			GroupName:       station.Info.Group.Name,
			Factor:          station.Info.FactorId,
			FactorName:      station.Info.Factor.Name,
			FactorProtoCode: station.Info.Proto.Code,
			FactorProtoName: station.Info.Proto.Name,
			Data:            station.Data.PyhData, // 分组下所有测点的主题数据
			CollectTime:     station.Data.CollectTime.Truncate(time.Millisecond),
			CreateTime:      time.Now().Truncate(time.Millisecond),
		}
		EsThemes = append(EsThemes, esTheme)
	}
	the.dataQueueGroup = append(the.dataQueueGroup, EsThemes...)
	if len(the.dataQueueGroup) >= the.batchCount {
		the.signBatch <- true
	}
}
func (the *SinkHandler) dumpGroupThemes(esThemes []common_models.EsGroupTheme) {
	for _, consumer := range the.storageConsumers {
		consumer.SaveGroupTheme(esThemes)
	}
}