diff --git a/node/app/app.go b/node/app/app.go index c598faf..169259a 100644 --- a/node/app/app.go +++ b/node/app/app.go @@ -6,6 +6,7 @@ import ( "et_analyze" "et_cache" "et_calc" + "et_calc/group" "et_print" "et_push" "et_sink" @@ -98,28 +99,22 @@ func addWorkStages(nodeStageManage *stages.StageManager) *stages.StageManager { // 单测点计算 calcHandler := et_calc.NewCalcHandler() nodeStageManage.AddStages(calcHandler.GetStage()) - + // 滑窗过滤 cacheHandler := et_cache.NewCacheHandler() nodeStageManage.AddStages(cacheHandler.GetStage()) + // 测点分组计算 + groupCalcHandler := group.NewGroupCalc() + nodeStageManage.AddStages(groupCalcHandler.GetStage()) // Theme 数据存储 sinkThemeHandler := et_sink.NewSinkThemeHandler() nodeStageManage.AddStages(sinkThemeHandler.GetStage()) - // 测点主题数据的阈值分析 + + // 测点阈值分析 stationAnalyzeHandler := et_analyze.NewThresholdHandler() nodeStageManage.AddStages(stationAnalyzeHandler.GetStage()) - // 测点分组计算 - //groupCalcHandler := group.NewGroupCalc() - //nodeStageManage.AddStages(groupCalcHandler.GetStage()) - // EsGroupTheme 数据存储 - //sinkGroupHandler := et_sink.NewSinkGroupHandler() - //nodeStageManage.AddStages(sinkGroupHandler.GetStage()) - - //// (阈值分析) 测点主题数据的阈值分析 + 分组计算后数据的阈值分析 - //groupAnalyzeHandler := et_analyze.NewThresholdHandler() - //nodeStageManage.AddStages(groupAnalyzeHandler.GetStage()) - + // 数据推送 publishHandler := et_push.NewPushHandler() nodeStageManage.AddStages(publishHandler.GetStage()) return nodeStageManage