fix issue with multiple topics using SendMessage

Author: toddtreece

pkg/mqtt/client.go

@@ -41,12 +41,12 @@ func NewClient(o Options) (*Client, error) {
 		opts.SetPassword(o.Password)
 	}
 
-	opts.SetPingTimeout(10 * time.Second)
-	opts.SetKeepAlive(10 * time.Second)
+	opts.SetPingTimeout(60 * time.Second)
+	opts.SetKeepAlive(60 * time.Second)
 	opts.SetAutoReconnect(true)
 	opts.SetMaxReconnectInterval(10 * time.Second)
 	opts.SetConnectionLostHandler(func(c paho.Client, err error) {
-		log.DefaultLogger.Error(fmt.Sprintf("MQTT Connection Lost: %s\n" + err.Error()))
+		log.DefaultLogger.Error(fmt.Sprintf("MQTT Connection Lost: %s", err.Error()))
 	})
 	opts.SetReconnectingHandler(func(c paho.Client, options *paho.ClientOptions) {
 		log.DefaultLogger.Debug("MQTT Reconnecting")
@@ -61,7 +61,7 @@ func NewClient(o Options) (*Client, error) {
 
 	return &Client{
 		client: &client,
-		stream: make(chan StreamMessage),
+		stream: make(chan StreamMessage, 1000),
 	}, nil
 }
 
@@ -83,12 +83,12 @@ func (c *Client) Messages(path string) ([]Message, bool) {
 	return topic.messages, true
 }
 
-func (c *Client) Stream() *chan StreamMessage {
-	return &c.stream
+func (c *Client) Stream() chan StreamMessage {
+	return c.stream
 }
 
 func (c *Client) HandleMessage(client paho.Client, msg paho.Message) {
-	log.DefaultLogger.Debug(fmt.Sprintf("Received MQTT Message %+v", msg))
+	log.DefaultLogger.Debug(fmt.Sprintf("Received MQTT Message for topic %s", msg.Topic()))
 	topic, ok := c.topics.Load(msg.Topic())
 
 	if ok {
@@ -129,6 +129,13 @@ func (c *Client) Subscribe(t string) {
 	}
 }
 
+func (c *Client) Unsubscribe(t string) {
+	log.DefaultLogger.Debug(fmt.Sprintf("Unsubscribing from MQTT topic: %s", t))
+	client := *c.client
+	client.Unsubscribe(t)
+	c.topics.Delete(t)
+}
+
 func (c *Client) Dispose() {
 	client := *c.client
 	log.DefaultLogger.Info("MQTT Disconnecting")

pkg/mqtt/topic.go

@@ -32,3 +32,7 @@ func (tm *TopicMap) Load(path string) (*Topic, bool) {
 func (tm *TopicMap) Store(topic *Topic) {
 	tm.Map.Store(topic.path, topic)
 }
+
+func (tm *TopicMap) Delete(path string) {
+	tm.Map.Delete(path)
+}

pkg/plugin/datasource.go

@@ -7,16 +7,18 @@ import (
 	"time"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana-plugin-sdk-go/backend/log"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 	"github.com/toddtreece/mqtt-datasource/pkg/mqtt"
 )
 
 type MQTTClient interface {
-	Stream() *chan mqtt.StreamMessage
+	Stream() chan mqtt.StreamMessage
 	IsConnected() bool
 	IsSubscribed(topic string) bool
 	Messages(topic string) ([]mqtt.Message, bool)
 	Subscribe(topic string)
+	Unsubscribe(topic string)
 }
 
 type MQTTDatasource struct {
@@ -97,16 +99,19 @@ func (m *MQTTDatasource) Query(query backend.DataQuery) backend.DataResponse {
 	}
 
 	frame := ToFrame(messages)
-	frame.SetMeta(&data.FrameMeta{
-		Channel: m.channelPrefix + qm.Topic,
-	})
+
+	if qm.Topic != "" {
+		frame.SetMeta(&data.FrameMeta{
+			Channel: m.channelPrefix + qm.Topic,
+		})
+	}
 
 	response.Frames = append(response.Frames, frame)
 	return response
 }
 
 func (m *MQTTDatasource) SendMessage(msg mqtt.StreamMessage, req *backend.RunStreamRequest, sender backend.StreamPacketSender) error {
-	if msg.Topic != req.Path {
+	if !m.Client.IsSubscribed(req.Path) {
 		return nil
 	}
 
@@ -125,6 +130,7 @@ func (m *MQTTDatasource) SendMessage(msg mqtt.StreamMessage, req *backend.RunStr
 		Data: bytes,
 	}
 
+	log.DefaultLogger.Debug(fmt.Sprintf("Sending message to client for topic %s", msg.Topic))
 	return sender.Send(packet)
 }
 

pkg/plugin/handler.go

@@ -3,12 +3,10 @@ package plugin
 import (
 	"context"
 	"errors"
-	"fmt"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/backend/datasource"
 	"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
-	"github.com/grafana/grafana-plugin-sdk-go/backend/log"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 	"github.com/toddtreece/mqtt-datasource/pkg/mqtt"
 )
@@ -59,7 +57,7 @@ func (h *Handler) CheckHealth(ctx context.Context, req *backend.CheckHealthReque
 		}, nil
 	}
 
-	if ds.Client.IsConnected() {
+	if !ds.Client.IsConnected() {
 		return &backend.CheckHealthResult{
 			Status:  backend.HealthStatusError,
 			Message: "MQTT Disconnected",
@@ -97,17 +95,15 @@ func (h *Handler) RunStream(ctx context.Context, req *backend.RunStreamRequest,
 		return err
 	}
 
+	defer ds.Client.Unsubscribe(req.Path)
+
 	for {
 		select {
 		case <-ctx.Done():
 			backend.Logger.Info("stop streaming (context canceled)")
 			return nil
-		case message := <-*ds.Client.Stream():
-			err := ds.SendMessage(message, req, sender)
-			if err != nil {
-				log.DefaultLogger.Error(fmt.Sprintf("unable to send message: %s", err.Error()))
-			}
-
+		case message := <-ds.Client.Stream():
+			go ds.SendMessage(message, req, sender)
 		}
 	}
 }

test_broker.js

@@ -2,7 +2,6 @@ const aedes = require('aedes')();
 const server = require('net').createServer(aedes.handle);
 
 const PORT = 1883;
-
 const publishers = {};
 
 const toMillis = {
@@ -41,4 +40,7 @@ const createPublisher = ({ topic, qos }) => {
 server.listen(PORT, () => {
   console.log('server started and listening on port ', PORT);
   aedes.on('subscribe', subscriptions => subscriptions.forEach(createPublisher));
+  aedes.on('connectionError', console.error);
+  aedes.on('clientDisconnect', client => console.log(`disconnect: ${client.id}`));
+  aedes.on('client', client => console.log(`connect: ${client.id}`));
 });