partly works
This commit is contained in:
@ -26,7 +26,7 @@ func InputArchiver() {
|
||||
}
|
||||
|
||||
func handleMessage(message database.Message) {
|
||||
log.Printf("Archiving %", message)
|
||||
log.Printf("Archiving Timestamp: %s, Topic: %s, Payload: %s", message.Time, message.Topic, message.Payload)
|
||||
counter.S("Stored")
|
||||
}
|
||||
|
||||
|
@ -12,6 +12,7 @@ import "ma/counter"
|
||||
type Message struct {
|
||||
Topic string
|
||||
Payload []byte
|
||||
Retained bool
|
||||
}
|
||||
|
||||
var InputChannel chan Message = make(chan Message, 100)
|
||||
@ -20,24 +21,30 @@ var OutputChannel chan Message = make(chan Message, 100)
|
||||
var mqttClient MQTT.Client
|
||||
|
||||
func onMessageReceived(client MQTT.Client, message MQTT.Message) {
|
||||
// log.Printf("Message received, topic: %s, payload: %s\n", message.Topic(), message.Payload())
|
||||
//log.Printf("Message received, topic: %s, payload: %s\n", message.Topic(), message.Payload())
|
||||
m := Message {
|
||||
Topic: message.Topic(),
|
||||
Payload: message.Payload(),
|
||||
Retained: message.Retained(),
|
||||
}
|
||||
if m.Retained {
|
||||
counter.S("Skipped")
|
||||
//log.Println("Retained message skipped")
|
||||
return
|
||||
}
|
||||
for _, i := range config.Config.ExcludeTopics {
|
||||
if i == m.Topic {
|
||||
if TopicMatchesSubscription(m.Topic, i) {
|
||||
counter.S("Skipped")
|
||||
log.Println("Message skipped")
|
||||
//log.Println("Message skipped")
|
||||
return
|
||||
}
|
||||
}
|
||||
select {
|
||||
case InputChannel <- m:
|
||||
counter.S("Received")
|
||||
log.Println("Message sent to channel")
|
||||
//log.Println("Message sent to channel")
|
||||
default:
|
||||
log.Println("Channel full, message lost")
|
||||
//log.Println("Channel full, message lost")
|
||||
counter.F("Received")
|
||||
}
|
||||
}
|
||||
|
Reference in New Issue
Block a user