feat: MessageClient can Pub/Sub binary data

MessageClient can Pub/Sub binary data without MessageEnvelope wrapped.

Signed-off-by: bruce <weichou1229@gmail.com>

internal/pkg/mqtt/client.go

@@ -342,3 +342,46 @@ 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 {
+	optionsReader := mc.mqttClient.OptionsReader()
+	for _, topic := range topics {
+		handler := newBinaryDataMessageHandler(topic.Messages)
+		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
+}
+
+// newBinaryDataMessageHandler creates a function which propagates the received messages to the proper channel.
+func newBinaryDataMessageHandler(messageChannel chan<- types.MessageEnvelope) 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
+	}
+}

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