|
@@ -22,14 +22,12 @@ import (
|
|
|
|
|
|
// 所有共享变量
|
|
// 所有共享变量
|
|
var (
|
|
var (
|
|
- latestTimeWindowEnd = util.GetTimeCustom(time.Now())
|
|
|
|
- pjisuvParam = pjisuv_param.PjisuvParam{
|
|
|
|
|
|
+ pjisuvParam = pjisuv_param.PjisuvParam{
|
|
ObjDicOfTpperception: make(map[uint32][]float32),
|
|
ObjDicOfTpperception: make(map[uint32][]float32),
|
|
ObjTypeDicOfTpperception: make(map[uint32]uint8),
|
|
ObjTypeDicOfTpperception: make(map[uint32]uint8),
|
|
ObjSpeedDicOfTpperception: make(map[uint32]float64),
|
|
ObjSpeedDicOfTpperception: make(map[uint32]float64),
|
|
}
|
|
}
|
|
- // 每个触发器5秒触发一次
|
|
|
|
- triggerInterval = 3.0
|
|
|
|
|
|
+
|
|
// 保存时间窗口需要锁,防止数据竟态
|
|
// 保存时间窗口需要锁,防止数据竟态
|
|
saveTimeWindowMutex sync.Mutex
|
|
saveTimeWindowMutex sync.Mutex
|
|
// /cicv_location
|
|
// /cicv_location
|
|
@@ -41,13 +39,15 @@ var (
|
|
// /data_read
|
|
// /data_read
|
|
mutexOfDataRead sync.RWMutex
|
|
mutexOfDataRead sync.RWMutex
|
|
// /pj_vehicle_fdb_pub
|
|
// /pj_vehicle_fdb_pub
|
|
- mutexOfPjVehicleFdbPub sync.RWMutex
|
|
|
|
- // /pj_vehicle_fdb_pub
|
|
|
|
mutexOfCicvAmrTrajectory sync.RWMutex
|
|
mutexOfCicvAmrTrajectory sync.RWMutex
|
|
|
|
+
|
|
|
|
+ latestTimeWindowEnd = util.GetTimeCustom(time.Now())
|
|
|
|
+ // 每个触发器5秒触发一次
|
|
|
|
+ triggerInterval = 3.0
|
|
)
|
|
)
|
|
|
|
|
|
-// PrepareTimeWindowProducerQueue 负责监听所有主题并修改时间窗口
|
|
|
|
-func PrepareTimeWindowProducerQueue() {
|
|
|
|
|
|
+// 负责监听所有主题并修改时间窗口
|
|
|
|
+func ProduceWindow() {
|
|
c_log.GlobalLogger.Info("订阅者 goroutine,启动。")
|
|
c_log.GlobalLogger.Info("订阅者 goroutine,启动。")
|
|
|
|
|
|
var err error
|
|
var err error
|
|
@@ -497,13 +497,12 @@ func PrepareTimeWindowProducerQueue() {
|
|
})
|
|
})
|
|
}
|
|
}
|
|
|
|
|
|
- // 11
|
|
|
|
- if topic == masterConfig.TopicOfCicvAmrTrajectory && (len(masterConfig.RuleOfCicvAmrTrajectory1) > 0 || len(masterConfig.RuleOfCicvAmrTrajectory2) > 0) {
|
|
|
|
|
|
+ // 11 有共享变量的订阅者必须被创建
|
|
|
|
+ if topic == masterConfig.TopicOfCicvAmrTrajectory {
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
Node: commonConfig.RosNode,
|
|
Node: commonConfig.RosNode,
|
|
Topic: topic,
|
|
Topic: topic,
|
|
Callback: func(data *pjisuv_msgs.Trajectory) {
|
|
Callback: func(data *pjisuv_msgs.Trajectory) {
|
|
-
|
|
|
|
subscribersTimeMutexes[i].Lock()
|
|
subscribersTimeMutexes[i].Lock()
|
|
if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
|
|
if time.Since(subscribersTimes[i]).Seconds() > triggerInterval {
|
|
subscribersMutexes[i].Lock()
|
|
subscribersMutexes[i].Lock()
|
|
@@ -551,8 +550,8 @@ func PrepareTimeWindowProducerQueue() {
|
|
})
|
|
})
|
|
}
|
|
}
|
|
|
|
|
|
- // 12
|
|
|
|
- if topic == masterConfig.TopicOfCicvLocation && (len(masterConfig.RuleOfCicvLocation1) > 0 || len(masterConfig.RuleOfCicvLocation2) > 0) {
|
|
|
|
|
|
+ // 12 有共享变量的订阅者必须被创建
|
|
|
|
+ if topic == masterConfig.TopicOfCicvLocation {
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
Node: commonConfig.RosNode,
|
|
Node: commonConfig.RosNode,
|
|
Topic: topic,
|
|
Topic: topic,
|
|
@@ -1000,8 +999,8 @@ func PrepareTimeWindowProducerQueue() {
|
|
})
|
|
})
|
|
}
|
|
}
|
|
|
|
|
|
- // 22
|
|
|
|
- if topic == masterConfig.TopicOfPjControlPub && (len(masterConfig.RuleOfPjControlPub1) > 0 || len(masterConfig.RuleOfPjControlPub2) > 0) {
|
|
|
|
|
|
+ // 22 有共享变量的订阅者必须被创建
|
|
|
|
+ if topic == masterConfig.TopicOfPjControlPub {
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
Node: commonConfig.RosNode,
|
|
Node: commonConfig.RosNode,
|
|
Topic: topic,
|
|
Topic: topic,
|
|
@@ -1363,8 +1362,8 @@ func PrepareTimeWindowProducerQueue() {
|
|
})
|
|
})
|
|
}
|
|
}
|
|
|
|
|
|
- // 30
|
|
|
|
- if topic == masterConfig.TopicOfTpperception && (len(masterConfig.RuleOfTpperception1) > 0 || len(masterConfig.RuleOfTpperception2) > 0) {
|
|
|
|
|
|
+ // 30 有共享变量的订阅者必须被创建
|
|
|
|
+ if topic == masterConfig.TopicOfTpperception {
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
Node: commonConfig.RosNode,
|
|
Node: commonConfig.RosNode,
|
|
Topic: topic,
|
|
Topic: topic,
|
|
@@ -1612,6 +1611,7 @@ func PrepareTimeWindowProducerQueue() {
|
|
|
|
|
|
// 35
|
|
// 35
|
|
if topic == masterConfig.TopicOfCameraImage && (len(masterConfig.RuleOfCameraImage1) > 0 || len(masterConfig.RuleOfCameraImage2) > 0) {
|
|
if topic == masterConfig.TopicOfCameraImage && (len(masterConfig.RuleOfCameraImage1) > 0 || len(masterConfig.RuleOfCameraImage2) > 0) {
|
|
|
|
+
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
Node: commonConfig.RosNode,
|
|
Node: commonConfig.RosNode,
|
|
Topic: topic,
|
|
Topic: topic,
|
|
@@ -1653,8 +1653,9 @@ func PrepareTimeWindowProducerQueue() {
|
|
},
|
|
},
|
|
})
|
|
})
|
|
}
|
|
}
|
|
- // 36
|
|
|
|
- if topic == masterConfig.TopicOfDataRead && (len(masterConfig.RuleOfDataRead1) > 0 || len(masterConfig.RuleOfDataRead2) > 0) {
|
|
|
|
|
|
+
|
|
|
|
+ // 36 有共享变量的订阅者必须被创建
|
|
|
|
+ if topic == masterConfig.TopicOfDataRead {
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
subscribers[i], err = goroslib.NewSubscriber(goroslib.SubscriberConf{
|
|
Node: commonConfig.RosNode,
|
|
Node: commonConfig.RosNode,
|
|
Topic: topic,
|
|
Topic: topic,
|