From 21c7378b611303226d7d47f76c21732141d64ea5 Mon Sep 17 00:00:00 2001
From: Pascal Larin <pascal@mobility-centos.vm>
Date: Tue, 12 Apr 2016 12:36:43 -0400
Subject: [PATCH] Handle onConnect

---
 plugins/inputs/mqtt_consumer/mqtt_consumer.go | 26 ++++++++++++-------
 1 file changed, 16 insertions(+), 10 deletions(-)

diff --git a/plugins/inputs/mqtt_consumer/mqtt_consumer.go b/plugins/inputs/mqtt_consumer/mqtt_consumer.go
index c64d2139b..0cba92eb6 100644
--- a/plugins/inputs/mqtt_consumer/mqtt_consumer.go
+++ b/plugins/inputs/mqtt_consumer/mqtt_consumer.go
@@ -116,6 +116,9 @@ func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error {
 		return err
 	}
 
+	opts.OnConnect = onConnect
+
+
 	m.client = mqtt.NewClient(opts)
 	if token := m.client.Connect(); token.Wait() && token.Error() != nil {
 		return token.Error()
@@ -124,21 +127,24 @@ func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error {
 	m.in = make(chan mqtt.Message, 1000)
 	m.done = make(chan struct{})
 
-	topics := make(map[string]byte)
-	for _, topic := range m.Topics {
-		topics[topic] = byte(m.QoS)
-	}
-	subscribeToken := m.client.SubscribeMultiple(topics, m.recvMessage)
-	subscribeToken.Wait()
-	if subscribeToken.Error() != nil {
-		return subscribeToken.Error()
-	}
+
 
 	go m.receiver()
 
 	return nil
 }
-
+ func onConnect(c *MQTT.Client) {
+	 topics := make(map[string]byte)
+	 for _, topic := range m.Topics {
+		 topics[topic] = byte(m.QoS)
+	 }
+	 subscribeToken := c.SubscribeMultiple(topics, m.recvMessage)
+	 subscribeToken.Wait()
+	 if subscribeToken.Error() != nil {
+		 log.Printf("MQTT SUBSCRIBE ERROR\ntopics: %s\nerror: %s",
+			 string(m.Topics), err.Error())
+	 }
+ }
 // receiver() reads all incoming messages from the consumer, and parses them into
 // influxdb metric points.
 func (m *MQTTConsumer) receiver() {