func (communication *MQTT) publishCSSOnWIoTP(orgID string, destType string, destID string, dataJSON []byte, chunked bool) common.SyncServiceError {
	client, err := communication.getClient(orgID)
	if err != nil {
		return err
	}
	if client == nil {
		message := fmt.Sprintf("Failed to find client to publish (org id = %s, dest type = %s, dest id = %s)\n", orgID, destType, destID)
		if trace.IsLogging(logger.ERROR) {
			trace.Error(message)
		}
		if log.IsLogging(logger.ERROR) {
			log.Error(message)
		}
		return nil
	}
	var strBuilder strings.Builder
	strBuilder.Grow(len(orgID) + len(destType) + len(destID) + 40)
	strBuilder.WriteString("iotint-1/")
	strBuilder.WriteString(orgID)
	strBuilder.WriteString("/type/")
	strBuilder.WriteString(destType)
	strBuilder.WriteString("/id/")
	strBuilder.WriteString(destID)
	strBuilder.WriteString("/sync/sync-cmd")
	topic := strBuilder.String()
	return publish(client, topic, dataJSON)
}
Publish messages from the CSS on the WIoTP to the ESS
package communications
import (
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"strings"
	"sync"
	"time"
	"github.com/eclipse/paho.mqtt.golang"
	"github.com/eclipse/paho.mqtt.golang/packets"
	"github.com/open-horizon/edge-sync-service/common"
	"github.com/open-horizon/edge-sync-service/core/leader"
	"github.com/open-horizon/edge-utilities/logger"
	"github.com/open-horizon/edge-utilities/logger/log"
	"github.com/open-horizon/edge-utilities/logger/trace"
)
type messagePayload struct {
	Version            common.SyncServiceVersion `json:"version"`
	Command            string                    `json:"command"`
	Meta               common.MetaData           `json:"meta,omitempty"`
	Offset             int64                     `json:"offset,omitempty"`
	Destination        common.Destination        `json:"destination,omitempty"`
	PersistentStorage  bool                      `json:"persistent,omitempty"`
	FeedbackCode       int                       `json:"feedback-code,omitempty"`
	FeedbackFromOrigin bool                      `json:"feedback-from-origin,omitempty"`
	RetryInterval      int32                     `json:"retry,omitempty"`
	Reason             string                    `json:"reason,omitempty"`
}
type brokerAddresses struct {
	MessagingGroup string   `json:"messagingGroup"`
	ServerURIs     []string `json:"serverURIs"`
}
type clientInfo struct {
	name      string
	client    mqtt.Client
	clientID  string
	timestamp time.Time
}
type publishMessageFunc func(orgID string, destType string, destID string, dataJSON []byte, chunked bool) common.SyncServiceError
type MQTT struct {
	clients                 []clientInfo
	orgToClient             map[string]*clientInfo
	isLeader                bool
	topics                  map[string]byte
	topic                   string
	leaderTopic             string
	isCheckingDB            bool
	checkStopChannel        chan int
	checkForUpdatesTicker   *time.Ticker
	checkUpdatesStopChannel chan int
	parallelParams          parallelMQTTParams
	commandQ                []chan *messageHandlerInfo
	dataQ                   chan *messageHandlerInfo
	queueStopChannel        chan int
	lastTimestamp           time.Time
	publishMessage          publishMessageFunc
	serverURIs              [][]string
	lock                    sync.RWMutex
}
type mqttClientContext struct {
	name          string
	client        mqtt.Client
	clientOptions *mqtt.ClientOptions
	communicator  *MQTT
	subAttempts   uint32
}
type mqttContext struct {
	contexts     []mqttClientContext
	communicator *MQTT
}
var nodeContext mqttContext
type parallelMQTTParams struct {
	isParallelMQTTOn         bool
	numCommandMQTTGoRoutines int  
	numDataMQTTGoRoutines    int  
	commandMQTTQueueSize     int
	dataMQTTQueueSize        int
}
type messageHandlerInfo struct {
	context        *mqttClientContext
	messagePayload messagePayload
	payload        []byte
}
func (communication *MQTT) serveMQTTQueue(c chan *messageHandlerInfo) {
	common.GoRoutineStarted()
	var messageInfo *messageHandlerInfo
	keepRunning := true
	for keepRunning {
		select {
		case messageInfo = <-c:
			processMessage(messageInfo)
		case <-communication.queueStopChannel:
			keepRunning = false
		}
	}
	common.GoRoutineEnded()
}
func (context *mqttClientContext) messageHandler(client mqtt.Client, msg mqtt.Message) {
	var messageInfo messageHandlerInfo
	ok := parseMessage(msg, &messageInfo)
	if !ok {
		return
	}
	messageInfo.context = context
	processMessage(&messageInfo)
}
func (context *mqttClientContext) parallelMessageHandler(client mqtt.Client, msg mqtt.Message) {
	var messageInfo messageHandlerInfo
	ok := parseMessage(msg, &messageInfo)
	if !ok {
		return
	}
	messageInfo.context = context
	command := messageInfo.messagePayload.Command
	if command == common.Getdata || command == common.Data {
		context.communicator.dataQ <- &messageInfo
	} else if command == common.AckRegister {
		handleRegAck()
	} else if command == common.AckResend {
		handleAckResend()
	} else {
		var orgID, destType, destID string
		if command == common.Register || command == common.Resend {
			dest := messageInfo.messagePayload.Destination
			orgID = dest.DestOrgID
			destType = dest.DestType
			destID = dest.DestID
		} else {
			meta := messageInfo.messagePayload.Meta
			orgID = meta.DestOrgID
			destType = meta.OriginType
			destID = meta.OriginID
		}
		qIndex := common.HashStrings(orgID, destType, destID) % uint32(len(context.communicator.commandQ))
		context.communicator.commandQ[qIndex] <- &messageInfo
	}
}
func parseMessage(msg mqtt.Message, messageInfo *messageHandlerInfo) bool {
	payload := msg.Payload()
	if len(payload) >= 4 && payload[0] == 0x01 && payload[1] == 0x01 && payload[2] == 0x01 && payload[3] == 0x01 {
		messageInfo.messagePayload.Command = common.Data
		messageInfo.payload = payload
	} else {
		if err := json.Unmarshal(payload, &messageInfo.messagePayload); err != nil {
			err = &Error{"Failed to unmarshal payload. Error: %s" + err.Error()}
			if log.IsLogging(logger.ERROR) {
				log.Error(err.Error())
			}
			if !Store.IsConnected() {
				if log.IsLogging(logger.TRACE) {
					log.Trace("Lost connection to the database: unsubscribing")
				}
				nodeContext.unsubscribe()
			}
			return false
		}
		if messageInfo.messagePayload.Version != common.Version {
			if log.IsLogging(logger.ERROR) {
				log.Error("Received message with unsupported version")
			}
			return false
		}
	}
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Incoming message: \nTopic: %s|%s\nMessage length: %d\n", msg.Topic(), messageInfo.messagePayload.Command, len(payload))
	}
	return true
}
func processMessage(messageInfo *messageHandlerInfo) {
	context := messageInfo.context
	payload := messageInfo.payload
	messagePayload := messageInfo.messagePayload
	meta := &messagePayload.Meta
	if common.Configuration.NodeType == common.CSS && messagePayload.Command != common.Data &&
		messagePayload.Command != common.Register && messagePayload.Command != common.RegisterNew && messagePayload.Command != common.Ping {
		destType := meta.OriginType
		destID := meta.OriginID
		destOrgID := meta.DestOrgID
		if messagePayload.Command == common.Updated || messagePayload.Command == common.Consumed ||
			messagePayload.Command == common.Received || messagePayload.Command == common.AckDelete ||
			messagePayload.Command == common.Deleted || messagePayload.Command == common.Getdata ||
			(messagePayload.Command == common.Feedback && !messagePayload.FeedbackFromOrigin) {
			destType = meta.DestType
			destID = meta.DestID
		} else if messagePayload.Command == common.Resend {
			destType = messagePayload.Destination.DestType
			destID = messagePayload.Destination.DestID
			destOrgID = messagePayload.Destination.DestOrgID
		}
		if ok := destinationExists(destOrgID, destType, destID); !ok {
			return
		}
	}
	var err error
	switch messagePayload.Command {
	case common.Register:
		err = handleRegistration(messagePayload.Destination, messagePayload.PersistentStorage)
	case common.AckRegister:
		handleRegAck()
	case common.Ping:
		err = handlePing(messagePayload.Destination)
	case common.RegisterNew:
		if err = context.communicator.storeMessagingGroup(context.client, messagePayload.Destination.DestOrgID); err == nil {
			err = handleRegisterNew(messagePayload.Destination, messagePayload.PersistentStorage)
		}
	case common.RegisterAsNew:
		err = handleRegisterAsNew()
	case common.Update:
		if int64(meta.ChunkSize) < meta.ObjectSize && !leader.CheckIfLeader() {
			err = &Error{"Non-leader received update message with chunked data, ignoring."}
		} else {
			err = handleUpdate(*meta, common.Configuration.MaxInflightChunks)
			if err != nil && !isIgnoredByHandler(err) {
				context.communicator.SendErrorMessage(err, meta, true)
			}
		}
	case common.Updated:
		err = handleObjectUpdated(meta.DestOrgID, meta.ObjectType, meta.ObjectID, meta.DestType, meta.DestID, meta.InstanceID, meta.DataID)
	case common.Consumed:
		err = handleObjectConsumed(meta.DestOrgID, meta.ObjectType, meta.ObjectID, meta.DestType, meta.DestID, meta.InstanceID, meta.DataID)
	case common.AckConsumed:
		err = handleAckConsumed(meta.DestOrgID, meta.ObjectType, meta.ObjectID, meta.OriginType, meta.OriginID, meta.InstanceID, meta.DataID)
	case common.Received:
		err = handleObjectReceived(meta.DestOrgID, meta.ObjectType, meta.ObjectID, meta.DestType, meta.DestID, meta.InstanceID, meta.DataID)
	case common.AckReceived:
		err = handleAckObjectReceived(meta.DestOrgID, meta.ObjectType, meta.ObjectID, meta.OriginType, meta.OriginID, meta.InstanceID, meta.DataID)
	case common.Delete:
		err = handleDelete(messagePayload.Meta)
	case common.AckDelete:
		err = handleAckDelete(meta.DestOrgID, meta.ObjectType, meta.ObjectID, meta.DestType, meta.DestID, meta.InstanceID, meta.DataID)
	case common.Deleted:
		err = handleObjectDeleted(messagePayload.Meta)
	case common.AckDeleted:
		err = handleAckObjectDeleted(meta.DestOrgID, meta.ObjectType, meta.ObjectID, meta.OriginType, meta.OriginID, meta.InstanceID)
	case common.Getdata:
		err = handleGetData(messagePayload.Meta, messagePayload.Offset)
		if err != nil && (isIgnoredByHandler(err) || common.IsNotFound(err)) {
			context.communicator.SendErrorMessage(&common.NotFound{}, &messagePayload.Meta, false)
		}
	case common.Data:
		meta, err = handleData(payload)
		if meta != nil && err != nil && !isIgnoredByHandler(err) {
			context.communicator.SendErrorMessage(err, meta, true)
		}
	case common.Resend:
		err = handleResendRequest(messagePayload.Destination)
	case common.AckResend:
		err = handleAckResend()
	case common.Feedback:
		destType := meta.DestType
		destID := meta.DestID
		if messagePayload.FeedbackFromOrigin {
			destType = meta.OriginType
			destID = meta.OriginID
		}
		err = handleFeedback(meta.DestOrgID, meta.ObjectType, meta.ObjectID, destType, destID, meta.InstanceID, meta.DataID, messagePayload.FeedbackCode,
			messagePayload.RetryInterval, messagePayload.Reason)
	default:
		err = &Error{"Received message that doesn't match any subscription."}
	}
	if err != nil && !isIgnoredByHandler(err) {
		if log.IsLogging(logger.ERROR) {
			log.Error(err.Error())
		}
		if !Store.IsConnected() {
			if log.IsLogging(logger.TRACE) {
				log.Trace("Lost connection to the database: unsubscribing")
			}
			nodeContext.unsubscribe()
		}
	}
}
func (communication *MQTT) storeMessagingGroup(client mqtt.Client, orgID string) common.SyncServiceError {
	if !common.Configuration.CSSOnWIoTP {
		return nil
	}
	communication.lock.Lock()
	defer communication.lock.Unlock()
	for _, group := range communication.clients {
		if group.client == client {
			communication.orgToClient[orgID] = &group
			return Store.StoreOrgToMessagingGroup(orgID, group.name)
		}
	}
	return &Error{"Failed to find client."}
}
func (communication *MQTT) getClient(orgID string) (mqtt.Client, common.SyncServiceError) {
	communication.lock.RLock()
	clientInfo, ok := communication.orgToClient[orgID]
	if ok {
		communication.lock.RUnlock()
		return clientInfo.client, nil
	}
	if common.Configuration.CSSOnWIoTP {
		groupName, err := Store.RetrieveMessagingGroup(orgID)
		if err != nil {
			communication.lock.RUnlock()
			return nil, err
		}
		if groupName == "" {
			communication.lock.RUnlock()
			return nil, nil
		}
		for _, c := range communication.clients {
			if c.name == groupName {
				communication.lock.RUnlock()
				communication.lock.Lock()
				communication.orgToClient[orgID] = &c
				communication.lock.Unlock()
				return c.client, nil
			}
		}
		communication.lock.RUnlock()
		return nil, nil
	}
	defer communication.lock.RUnlock()
	org, err := Store.RetrieveOrganizationInfo(orgID)
	if err != nil {
		return nil, err
	}
	if org == nil {
		return nil, nil
	}
	if err := communication.UpdateOrganization(org.Org, org.Timestamp); err != nil {
		return nil, err
	}
	clientInfo, ok = communication.orgToClient[orgID]
	if !ok {
		return nil, &Error{"Failed to retrieve client for updated organization"}
	}
	return clientInfo.client, nil
}
func newTLSConfig() *tls.Config {
	tlsConfig := tls.Config{}
	if common.Configuration.MQTTCACertificate != "" {
		certpool := x509.NewCertPool()
		var caCert string
		if strings.HasPrefix(common.Configuration.MQTTCACertificate, "/") {
			caCert = common.Configuration.MQTTCACertificate
		} else {
			caCert = common.Configuration.PersistenceRootPath + common.Configuration.MQTTCACertificate
		}
		pemCerts, err := ioutil.ReadFile(caCert)
		if err != nil {
			if _, ok := err.(*os.PathError); ok {
				pemCerts = []byte(common.Configuration.MQTTCACertificate)
				err = nil
			} else {
				if log.IsLogging(logger.ERROR) {
					log.Error(err.Error())
				}
			}
		}
		if err == nil {
			certpool.AppendCertsFromPEM(pemCerts)
		}
		tlsConfig.RootCAs = certpool
	}
	if common.Configuration.MQTTSSLCert != "" && common.Configuration.MQTTSSLKey != "" {
		var cert, key string
		if strings.HasPrefix(common.Configuration.MQTTSSLCert, "/") {
			cert = common.Configuration.MQTTSSLCert
		} else {
			cert = common.Configuration.PersistenceRootPath + common.Configuration.MQTTSSLCert
		}
		if strings.HasPrefix(common.Configuration.MQTTSSLKey, "/") {
			key = common.Configuration.MQTTSSLKey
		} else {
			key = common.Configuration.PersistenceRootPath + common.Configuration.MQTTSSLKey
		}
		clientCert, err := tls.LoadX509KeyPair(cert, key)
		if err != nil {
			if _, ok := err.(*os.PathError); ok {
				clientCert, err = tls.X509KeyPair([]byte(common.Configuration.MQTTSSLCert), []byte(common.Configuration.MQTTSSLKey))
			}
			if err != nil {
				if log.IsLogging(logger.ERROR) {
					log.Error(err.Error())
				}
			}
		}
		if err == nil {
			tlsConfig.Certificates = []tls.Certificate{clientCert}
		}
	}
	if common.Configuration.MQTTAllowInvalidCertificates {
		tlsConfig.InsecureSkipVerify = true
	}
	return &tlsConfig
}
func (communication *MQTT) createClients() ([]clientInfo, common.SyncServiceError) {
	communication.topics = make(map[string]byte, 0)
	qos := byte(0)
	if common.Configuration.NodeType == common.ESS {
		if common.Configuration.CSSOnWIoTP {
			communication.topic = "iotint-1/" + common.Configuration.OrgID + "/type/" + common.Configuration.DestinationType +
				"/id/" + common.Configuration.DestinationID + "/sync/sync-cmd"
			if common.Configuration.UsingEdgeConnector {
				communication.publishMessage = communication.publishESSOnWIoTPEC
			} else {
				communication.publishMessage = communication.publishESSOnWIoTPNotEC
			}
		} else {
			communication.topic = "iot-2/type/" + common.Configuration.DestinationType + "/id/" +
				common.Configuration.DestinationID + "/cmd/sync-cmd/fmt/bin"
			if common.Configuration.UsingEdgeConnector {
				communication.publishMessage = communication.publishESSOutsideWIoTPEC
			} else {
				communication.publishMessage = communication.publishESSOutsideWIoTPNotEC
			}
		}
	} else {
		if common.Configuration.CSSOnWIoTP {
			communication.topic = "$SharedSubscription/sync-service/iotintdev-1/type/+/id/+/sync/sync-cmd"
			communication.leaderTopic = "$SharedSubscription/sync-service/iotintdev-1/type/+/id/+/sync/sync-cmd-leader"
			communication.publishMessage = communication.publishCSSOnWIoTP
		} else {
			communication.topic = "iot-2/type/+/id/+/evt/sync-cmd/fmt/bin"
			communication.leaderTopic = "iot-2/type/+/id/+/evt/sync-cmd-leader/fmt/bin"
			communication.publishMessage = communication.publishCSSOutsideWIoTP
		}
		if communication.isLeader {
			communication.topics[communication.leaderTopic] = qos
		}
	}
	communication.topics[communication.topic] = qos
	clients := make([]clientInfo, 0)
	usernames := make([]string, 0)
	passwords := make([]string, 0)
	var orgs []common.StoredOrganization
	if strings.HasPrefix(common.Configuration.BrokerAddress, "[") {
		var brokerAddresses []brokerAddresses
		err := json.Unmarshal([]byte(common.Configuration.BrokerAddress), &brokerAddresses)
		if err != nil {
			return nil, &Error{"Failed to unmarshal broker addresses. Error: %s" + err.Error()}
		}
		if len(brokerAddresses) < 1 {
			return nil, &Error{"One or more broker addresses must be provided."}
		}
		for i, group := range brokerAddresses {
			mg := clientInfo{name: group.MessagingGroup, clientID: common.Configuration.MQTTClientID}
			clients = append(clients, mg)
			usernames = append(usernames, common.Configuration.MQTTUserName)
			passwords = append(passwords, common.Configuration.MQTTPassword)
			communication.serverURIs = append(communication.serverURIs, make([]string, 0))
			for _, brokerURI := range group.ServerURIs {
				communication.serverURIs[i] = append(communication.serverURIs[i], brokerURI)
			}
		}
	} else {
		protocol := "ssl"
		if !common.Configuration.MQTTUseSSL {
			protocol = "tcp"
		}
		if !common.SingleOrgCSS && common.Configuration.NodeType == common.CSS {
			var err error
			orgs, err = Store.RetrieveOrganizations()
			if err != nil {
				return nil, err
			}
			for _, org := range orgs {
				brokerURI := org.Org.Address
				if brokerURI == "" {
					if common.Configuration.CommunicationProtocol == common.MQTTProtocol ||
						common.Configuration.CommunicationProtocol == common.HybridMQTT {
						message := fmt.Sprintf("Can't create MQTT client for organization %s: no broker address\n", org.Org.OrgID)
						if trace.IsLogging(logger.ERROR) {
							trace.Error(message)
						}
						if log.IsLogging(logger.ERROR) {
							log.Error(message)
						}
						continue
					}
					brokerURI = fmt.Sprintf("%s://%s.%s:%d", protocol, org.Org.OrgID, common.Configuration.BrokerAddress,
						common.Configuration.BrokerPort)
				}
				communication.serverURIs = append(communication.serverURIs, make([]string, 0))
				communication.serverURIs[len(communication.serverURIs)-1] = append(communication.serverURIs[len(communication.serverURIs)-1], brokerURI)
				clientInfo := clientInfo{name: org.Org.OrgID, clientID: "A:" + org.Org.OrgID + ":CSS", timestamp: org.Timestamp}
				clients = append(clients, clientInfo)
				usernames = append(usernames, org.Org.User)
				passwords = append(passwords, org.Org.Password)
				communication.orgToClient[org.Org.OrgID] = &clientInfo
			}
		} else if common.SingleOrgCSS || common.Configuration.NodeType == common.ESS {
			communication.serverURIs = append(communication.serverURIs, make([]string, 0))
			brokerURI := fmt.Sprintf("%s://%s:%d", protocol, common.Configuration.BrokerAddress, common.Configuration.BrokerPort)
			communication.serverURIs[len(communication.serverURIs)-1] = append(communication.serverURIs[len(communication.serverURIs)-1], brokerURI)
			name := "default"
			if common.SingleOrgCSS {
				name = common.Configuration.OrgID
			}
			clientInfo := clientInfo{name: name, clientID: common.Configuration.MQTTClientID}
			usernames = append(usernames, common.Configuration.MQTTUserName)
			passwords = append(passwords, common.Configuration.MQTTPassword)
			clients = append(clients, clientInfo)
			if common.SingleOrgCSS {
				communication.orgToClient[name] = &clientInfo
			}
		}
	}
	for i, groupURIs := range communication.serverURIs {
		context := mqttClientContext{name: clients[i].name, communicator: communication}
		c, err := context.createAndConnectClient(clients[i], usernames[i], passwords[i], groupURIs)
		if err != nil {
			return nil, err
		}
		clients[i].client = c
		clientInfo := communication.orgToClient[clients[i].name]
		if clientInfo != nil {
			clientInfo.client = c
			communication.orgToClient[clients[i].name] = clientInfo
		}
		context.client = c
		nodeContext.contexts = append(nodeContext.contexts, context)
	}
	return clients, nil
}
func (context *mqttClientContext) createAndConnectClient(clientInfo clientInfo, username string, password string, servers []string) (mqtt.Client, common.SyncServiceError) {
	opts := mqtt.NewClientOptions()
	opts.SetClientID(clientInfo.clientID)
	opts.SetKeepAlive(120 * time.Second)
	if context.communicator.parallelParams.isParallelMQTTOn {
		opts.SetDefaultPublishHandler(context.parallelMessageHandler)
	} else {
		opts.SetDefaultPublishHandler(context.messageHandler)
	}
	opts.SetPingTimeout(120 * time.Second)
	opts.SetAutoReconnect(true)
	opts.SetOnConnectHandler(context.onReconnect)
	opts.SetConnectionLostHandler(context.onConnectionLost)
	opts.SetMaxReconnectInterval(30 * time.Second)
	opts.Username = username
	opts.Password = password
	if common.Configuration.MQTTUseSSL {
		opts.SetTLSConfig(newTLSConfig())
	}
	for _, serverURI := range servers {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Adding the broker %s for %s\n", serverURI, clientInfo.name)
		}
		opts.AddBroker(serverURI)
	}
	client := mqtt.NewClient(opts)
	var tokenError error
	for connectTime := 0; connectTime < common.Configuration.MQTTBrokerConnectTimeout; connectTime += 10 {
		if token := client.Connect(); token.WaitTimeout(time.Duration(10*time.Second)) && token.Error() != nil {
			tokenError = token.Error()
			if strings.HasPrefix(tokenError.Error(), packets.ConnErrors[packets.ErrRefusedServerUnavailable].Error()) ||
				strings.HasPrefix(tokenError.Error(), packets.ConnErrors[packets.ErrNetworkError].Error()) {
				if trace.IsLogging(logger.TRACE) {
					trace.Trace("Connection to broker %s failed. Retrying to connect.\n", clientInfo.name)
				}
				time.Sleep(time.Duration(10 * time.Second))
			} else {
				break
			}
		} else {
			return client, nil
		}
	}
	return nil, tokenError
}
func (context *mqttClientContext) onConnectionLost(client mqtt.Client, err error) {
	if trace.IsLogging(logger.ERROR) {
		trace.Error("Lost connection to the MQTT broker %s. Error: %s\n", context.name, err.Error())
	}
	if log.IsLogging(logger.ERROR) {
		log.Error("Lost connection to the MQTT broker %s. Error: %s\n", context.name, err.Error())
	}
	common.HealthStatus.DisconnectedFromBroker()
}
func (communication *MQTT) checkIfOrgExists(client mqtt.Client) (exists bool, username string, password string) {
	if common.Configuration.NodeType == common.ESS || common.Configuration.CSSOnWIoTP || common.SingleOrgCSS {
		return true, common.Configuration.MQTTUserName, common.Configuration.MQTTPassword
	}
	communication.lock.RLock()
	defer communication.lock.RUnlock()
	for _, c := range communication.clients {
		if c.client == client {
			org := c.name
			orgInfo, err := Store.RetrieveOrganizationInfo(org)
			if orgInfo != nil && err == nil {
				return true, orgInfo.Org.User, orgInfo.Org.Password
			}
			communication.DeleteOrganization(org)
			return false, "", ""
		}
	}
	return false, "", ""
}
func (context *mqttClientContext) subscribe() {
	client := context.client
	if err := subscribe(client, context.communicator.topics); err != nil {
		exists, username, password := context.communicator.checkIfOrgExists(client)
		if !exists {
			return
		}
		common.HealthStatus.SubscribeFailed()
		if context.subAttempts > 0 && context.subAttempts <= 10 {
			if trace.IsLogging(logger.ERROR) {
				trace.Error(err.Error())
			}
			client.Disconnect(0)
			context.subAttempts++
			context.communicator.lock.RLock()
			found := false
			for i, c := range context.communicator.clients {
				if c.client == client {
					context.communicator.lock.RUnlock()
					found = true
					newClient, err := context.createAndConnectClient(c, username, password, context.communicator.serverURIs[i])
					if err != nil {
						if log.IsLogging(logger.FATAL) {
							log.Fatal(err.Error())
						}
						if trace.IsLogging(logger.FATAL) {
							trace.Fatal(err.Error())
						}
						os.Exit(99)
					}
					context.communicator.lock.Lock()
					context.communicator.clients[i].client = newClient
					context.client = newClient
					context.communicator.lock.Unlock()
					break
				}
			}
			if !found {
				context.communicator.lock.RUnlock()
			}
		} else {
			if log.IsLogging(logger.FATAL) {
				log.Fatal(err.Error())
			}
			if trace.IsLogging(logger.FATAL) {
				trace.Fatal(err.Error())
			}
			os.Exit(99)
		}
	}
	context.subAttempts = 1
}
func (context *mqttClientContext) onReconnect(client mqtt.Client) {
	if trace.IsLogging(logger.INFO) {
		trace.Info("Connected to the MQTT broker %s\n", context.name)
	}
	if log.IsLogging(logger.INFO) {
		log.Info("Connected to the MQTT broker %s\n", context.name)
	}
	common.HealthStatus.ReconnectedToBroker()
	if len(context.communicator.topics) > 0 {
		context.subscribe()
	}
	if common.Configuration.NodeType == common.ESS {
		context.communicator.Register()
	}
}
func publish(client mqtt.Client, topic string, payload []byte) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Publishing on topic: %s\n", topic)
	}
	if token := client.Publish(topic, 0, false, payload); token.WaitTimeout(time.Duration(10*time.Second)) && token.Error() != nil {
		common.HealthStatus.PublishFailed()
		message := fmt.Sprintf("Failed to publish on topic %s. Error: ", topic)
		return &Error{message + token.Error().Error()}
	}
	return nil
}
func subscribe(client mqtt.Client, topics map[string]byte) common.SyncServiceError {
	if topics == nil {
		return &Error{"Failed to subscribe: no topics provided"}
	}
	if token := client.SubscribeMultiple(topics, nil); token.WaitTimeout(time.Duration(10*time.Second)) && token.Error() != nil {
		message := fmt.Sprintf("Failed to subscribe. Error: %s\n", token.Error().Error())
		return &Error{message}
	}
	return nil
}
func (communication *MQTT) StartCommunication() common.SyncServiceError {
	nodeContext = mqttContext{make([]mqttClientContext, 0), communication}
	communication.orgToClient = make(map[string]*clientInfo)
	switch common.Configuration.MQTTParallelMode {
	case common.ParallelMQTTNone:
		communication.parallelParams.isParallelMQTTOn = false
	case common.ParallelMQTTSmall:
		communication.parallelParams = parallelMQTTParams{true, 4, 16, 4, 32}
	case common.ParallelMQTTMedium:
		communication.parallelParams = parallelMQTTParams{true, 8, 32, 8, 64}
	case common.ParallelMQTTLarge:
		communication.parallelParams = parallelMQTTParams{true, 16, 64, 16, 128}
	default:
		communication.parallelParams.isParallelMQTTOn = false
	}
	if communication.parallelParams.isParallelMQTTOn {
		communication.queueStopChannel = make(chan int,
			communication.parallelParams.numCommandMQTTGoRoutines+communication.parallelParams.numDataMQTTGoRoutines)
		communication.commandQ = make([]chan *messageHandlerInfo, communication.parallelParams.numCommandMQTTGoRoutines)
		for i := 0; i < communication.parallelParams.numCommandMQTTGoRoutines; i++ {
			communication.commandQ[i] = make(chan *messageHandlerInfo, communication.parallelParams.commandMQTTQueueSize)
			go communication.serveMQTTQueue(communication.commandQ[i])
		}
		communication.dataQ = make(chan *messageHandlerInfo, communication.parallelParams.dataMQTTQueueSize)
		for i := 0; i < communication.parallelParams.numDataMQTTGoRoutines; i++ {
			go communication.serveMQTTQueue(communication.dataQ)
		}
	}
	communication.isLeader = leader.CheckIfLeader()
	clients, err := communication.createClients()
	if err != nil {
		return &Error{"Failed to create an MQTT client. Error: " + err.Error()}
	}
	communication.clients = clients
	leader.SetChangeLeaderCallback(nodeContext.changeLeadership)
	leader.SetUnsubcribeCallback(nodeContext.unsubscribe)
	if common.Configuration.NodeType == common.CSS && !common.SingleOrgCSS {
		communication.lastTimestamp, err = Store.RetrieveTimeOnServer()
		if err != nil {
			message := fmt.Sprintf("Failed to retrieve time on server. Error: %s\n", err.Error())
			if trace.IsLogging(logger.ERROR) {
				trace.Error(message)
			}
			if log.IsLogging(logger.ERROR) {
				log.Error(message)
			}
		}
		communication.checkForUpdates()
	}
	return nil
}
func (communication *MQTT) StopCommunication() common.SyncServiceError {
	if communication.isCheckingDB {
		communication.checkStopChannel <- 1
	}
	if communication.checkForUpdatesTicker != nil {
		communication.checkForUpdatesTicker.Stop()
		communication.checkUpdatesStopChannel <- 1
	}
	for i := 0; i < communication.parallelParams.numCommandMQTTGoRoutines+communication.parallelParams.numDataMQTTGoRoutines; i++ {
		communication.queueStopChannel <- 1
	}
	for _, info := range communication.clients {
		info.client.Disconnect(0)
		if trace.IsLogging(logger.INFO) {
			trace.Info("Disconnecting from the MQTT broker %s\n", info.name)
		}
		if log.IsLogging(logger.INFO) {
			log.Info("Disconnecting from the MQTT broker %s\n", info.name)
		}
	}
	return nil
}
func (communication *MQTT) publishESSOnWIoTPEC(orgID string, destType string, destID string, dataJSON []byte, chunked bool) common.SyncServiceError {
	client := communication.clients[0].client
	topicType := "sync-cmd"
	if chunked {
		topicType = "sync-cmd-leader"
	}
	var strBuilder strings.Builder
	strBuilder.Grow(len(common.Configuration.DestinationType) + len(common.Configuration.DestinationID) + 90)
	strBuilder.WriteString("$IBUS-1/src/SYNC/dst/CC_DNS/type/app/topic/iotintdev-1/type/")
	strBuilder.WriteString(common.Configuration.DestinationType)
	strBuilder.WriteString("/id/")
	strBuilder.WriteString(common.Configuration.DestinationID)
	strBuilder.WriteString("/sync/")
	strBuilder.WriteString(topicType)
	topic := strBuilder.String()
	return publish(client, topic, dataJSON)
}
func (communication *MQTT) publishESSOnWIoTPNotEC(orgID string, destType string, destID string, dataJSON []byte, chunked bool) common.SyncServiceError {
	client := communication.clients[0].client
	topicType := "sync-cmd"
	if chunked {
		topicType = "sync-cmd-leader"
	}
	var strBuilder strings.Builder
	strBuilder.Grow(len(common.Configuration.DestinationType) + len(common.Configuration.DestinationID) + 30)
	strBuilder.WriteString("iotintdev-1/type/")
	strBuilder.WriteString(common.Configuration.DestinationType)
	strBuilder.WriteString("/id/")
	strBuilder.WriteString(common.Configuration.DestinationID)
	strBuilder.WriteString("/sync/")
	strBuilder.WriteString(topicType)
	topic := strBuilder.String()
	return publish(client, topic, dataJSON)
}
func (communication *MQTT) publishESSOutsideWIoTPEC(orgID string, destType string, destID string, dataJSON []byte, chunked bool) common.SyncServiceError {
	client := communication.clients[0].client
	topicType := "sync-cmd"
	if chunked {
		topicType = "sync-cmd-leader"
	}
	var strBuilder strings.Builder
	strBuilder.Grow(len(common.Configuration.DestinationType) + len(common.Configuration.DestinationID) + 90)
	strBuilder.WriteString("$IBUS-1/src/SYNC/dst/CC_DNS/type/app/topic/iot-2/type/")
	strBuilder.WriteString(common.Configuration.DestinationType)
	strBuilder.WriteString("/id/")
	strBuilder.WriteString(common.Configuration.DestinationID)
	strBuilder.WriteString("/evt/")
	strBuilder.WriteString(topicType)
	strBuilder.WriteString("/fmt/bin")
	topic := strBuilder.String()
	return publish(client, topic, dataJSON)
}
func (communication *MQTT) publishESSOutsideWIoTPNotEC(orgID string, destType string, destID string, dataJSON []byte, chunked bool) common.SyncServiceError {
	client := communication.clients[0].client
	topicType := "sync-cmd"
	if chunked {
		topicType = "sync-cmd-leader"
	}
	var strBuilder strings.Builder
	strBuilder.Grow(len(common.Configuration.DestinationType) + len(common.Configuration.DestinationID) + 30)
	strBuilder.WriteString("iot-2/type/")
	strBuilder.WriteString(common.Configuration.DestinationType)
	strBuilder.WriteString("/id/")
	strBuilder.WriteString(common.Configuration.DestinationID)
	strBuilder.WriteString("/evt/")
	strBuilder.WriteString(topicType)
	strBuilder.WriteString("/fmt/bin")
	topic := strBuilder.String()
	return publish(client, topic, dataJSON)
}