feat: MessageClient can Pub/Sub binary data

MessageClient can Pub/Sub binary data without MessageEnvelope wrapped.

Signed-off-by: bruce <[email protected]>

internal/pkg/mqtt/client.go

@@ -39,6 +39,9 @@ type MessageMarshaller func(v interface{}) ([]byte, error)
 // MessageUnmarshaller defines the function signature for unmarshaling []byte into structs.
 type MessageUnmarshaller func(data []byte, v interface{}) error
 
+type MessageHandlerCreator func(unmarshaler MessageUnmarshaller,
+	messageChannel chan<- types.MessageEnvelope, errorChannel chan<- error) pahoMqtt.MessageHandler
+
 // Client facilitates communication to an MQTT server and provides functionality needed to send and receive MQTT
 // messages.
 type Client struct {
@@ -159,30 +162,7 @@ func (mc *Client) Publish(message types.MessageEnvelope, topic string) error {
 
 // Subscribe creates a subscription for the specified topics.
 func (mc *Client) Subscribe(topics []types.TopicChannel, messageErrors chan error) error {
-	optionsReader := mc.mqttClient.OptionsReader()
-
-	mc.subscriptionMutex.Lock()
-	defer mc.subscriptionMutex.Unlock()
-
-	for _, topic := range topics {
-		handler := newMessageHandler(mc.unmarshaller, topic.Messages, messageErrors)
-		qos := optionsReader.WillQos()
-
-		token := mc.mqttClient.Subscribe(topic.Topic, qos, handler)
-		err := getTokenError(token, optionsReader.ConnectTimeout(), SubscribeOperation, "Failed to create subscription")
-		if err != nil {
-			return err
-		}
-
-		mc.existingSubscriptions[topic.Topic] = existingSubscription{
-			topic:   topic.Topic,
-			qos:     qos,
-			handler: handler,
-			errors:  messageErrors,
-		}
-	}
-
-	return nil
+	return mc.subscribe(topics, messageErrors, newMessageHandler)
 }
 
 // Request publishes a request and waits for a response
@@ -342,3 +322,59 @@ func createClientOptions(
 
 	return clientOptions, nil
 }
+
+func (mc *Client) PublishBinaryData(data []byte, topic string) error {
+	optionsReader := mc.mqttClient.OptionsReader()
+	return getTokenError(
+		mc.mqttClient.Publish(
+			topic,
+			optionsReader.WillQos(),
+			optionsReader.WillRetained(),
+			data),
+		optionsReader.ConnectTimeout(),
+		PublishOperation,
+		"Unable to publish message")
+}
+
+func (mc *Client) SubscribeBinaryData(topics []types.TopicChannel, messageErrors chan error) error {
+	return mc.subscribe(topics, messageErrors, newBinaryDataMessageHandler)
+}
+
+// newBinaryDataMessageHandler creates a function which propagates the received messages to the proper channel.
+func newBinaryDataMessageHandler(_ MessageUnmarshaller,
+	messageChannel chan<- types.MessageEnvelope,
+	_ chan<- error) pahoMqtt.MessageHandler {
+	return func(client pahoMqtt.Client, message pahoMqtt.Message) {
+		// Use MessageEnvelope.Payload to store the binary data instead of unmarshalling binary to MessageEnvelope
+		messageEnvelope := types.NewMessageEnvelopeForRequest(message.Payload(), nil)
+		messageEnvelope.ReceivedTopic = message.Topic()
+		messageChannel <- messageEnvelope
+	}
+}
+
+func (mc *Client) subscribe(topics []types.TopicChannel, messageErrors chan error, messageHandlerCreator MessageHandlerCreator) error {
+	optionsReader := mc.mqttClient.OptionsReader()
+
+	mc.subscriptionMutex.Lock()
+	defer mc.subscriptionMutex.Unlock()
+
+	for _, topic := range topics {
+		handler := messageHandlerCreator(mc.unmarshaller, topic.Messages, messageErrors)
+		qos := optionsReader.WillQos()
+
+		token := mc.mqttClient.Subscribe(topic.Topic, qos, handler)
+		err := getTokenError(token, optionsReader.ConnectTimeout(), SubscribeOperation, "Failed to create subscription")
+		if err != nil {
+			return err
+		}
+
+		mc.existingSubscriptions[topic.Topic] = existingSubscription{
+			topic:   topic.Topic,
+			qos:     qos,
+			handler: handler,
+			errors:  messageErrors,
+		}
+	}
+
+	return nil
+}

internal/pkg/nats/client.go

@@ -223,3 +223,11 @@ func (c *Client) Disconnect() error {
 	}
 	return c.connection.Drain()
 }
+
+func (c *Client) PublishBinaryData(data []byte, topic string) error {
+	return fmt.Errorf("not supported PublishBinaryData func")
+}
+
+func (c *Client) SubscribeBinaryData(topics []types.TopicChannel, messageErrors chan error) error {
+	return fmt.Errorf("not supported SubscribeBinaryData func")
+}

internal/pkg/noopclient.go

@@ -18,6 +18,7 @@
 package pkg
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/edgexfoundry/go-mod-messaging/v4/pkg/types"
@@ -48,3 +49,10 @@ func (n NoopClient) Subscribe(topics []types.TopicChannel, messageErrors chan er
 func (n NoopClient) Disconnect() error {
 	panic("implement me")
 }
+
+func (n NoopClient) PublishBinaryData(data []byte, topic string) error {
+	return fmt.Errorf("not supported PublishBinaryData func")
+}
+func (n NoopClient) SubscribeBinaryData(topics []types.TopicChannel, messageErrors chan error) error {
+	return fmt.Errorf("not supported SubscribeBinaryData func")
+}

internal/pkg/redis/client.go

@@ -311,3 +311,10 @@ func convertFromRedisTopicScheme(topic string) string {
 
 	return topic
 }
+
+func (c Client) PublishBinaryData(data []byte, topic string) error {
+	return fmt.Errorf("not supported PublishBinaryData func")
+}
+func (c Client) SubscribeBinaryData(topics []types.TopicChannel, messageErrors chan error) error {
+	return fmt.Errorf("not supported SubscribeBinaryData func")
+}

messaging/interface.go

@@ -47,6 +47,12 @@ type MessageClient interface {
 	// the timeout period, a timed out  error returned.
 	Request(message types.MessageEnvelope, requestTopic string, responseTopicPrefix string, timeout time.Duration) (*types.MessageEnvelope, error)
 
+	// PublishBinaryData sends binary data to the message bus
+	PublishBinaryData(data []byte, topic string) error
+
+	// SubscribeBinaryData receives binary data from the specified topic, and wrap it in MessageEnvelope.
+	SubscribeBinaryData(topics []types.TopicChannel, messageErrors chan error) error
+
 	// Unsubscribe to unsubscribe from the specified topics.
 	Unsubscribe(topics ...string) error