func newMessageHandler(
	unmarshaler MessageUnmarshaller,
	messageChannel chan<- types.MessageEnvelope,
	errorChannel chan<- error) pahoMqtt.MessageHandler {
	return func(client pahoMqtt.Client, message pahoMqtt.Message) {
		var messageEnvelope types.MessageEnvelope
		payload := message.Payload()
		err := unmarshaler(payload, &messageEnvelope)
		if err != nil {
			errorChannel <- err
		}
		messageEnvelope.ReceivedTopic = message.Topic()
		messageChannel <- messageEnvelope
	}
}
newMessageHandler creates a function which meets the criteria for a MessageHandler and propagates the received
messages to the proper channel.
package mqtt
import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"time"
	"github.com/edgexfoundry/go-mod-messaging/v2/internal/pkg"
	"github.com/edgexfoundry/go-mod-messaging/v2/pkg/types"
	pahoMqtt "github.com/eclipse/paho.mqtt.golang"
)
type ClientCreator func(config types.MessageBusConfig, handler pahoMqtt.OnConnectHandler) (pahoMqtt.Client, error)
type MessageMarshaller func(v interface{}) ([]byte, error)
type MessageUnmarshaller func(data []byte, v interface{}) error
type Client struct {
	creator             ClientCreator
	configuration       types.MessageBusConfig
	mqttClient          pahoMqtt.Client
	marshaller          MessageMarshaller
	unmarshaller        MessageUnmarshaller
	activeSubscriptions []activeSubscription
}
type activeSubscription struct {
	topic   string
	qos     byte
	handler pahoMqtt.MessageHandler
	errors  chan error
}
func NewMQTTClient(config types.MessageBusConfig) (*Client, error) {
	client := &Client{
		creator:       DefaultClientCreator(),
		configuration: config,
		marshaller:    json.Marshal,
		unmarshaller:  json.Unmarshal,
	}
	return client, nil
}
func NewMQTTClientWithCreator(
	config types.MessageBusConfig,
	marshaller MessageMarshaller,
	unmarshaller MessageUnmarshaller,
	creator ClientCreator) (*Client, error) {
	client := &Client{
		creator:       creator,
		configuration: config,
		marshaller:    marshaller,
		unmarshaller:  unmarshaller,
	}
	return client, nil
}
func (mc *Client) Connect() error {
	if mc.mqttClient == nil {
		mqttClient, err := mc.creator(mc.configuration, mc.onConnectHandler)
		if err != nil {
			return err
		}
		mc.mqttClient = mqttClient
	}
	if mc.mqttClient.IsConnected() {
		return nil
	}
	optionsReader := mc.mqttClient.OptionsReader()
	return getTokenError(
		mc.mqttClient.Connect(),
		optionsReader.ConnectTimeout(),
		ConnectOperation,
		"Unable to connect")
}
func (mc *Client) onConnectHandler(_ pahoMqtt.Client) {
	optionsReader := mc.mqttClient.OptionsReader()
	for _, subscription := range mc.activeSubscriptions {
		token := mc.mqttClient.Subscribe(subscription.topic, subscription.qos, subscription.handler)
		message := fmt.Sprintf("Failed to re-create subscription for topic=%s", subscription.topic)
		err := getTokenError(token, optionsReader.ConnectTimeout(), SubscribeOperation, message)
		if err != nil {
			subscription.errors <- err
		}
	}
}
func (mc *Client) Publish(message types.MessageEnvelope, topic string) error {
	marshaledMessage, err := mc.marshaller(message)
	if err != nil {
		return NewOperationErr(PublishOperation, err.Error())
	}
	optionsReader := mc.mqttClient.OptionsReader()
	return getTokenError(
		mc.mqttClient.Publish(
			topic,
			optionsReader.WillQos(),
			optionsReader.WillRetained(),
			marshaledMessage),
		optionsReader.ConnectTimeout(),
		PublishOperation,
		"Unable to publish message")
}
func (mc *Client) Subscribe(topics []types.TopicChannel, messageErrors chan error) error {
	optionsReader := mc.mqttClient.OptionsReader()
	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.activeSubscriptions = append(mc.activeSubscriptions, activeSubscription{
			topic:   topic.Topic,
			qos:     qos,
			handler: handler,
			errors:  messageErrors,
		})
	}
	return nil
}
func (mc *Client) Disconnect() error {
	optionsReader := mc.mqttClient.OptionsReader()
	mc.mqttClient.Disconnect(uint(optionsReader.ConnectTimeout() * time.Millisecond))
	return nil
}
func DefaultClientCreator() ClientCreator {
	return func(config types.MessageBusConfig, handler pahoMqtt.OnConnectHandler) (pahoMqtt.Client, error) {
		clientConfiguration, err := CreateMQTTClientConfiguration(config)
		if err != nil {
			return nil, err
		}
		clientOptions, err := createClientOptions(clientConfiguration, tls.X509KeyPair, tls.LoadX509KeyPair)
		if err != nil {
			return nil, err
		}
		clientOptions.OnConnect = handler
		return pahoMqtt.NewClient(clientOptions), nil
	}
}
func ClientCreatorWithCertLoader(certCreator pkg.X509KeyPairCreator, certLoader pkg.X509KeyLoader) ClientCreator {
	return func(options types.MessageBusConfig, handler pahoMqtt.OnConnectHandler) (pahoMqtt.Client, error) {
		clientConfiguration, err := CreateMQTTClientConfiguration(options)
		if err != nil {
			return nil, err
		}
		clientOptions, err := createClientOptions(clientConfiguration, certCreator, certLoader)
		if err != nil {
			return nil, err
		}
		clientOptions.OnConnect = handler
		return pahoMqtt.NewClient(clientOptions), nil
	}
}