func (communication *HTTP) SendNotificationMessage(notificationTopic string, destType string, destID string, instanceID int64, dataID int64,
	metaData *common.MetaData) common.SyncServiceError {
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In SendNotificationMessage for %s. notificationTopic: %s destType: %s destID: %s\n", common.Configuration.NodeType, notificationTopic, destType, destID)
	}
	if common.Configuration.NodeType == common.CSS {
		var status string
		switch notificationTopic {
		case common.Update:
			status = common.UpdatePending
		case common.Delete:
			status = common.DeletePending
		case common.Deleted:
			status = common.DeletedPending
		case common.Consumed:
			status = common.ConsumedPending
		case common.Received:
			status = common.ReceivedPending
		default:
			return nil
		}
		lockIndex := common.HashStrings(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
		common.ObjectLocks.Lock(lockIndex)
		defer common.ObjectLocks.Unlock(lockIndex)
		notification := common.Notification{ObjectID: metaData.ObjectID, ObjectType: metaData.ObjectType,
			DestOrgID: metaData.DestOrgID, DestID: destID, DestType: destType, Status: status, InstanceID: instanceID, DataID: dataID}
		return Store.UpdateNotificationRecord(notification)
	}
	url := buildObjectURL(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, instanceID, dataID, notificationTopic)
	var request *http.Request
	var response *http.Response
	var err error
	for i := 0; i < common.Configuration.ESSSPIMaxRetry; i++ {
		if notificationTopic == common.Update || notificationTopic == common.Delete || notificationTopic == common.Deleted {
			if metaData == nil {
				return &Error{"No meta data"}
			}
			body, err := json.MarshalIndent(metaData, "", "  ")
			if err != nil {
				return &Error{"Failed to marshal payload. Error: " + err.Error()}
			}
			request, err = http.NewRequest("PUT", url, bytes.NewReader(body))
			if err != nil {
				return &Error{"Failed to create HTTP request. Error: " + err.Error()}
			}
			request.ContentLength = int64(len(body))
		} else {
			request, err = http.NewRequest("PUT", url, nil)
			if err != nil {
				return &Error{"Failed to create HTTP request. Error: " + err.Error()}
			}
		}
		security.AddIdentityToSPIRequest(request, url)
		request.Close = true
		response, err = communication.requestWrapper.do(request)
		if response != nil && response.Body != nil {
			defer response.Body.Close()
		}
		if IsTransportError(response, err) {
			respCode := 0
			errMsg := ""
			if response != nil {
				respCode = response.StatusCode
			}
			if err != nil {
				errMsg = err.Error()
			}
			if trace.IsLogging(logger.DEBUG) {
				trace.Debug("In SendNotificationMessage, receive transport error %s from topic: (%d) %s, response code is %d, maxRetry: %d, retry...", errMsg, i, notificationTopic, respCode, common.Configuration.ESSSPIMaxRetry)
			}
			time.Sleep(time.Duration(common.Configuration.ESSCallSPIRetryInterval) * time.Second)
			continue
		} else if err != nil {
			return &Error{"Failed to send HTTP request. Error: " + err.Error()}
		} else if response == nil {
			return &Error{"Received nil response from HTTP request. Error: " + err.Error()}
		} else {
			if response.StatusCode == http.StatusNoContent {
				switch notificationTopic {
				case common.Update:
					if metaData.Link == "" && !metaData.NoData && !metaData.MetaOnly {
						if err = communication.pushData(metaData); err != nil {
							return err
						}
					}
					if err = handleObjectUpdated(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID,
						destType, destID, instanceID, dataID); err != nil {
						return err
					}
				case common.Delete:
					return handleAckDelete(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID,
						destType, destID, instanceID, dataID)
				case common.Deleted:
					return handleAckObjectDeleted(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID,
						destType, destID, instanceID)
				case common.Consumed:
					return handleAckConsumed(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, destType, destID, instanceID, dataID)
				case common.Received:
					return handleAckObjectReceived(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, destType, destID, instanceID, dataID)
				}
				return nil
			} else if response.StatusCode == http.StatusConflict {
				if trace.IsLogging(logger.TRACE) {
					trace.Trace("A notification of type %s was ignored by the other side (object %s:%s, instance id = %d)\n", notificationTopic,
						metaData.ObjectType, metaData.ObjectID, instanceID)
				}
				switch notificationTopic {
				case common.Deleted:
					return handleAckObjectDeleted(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, destType, destID, instanceID)
				case common.Consumed:
					return handleAckConsumed(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, destType, destID, instanceID, dataID)
				case common.Received:
					return handleAckObjectReceived(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, destType, destID, instanceID, dataID)
				}
				return nil
			}
		}
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In SendNotificationMessage, out of retry for %s %s %s", metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
	}
	return communication.createError(response, "send notification "+notificationTopic)
}
SendNotificationMessage sends a notification message from the CSS to the ESS or from the ESS to the CSS
package communications
import (
	"bytes"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
	"github.com/open-horizon/edge-sync-service/common"
	"github.com/open-horizon/edge-sync-service/core/dataURI"
	"github.com/open-horizon/edge-sync-service/core/dataVerifier"
	"github.com/open-horizon/edge-sync-service/core/security"
	"github.com/open-horizon/edge-utilities/logger"
	"github.com/open-horizon/edge-utilities/logger/log"
	"github.com/open-horizon/edge-utilities/logger/trace"
)
const registerURL = "/spi/v1/register/"
const registerNewURL = "/spi/v1/register-new/"
const unregisterURL = "/spi/v1/unregister/"
const pingURL = "/spi/v1/ping/"
const objectRequestURL = "/spi/v1/objects/"
var unauthorizedBytes = []byte("Unauthorized")
type HTTP struct {
	httpClient          http.Client
	started             bool
	httpPollTimer       *time.Timer
	httpPollStopChannel chan int
	requestWrapper      *httpRequestWrapper
}
type updateMessage struct {
	Type     string
	MetaData common.MetaData
}
type feedbackMessage struct {
	Code          int
	RetryInterval int32
	Reason        string
}
func (communication *HTTP) StartCommunication() common.SyncServiceError {
	if common.Configuration.NodeType == common.CSS {
		http.Handle(registerURL, http.StripPrefix(registerURL, http.HandlerFunc(communication.handleRegister)))
		http.Handle(registerNewURL, http.StripPrefix(registerNewURL, http.HandlerFunc(communication.handleRegisterNew)))
		http.Handle(unregisterURL, http.StripPrefix(unregisterURL, http.HandlerFunc(communication.handleUnregister)))
		http.Handle(pingURL, http.StripPrefix(pingURL, http.HandlerFunc(communication.handlePing)))
		http.Handle(objectRequestURL, http.StripPrefix(objectRequestURL, http.HandlerFunc(communication.handleObjects)))
	} else {
		communication.httpClient = http.Client{
			Transport: &http.Transport{},
			Timeout:   time.Second * time.Duration(common.Configuration.HTTPESSClientTimeout),
		}
		if common.Configuration.HTTPCSSUseSSL && len(common.Configuration.HTTPCSSCACertificate) > 0 {
			var caFile string
			if strings.HasPrefix(common.Configuration.HTTPCSSCACertificate, "/") {
				caFile = common.Configuration.HTTPCSSCACertificate
			} else {
				caFile = common.Configuration.PersistenceRootPath + common.Configuration.HTTPCSSCACertificate
			}
			certificate, err := ioutil.ReadFile(caFile)
			if err != nil {
				if _, ok := err.(*os.PathError); ok {
					certificate = []byte(common.Configuration.HTTPCSSCACertificate)
				} else {
					return err
				}
			}
			caCertPool := x509.NewCertPool()
			caCertPool.AppendCertsFromPEM(certificate)
			tlsConfig := &tls.Config{RootCAs: caCertPool}
			communication.httpClient.Transport = &http.Transport{TLSClientConfig: tlsConfig}
		}
		communication.httpPollStopChannel = make(chan int, 1)
		communication.requestWrapper = newHTTPRequestWrapper(communication.httpClient)
	}
	communication.started = true
	return nil
}
func (communication *HTTP) startPolling() {
	configuredInterval := int(common.Configuration.HTTPPollingInterval) * 1000
	go func() {
		common.GoRoutineStarted()
		keepRunning := true
		initialPoll := true
		interval := 1000
		communication.httpPollTimer = time.NewTimer(time.Millisecond * time.Duration(interval))
		for keepRunning {
			select {
			case <-communication.httpPollTimer.C:
				update := false
				for communication.Poll() {
					update = true
				}
				if initialPoll || update {
					interval = configuredInterval / 10
					update = false
					initialPoll = false
				} else if interval < configuredInterval {
					interval += configuredInterval / 10
				}
				communication.httpPollTimer = time.NewTimer(time.Millisecond * time.Duration(interval))
			case <-communication.httpPollStopChannel:
				keepRunning = false
			}
		}
		communication.httpPollTimer = nil
		common.GoRoutineEnded()
	}()
}
func (communication *HTTP) StopCommunication() common.SyncServiceError {
	communication.started = false
	communication.httpPollStopChannel <- 1
	if communication.httpPollTimer != nil {
		communication.httpPollTimer.Stop()
	}
	communication.requestWrapper.cancel()
	return nil
}
func (communication *HTTP) HandleRegAck() {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Received regack")
	}
	communication.startPolling()
}
func (communication *HTTP) createError(response *http.Response, action string) common.SyncServiceError {
	message := ""
	if response == nil {
		message = fmt.Sprintf("Failed to %s. Received nil response.", action)
	} else {
		message = fmt.Sprintf("Failed to %s. Received code: %d %s.", action, response.StatusCode, response.Status)
		contents, err := ioutil.ReadAll(response.Body)
		if err == nil {
			message += " Error: " + string(contents)
		}
	}
	if log.IsLogging(logger.ERROR) {
		log.Error(message)
	}
	return &Error{message}
}
func (communication *HTTP) handleGetUpdates(writer http.ResponseWriter, request *http.Request) {
	ok, orgID, destType, destID := security.ValidateSPIRequestIdentity(request)
	if !ok {
		writer.WriteHeader(http.StatusForbidden)
		writer.Write(unauthorizedBytes)
		return
	}
	if ok := destinationExists(orgID, destType, destID); !ok {
		writer.WriteHeader(http.StatusFailedDependency)
		return
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In handleGetUpdates. orgID: %s destType: %s destID: %s\n", orgID, destType, destID)
	}
	payload := make([]updateMessage, 0)
	notifications, err := Store.RetrievePendingNotifications(orgID, destType, destID)
	if err != nil {
		if log.IsLogging(logger.ERROR) {
			log.Error(err.Error())
		}
		SendErrorResponse(writer, err, "", 0)
		return
	}
	if len(notifications) == 0 {
		writer.WriteHeader(http.StatusNoContent)
		return
	}
	for _, n := range notifications {
		metaData, err := Store.RetrieveObject(n.DestOrgID, n.ObjectType, n.ObjectID)
		if err != nil {
			message := fmt.Sprintf("Error in handleGetUpdates. Error: %s\n", err)
			if log.IsLogging(logger.ERROR) {
				log.Error(message)
			}
			continue
		}
		if metaData == nil || metaData.InstanceID != n.InstanceID {
			continue
		}
		var status string
		switch n.Status {
		case common.UpdatePending:
			status = common.Update
		case common.ConsumedPending:
			status = common.Consumed
		case common.DeletePending:
			status = common.Delete
		case common.DeletedPending:
			status = common.Deleted
		case common.ReceivedPending:
			status = common.Received
		}
		metaData.DestID = n.DestID
		message := updateMessage{status, *metaData}
		payload = append(payload, message)
	}
	body, err := json.MarshalIndent(payload, "", "  ")
	if err != nil {
		SendErrorResponse(writer, err, "", 0)
		return
	}
	writer.Header().Add("Content-Type", "application/json")
	writer.WriteHeader(http.StatusOK)
	if _, err := writer.Write(body); err != nil && log.IsLogging(logger.ERROR) {
		log.Error("Failed to write response body, error: " + err.Error())
	}
}