Add message forwarded check to handler

Signed-off-by: Darko Draskovic <darko.draskovic@gmail.com>

pkg/messaging/mqtt/pubsub_test.go

@@ -15,13 +15,13 @@ import (
 
 const (
 	topic    = "topic"
-	data     = "payload"
 	channel  = "9b7b1b3f-b1b0-46a8-a717-b8213f9eda3b"
 	subtopic = "engine"
 )
 
 var (
 	msgChan = make(chan messaging.Message)
+	data    = []byte("payload")
 )
 
 func TestPubsub(t *testing.T) {
@@ -39,22 +39,22 @@ func TestPubsub(t *testing.T) {
 			payload: nil,
 		},
 		{
-			desc:    "publish message with string payload",
-			payload: []byte(data),
+			desc:    "publish message with non-nil payload",
+			payload: data,
 		},
 		{
 			desc:    "publish message with channel",
-			payload: []byte(data),
+			payload: data,
 			channel: channel,
 		},
 		{
 			desc:     "publish message with subtopic",
-			payload:  []byte(data),
+			payload:  data,
 			subtopic: subtopic,
 		},
 		{
 			desc:     "publish message with channel and subtopic",
-			payload:  []byte(data),
+			payload:  data,
 			channel:  channel,
 			subtopic: subtopic,
 		},
@@ -69,21 +69,37 @@ func TestPubsub(t *testing.T) {
 		require.Nil(t, err, fmt.Sprintf("got unexpected error: %s", err))
 
 		receivedMsg := <-msgChan
-		// _ = receivedMsg
 		assert.Equal(t, expectedMsg, receivedMsg, fmt.Sprintf("%s: expected %+v got %+v\n", tc.desc, expectedMsg, receivedMsg))
 	}
 }
 
+var createds []int64
+var created int64
+
+func contains(s []int64, e int64) bool {
+	for _, a := range s {
+		if a == e {
+			return true
+		}
+	}
+	return false
+}
+
 func handler(msg messaging.Message) error {
-	msgChan <- msg
+	if !contains(createds, msg.Created) {
+		createds = append(createds, created)
+		msgChan <- msg
+	}
 	return nil
 }
 
 func message(channel, subtopic string, payload []byte) messaging.Message {
+	created++
 	return messaging.Message{
 		Channel:  channel,
 		Subtopic: subtopic,
 		Payload:  payload,
+		Created:  created,
 	}
 }
 func payload(m messaging.Message) ([]byte, error) {

pkg/messaging/mqtt/setup_test.go

@@ -21,7 +21,7 @@ import (
 const (
 	protocol = "mqtt"
 	id       = "mqtt-publisher"
-	timeout  = time.Second * 3
+	timeout  = time.Second * 5
 	qos      = 1
 )
 

pkg/messaging/nats/pubsub_test.go

@@ -15,13 +15,13 @@ import (
 const (
 	topic       = "topic"
 	chansPrefix = "channels"
-	data        = "payload"
 	channel     = "9b7b1b3f-b1b0-46a8-a717-b8213f9eda3b"
 	subtopic    = "engine"
 )
 
 var (
 	msgChan = make(chan messaging.Message)
+	data    = []byte("payload")
 )
 
 func TestPubsub(t *testing.T) {
@@ -42,21 +42,21 @@ func TestPubsub(t *testing.T) {
 		},
 		{
 			desc:    "publish message with string payload",
-			payload: []byte(data),
+			payload: data,
 		},
 		{
 			desc:    "publish message with channel",
-			payload: []byte(data),
+			payload: data,
 			channel: channel,
 		},
 		{
 			desc:     "publish message with subtopic",
-			payload:  []byte(data),
+			payload:  data,
 			subtopic: subtopic,
 		},
 		{
 			desc:     "publish message with channel and subtopic",
-			payload:  []byte(data),
+			payload:  data,
 			channel:  channel,
 			subtopic: subtopic,
 		},