wn/universal-data-ingest
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

optimize

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2023-12-04 13:56:39 +01:00
parent ac4d42620a
commit 8e0418791d
1 changed files with 19 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/udi/dispatcher/dispatcher.go
View File

@ -72,23 +72,27 @@ func InputDispatcher() {
log.Printf("Message arrived in inputDispatcher, topic: %s\n", mqttMessage.Topic) log.Printf("Message arrived in inputDispatcher, topic: %s\n", mqttMessage.Topic)
message := handler.MessageT { time.Now(), mqttMessage.Topic, string(mqttMessage.Payload) } message := handler.MessageT { time.Now(), mqttMessage.Topic, string(mqttMessage.Payload) }
archiverChannel <- message archiverChannel <- message
handleMessage(message)
}
}
}
for _, mapping := range config.Config.TopicMappings { func handleMessage(message handler.MessageT) {
log.Printf("Testing %s -> %s", mapping.Topics, mapping.Handler) for _, mapping := range config.Config.TopicMappings {
for _, subscribedTopic := range mapping.Topics { log.Printf("Testing %s -> %s", mapping.Topics, mapping.Handler)
log.Printf("Testing %s in %s", message.Topic, subscribedTopic) for _, subscribedTopic := range mapping.Topics {
if mqtt.TopicMatchesSubscription(message.Topic, subscribedTopic) { log.Printf("Testing %s in %s", message.Topic, subscribedTopic)
log.Printf("Handle message in handler %s", mapping.Handler) if mqtt.TopicMatchesSubscription(message.Topic, subscribedTopic) {
handler, exists := handlerMap[mapping.Handler] log.Printf("Handle message in handler %s", mapping.Handler)
if exists { handler, exists := handlerMap[mapping.Handler]
handler.Handle(message) if exists {
} else { handler.Handle(message)
log.Printf("Handler not found, message is lost") return
} } else {
} else { log.Printf("Handler not found, message is lost")
log.Printf("no match")
}
} }
} else {
log.Printf("no match")
} }
} }
} }