dom
/
telegraf
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Fix panic on connection loss with undelivered messages (#6806)

This commit is contained in:
Daniel Nelson 2019-12-27 15:48:45 -08:00 committed by GitHub
parent ef7fd9d030
commit 8b73625492
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 2 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
plugins/inputs/mqtt_consumer/mqtt_consumer.go
View File

@ -187,6 +187,7 @@ func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error {
m.state = Disconnected
m.acc = acc.WithTracking(m.MaxUndeliveredMessages)
m.sem = make(semaphore, m.MaxUndeliveredMessages)
m.ctx, m.cancel = context.WithCancel(context.Background())
m.client = m.clientFactory(m.opts)
@ -215,7 +216,6 @@ func (m *MQTTConsumer) connect() error {
m.Log.Infof("Connected %v", m.Servers)
m.state = Connected
m.sem = make(semaphore, m.MaxUndeliveredMessages)
m.messages = make(map[telegraf.TrackingID]bool)
// Presistent sessions should skip subscription if a session is present, as
@ -254,12 +254,12 @@ func (m *MQTTConsumer) recvMessage(c mqtt.Client, msg mqtt.Message) {
for {
select {
case track := <-m.acc.Delivered():
<-m.sem
_, ok := m.messages[track.ID()]
if !ok {
// Added by a previous connection
continue
}
<-m.sem
// No ack, MQTT does not support durable handling
delete(m.messages, track.ID())
case m.sem <- empty{}: