func handleObjectDeleted(metaData common.MetaData) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling object deleted of %s %s\n", metaData.ObjectType, metaData.ObjectID)
	}
	lockIndex := common.HashStrings(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
	common.ObjectLocks.Lock(lockIndex)
	notification, err := Store.RetrieveNotificationRecord(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, metaData.DestType, metaData.DestID)
	if err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectDeleted: failed to retrieve notification record. Error: %s\n", err)}
	}
	if notification == nil || notification.InstanceID != metaData.InstanceID ||
		(notification.Status != common.Delete && notification.Status != common.DeletePending && notification.Status != common.AckDelete) {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Ignoring object deleted of %s %s %s %s\n", metaData.ObjectType, metaData.ObjectID, metaData.DestType, metaData.DestID)
		}
		common.ObjectLocks.Unlock(lockIndex)
		Comm.SendNotificationMessage(common.AckDeleted, metaData.DestType, metaData.DestID, metaData.InstanceID, metaData.DataID, &metaData)
		return &ignoredByHandler{"Ignore object deleted"}
	}
	err = Store.DeleteNotificationRecords(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, metaData.DestType, metaData.DestID)
	if err != nil && log.IsLogging(logger.ERROR) {
		log.Error("Error in handleObjectDeleted: failed to delete notification records. Error: %s\n", err)
	}
	removeNotificationChunksInfo(metaData, metaData.OriginType, metaData.OriginID)
	common.ObjectLocks.Unlock(lockIndex)
	if err := Comm.SendNotificationMessage(common.AckDeleted, metaData.DestType, metaData.DestID, metaData.InstanceID, metaData.DataID,
		&metaData); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectDeleted: failed to send notification. Error: %s\n", err)}
	}
	return nil
}
Handle a notification that an object was deleted by the other side
package communications
import (
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"os"
	"sort"
	"sync"
	"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/leader"
	"github.com/open-horizon/edge-sync-service/core/storage"
	"github.com/open-horizon/edge-utilities/logger"
	"github.com/open-horizon/edge-utilities/logger/log"
	"github.com/open-horizon/edge-utilities/logger/trace"
)
type notificationHandlerError struct {
	message string
}
func (e *notificationHandlerError) Error() string {
	return e.message
}
type notificationChunksInfo struct {
	maxRequestedOffset int64
	maxReceivedOffset  int64
	receivedDataSize   int64
	chunkResendTimes   map[int64]int64  
	chunksReceived     []byte           
	chunkSize          int
	resendTime         int64
}
var registerAsNew bool
var notificationLock sync.RWMutex
var consumedLock sync.RWMutex
var dataChunksLocks common.Locks
var notificationChunks map[string]notificationChunksInfo
func init() {
	notificationChunks = make(map[string]notificationChunksInfo)
	dataChunksLocks = *common.NewLocks("notification")
}
func handleRegistration(dest common.Destination, persistentStorage bool) common.SyncServiceError {
	if common.Configuration.NodeType == common.ESS {
		return &notificationHandlerError{"ESS cannot register other services"}
	}
	if !common.IsValidName(dest.DestType) || !common.IsValidName(dest.DestID) {
		return &notificationHandlerError{("Error in handleRegistration: destination contains invalid characters")}
	}
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling registration of %s %s\n", dest.DestType, dest.DestID)
	}
	reconnection, err := Store.DestinationExists(dest.DestOrgID, dest.DestType, dest.DestID)
	if err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleRegistration: failed to check destination's existence. Error: %s\n", err)}
	}
	if !reconnection {
		if err := Comm.RegisterAsNew(dest); err != nil {
			return &notificationHandlerError{"Error in handleRegistration: failed to send register as new notification. Error: " + err.Error()}
		}
		return &ignoredByHandler{}
	}
	if err := Store.StoreDestination(dest); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleRegistration: failed to store destination. Error: %s\n", err)}
	}
	if err := Comm.RegisterAck(dest); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleRegistration: failed to send ack. Error: %s\n", err)}
	}
	if log.IsLogging(logger.INFO) {
		log.Info("Reconnection of: %s %s %s\n", dest.DestOrgID, dest.DestType, dest.DestID)
	}
	if err := resendNotificationsForDestination(dest, !persistentStorage); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleRegistration. Error: %s\n", err)}
	}
	return nil
}
func handleRegisterNew(dest common.Destination, persistentStorage bool) common.SyncServiceError {
	if common.Configuration.NodeType == common.ESS {
		return &notificationHandlerError{"ESS cannot register other services"}
	}
	if !common.IsValidName(dest.DestType) || !common.IsValidName(dest.DestID) {
		return &notificationHandlerError{("Error in handleRegisterNew: destination contains invalid characters")}
	}
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling registration of a new ESS: %s %s\n", dest.DestType, dest.DestID)
	}
	if err := Store.StoreDestination(dest); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleRegisterNew: failed to store destination. Error: %s\n", err)}
	}
	if log.IsLogging(logger.INFO) {
		log.Info("New destination: %s %s %s", dest.DestOrgID, dest.DestType, dest.DestID)
	}
	if err := Comm.RegisterAck(dest); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleRegisterNew: failed to send ack. Error: %s\n", err)}
	}
	resend := common.ResendDelivered
	if persistentStorage {
		resend = common.ResendUndelivered
	}
	objects, err := Store.RetrieveObjects(dest.DestOrgID, dest.DestType, dest.DestID, resend)
	if err != nil {
		return err
	}
	if len(objects) > 0 {
		destinations := make([]common.Destination, 1)
		destinations[0] = dest
		for _, metaData := range objects {
			notificationsInfo, err := PrepareUpdateNotification(metaData, destinations)
			if err == nil {
				if err := SendNotifications(notificationsInfo); err != nil {
					return err
				}
			} else {
				return err
			}
		}
	}
	return nil
}
func handleUnregistration(dest common.Destination) common.SyncServiceError {
	if common.Configuration.NodeType == common.ESS {
		return &notificationHandlerError{"Error: Only CSS can handle the unregistration"}
	}
	if !common.IsValidName(dest.DestType) || !common.IsValidName(dest.DestID) {
		return &notificationHandlerError{("Error in handleUnregistration: destination contains invalid characters")}
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("Handling unregistration of %s %s\n", dest.DestType, dest.DestID)
	}
	var err common.SyncServiceError
	metadataList := make([]common.MetaData, 0)
	if metadataList, err = Store.RetrieveAllObjectsAndUpdateDestinationListForDestination(dest.DestOrgID, dest.DestType, dest.DestID); err != nil {
		trace.Error("Failed to retrieve objects for destination and update destinations for those objects %s %s %s, Error: %s\n", dest.DestOrgID, dest.DestType, dest.DestID, err)
		return err
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("Get %d Metadata for destination %s %s %s: \n", len(metadataList), dest.DestOrgID, dest.DestType, dest.DestID)
		for _, meta := range metadataList {
			trace.Debug("%s/%s\n", meta.ObjectType, meta.ObjectID)
		}
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("Deleting notification records for destination %s %s %s: \n", dest.DestOrgID, dest.DestType, dest.DestID)
	}
	if err := Store.DeleteNotificationRecords(dest.DestOrgID, "", "", dest.DestType, dest.DestID); err != nil {
		trace.Error("Failed to delete notification records for destination %s %s %s, Error: %s\n", dest.DestOrgID, dest.DestType, dest.DestID, err)
		return err
	}
	for _, metaData := range metadataList {
		removeNotificationChunksInfo(metaData, metaData.OriginType, metaData.OriginID)
	}
	if err := Store.DeleteDestination(dest.DestOrgID, dest.DestType, dest.DestID); err != nil {
		trace.Error("Failed to delete destination %s %s %s, Error: %s\n", dest.DestOrgID, dest.DestType, dest.DestID, err)
		return err
	}
	return nil
}
func handlePing(dest common.Destination) common.SyncServiceError {
	if common.Configuration.NodeType == common.ESS {
		return &notificationHandlerError{"ESS received ping"}
	}
	if !common.IsValidName(dest.DestType) || !common.IsValidName(dest.DestID) {
		return &notificationHandlerError{("Error in handlePing: destination contains invalid characters")}
	}
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling ping of %s %s\n", dest.DestType, dest.DestID)
	}
	err := Store.UpdateDestinationLastPingTime(dest)
	if err == nil {
		return nil
	}
	if !storage.IsNotFound(err) {
		return &notificationHandlerError{fmt.Sprintf("Error in handlePing: failed to update destination's last ping time. Error: %s\n", err)}
	}
	if err := Comm.RegisterAsNew(dest); err != nil {
		return &notificationHandlerError{"Error in handlePing: failed to send register as new notification. Error: " + err.Error()}
	}
	return &ignoredByHandler{}
}
func handleRegisterAsNew() common.SyncServiceError {
	if common.Configuration.NodeType == common.CSS {
		return &notificationHandlerError{"CSS received registerAsNew"}
	}
	if err := Store.DeleteNotificationRecords("", "", "", "", ""); err != nil {
		if trace.IsLogging(logger.ERROR) {
			trace.Error("Failed to delete notifications before registering as new. Error: " + err.Error())
		}
	}
	if err := Store.CleanObjects(); err != nil {
		if trace.IsLogging(logger.ERROR) {
			trace.Error("Failed to delete objects before registering as new. Error: " + err.Error())
		}
	}
	registerAsNew = true
	common.Registered = false
	Comm.RegisterNew()
	return nil
}
func handleRegAck() {
	common.Registered = true
	if registerAsNew {
		registerAsNew = false
		objects, err := Store.RetrieveObjects("", "", "", common.ResendAll)
		if err != nil {
			if trace.IsLogging(logger.ERROR) {
				trace.Error("Failed to retrieve objects to resend. Error: " + err.Error())
			}
		}
		if len(objects) > 0 {
			destinations, _ := Store.GetObjectDestinations(objects[0])
			for _, metaData := range objects {
				notificationsInfo, err := PrepareUpdateNotification(metaData, destinations)
				if err == nil {
					if err := SendNotifications(notificationsInfo); err != nil {
						if trace.IsLogging(logger.ERROR) {
							trace.Error(err.Error())
						}
					}
				} else {
					if trace.IsLogging(logger.ERROR) {
						trace.Error(err.Error())
					}
				}
			}
		}
	}
	Comm.HandleRegAck()
}
func handleUpdate(metaData common.MetaData, maxInflightChunks int) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling update of %s %s, instanceID: %d\n", metaData.ObjectType, metaData.ObjectID, metaData.InstanceID)
	}
	lockIndex := common.HashStrings(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
	common.ObjectLocks.Lock(lockIndex)
	notificationDataID := int64(-1)
	if notification, err := Store.RetrieveNotificationRecord(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID,
		metaData.OriginType, metaData.OriginID); err == nil && notification != nil && notification.Status != common.ReceiverError {
		if notification.InstanceID >= metaData.InstanceID {
			if trace.IsLogging(logger.TRACE) {
				trace.Trace("Ignoring object update of %s %s %s %s, notification status: %s, notification.InstanceID: %d, send notification to other side\n", metaData.ObjectType, metaData.ObjectID, metaData.OriginType, metaData.OriginID, notification.Status, notification.InstanceID)
			}
			existingMetaData, status, _ := Store.RetrieveObjectAndStatus(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
			if existingMetaData != nil && notification.InstanceID == metaData.InstanceID && (status == common.CompletelyReceived || status == common.ObjReceived) {
				if trace.IsLogging(logger.DEBUG) {
					trace.Debug("The new object polled %s %s %s %s, existingMetaData.InstanceID: %d, object status %s is already completelyReceived, send %s notification to the other side", metaData.ObjectType, metaData.ObjectID, metaData.OriginType, metaData.OriginID, existingMetaData.InstanceID, status, common.Received)
				}
				common.ObjectLocks.Unlock(lockIndex)
				Comm.SendNotificationMessage(common.Received, metaData.OriginType, metaData.OriginID, metaData.InstanceID, metaData.DataID,
					&metaData)
			} else {
				if trace.IsLogging(logger.DEBUG) {
					if existingMetaData == nil {
						trace.Debug("ExistingMetaData is nil")
					} else {
						trace.Debug("The new object polled %s %s %s %s, existingMetaData status is %s, existingMetaData.InstanceID: %d, send %s notification to the other side\n", metaData.ObjectType, metaData.ObjectID, metaData.OriginType, metaData.OriginID, status, existingMetaData.InstanceID, common.Updated)
					}
				}
				common.ObjectLocks.Unlock(lockIndex)
				Comm.SendNotificationMessage(common.Updated, metaData.OriginType, metaData.OriginID, metaData.InstanceID, metaData.DataID,
					&metaData)
			}
			return &ignoredByHandler{"Ignore object update"}
		}
		if trace.IsLogging(logger.DEBUG) {
			trace.Debug("notification.InstanceID(%d) < metaData.InstanceID(%d), delete local notification record\n", notification.InstanceID, metaData.InstanceID)
		}
		Store.DeleteNotificationRecords(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID,
			metaData.OriginType, metaData.OriginID)
		removeNotificationChunksInfo(metaData, metaData.OriginType, metaData.OriginID)
		notificationDataID = notification.DataID
	} else if notification != nil {
		if trace.IsLogging(logger.DEBUG) {
			trace.Debug("Notification status %s, notificaiton.InstanceID: %d, metadata.InstanceID: %d\n", notification.Status, notification.InstanceID, metaData.InstanceID)
		}
	}
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Finish process notification, then set status to partiallyReceived of %s %s\n", metaData.ObjectType, metaData.ObjectID)
	}
	status := common.PartiallyReceived
	if metaData.Link != "" || metaData.NoData || (metaData.MetaOnly && (metaData.DataID == notificationDataID || metaData.DataID == 0)) {
		if trace.IsLogging(logger.DEBUG) {
			trace.Debug("Set status to completelyReceived for %s %s\n", metaData.ObjectType, metaData.ObjectID)
		}
		status = common.CompletelyReceived
	}
	existingMeta, existingLastDestinationPolicyServices, err := Store.RetrieveObjectAndRemovedDestinationPolicyServices(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
	if err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleUpdate: failed to get existing ESS object and existingLastDestinationPolicyServices. Error: %s\n", err)}
	}
	if existingMeta == nil {
		if trace.IsLogging(logger.DEBUG) {
			trace.Debug("existingMeta is nil\n")
		}
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("existingLastDestinationPolicyServices length: %d\n", len(existingLastDestinationPolicyServices))
	}
	_, existingObjStatus, _ := Store.RetrieveObjectAndStatus(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
	if existingObjStatus == common.ReadyToSend || existingObjStatus == common.NotReadyToSend {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleUpdate: cannot update object from the receiver side.\n")}
	}
	if _, err := Store.StoreObject(metaData, nil, status); err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleUpdate: failed to store object. Error: %s\n", err)}
	}
	if common.Configuration.NodeType == common.ESS {
		if trace.IsLogging(logger.DEBUG) {
			trace.Trace("It is ESS, then compare new polled object with object from ESS storage\n")
		}
		if existingMeta != nil && existingMeta.DestinationPolicy != nil && metaData.DestinationPolicy != nil {
			if trace.IsLogging(logger.DEBUG) {
				trace.Debug("Starting to compare destination policy services...\n")
			}
			removedPolicyServices := common.GetRemovedPolicyServices(existingMeta.DestinationPolicy, metaData.DestinationPolicy)
			if trace.IsLogging(logger.DEBUG) {
				trace.Debug("get %d removed policy services\n", len(removedPolicyServices))
			}
			for _, removedPolicyService := range removedPolicyServices {
				if sliceContains := common.ServiceListContains(existingLastDestinationPolicyServices, removedPolicyService); !sliceContains {
					if trace.IsLogging(logger.TRACE) {
						trace.Trace("Adding destination policy service %s to removedDestinationPolicyServices\n", removedPolicyService.ServiceName)
					}
					existingLastDestinationPolicyServices = append(existingLastDestinationPolicyServices, removedPolicyService)
				} else {
					if trace.IsLogging(logger.TRACE) {
						trace.Trace("existingLastDestinationPolicyServices don't contain service: %s/%s/%s\n", removedPolicyService.OrgID, removedPolicyService.Version, removedPolicyService.ServiceName)
					}
				}
			}
			if trace.IsLogging(logger.DEBUG) {
				trace.Debug("Updating RemovedDestinationPolicyServices in storage...\n")
			}
			if err = Store.UpdateRemovedDestinationPolicyServices(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, existingLastDestinationPolicyServices); err != nil {
				common.ObjectLocks.Unlock(lockIndex)
				return &notificationHandlerError{fmt.Sprintf("Error in handleUpdate: failed to update removedDestinationPolicyServices. Error: %s\n", err)}
			}
		}
	}
	common.ObjectLocks.Unlock(lockIndex)
	if status == common.CompletelyReceived {
		notificationsInfo, err := PrepareObjectStatusNotification(metaData, common.Received)
		if err != nil {
			return err
		}
		return SendNotifications(notificationsInfo)
	}
	if err := Comm.SendNotificationMessage(common.Updated, metaData.OriginType, metaData.OriginID, metaData.InstanceID, metaData.DataID,
		&metaData); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleUpdate: failed to send notification. Error: %s\n", err)}
	}
	Comm.LockDataChunks(lockIndex, &metaData)
	defer Comm.UnlockDataChunks(lockIndex, &metaData)
	if metaData.ChunkSize <= 0 || metaData.ObjectSize <= 0 {
		if err := Comm.GetData(metaData, 0); err != nil {
			return err
		}
	} else {
		var offset int64
		for i := 0; i < maxInflightChunks && offset < metaData.ObjectSize; i++ {
			if err := Comm.GetData(metaData, offset); err != nil {
				return err
			}
			offset += int64(metaData.ChunkSize)
		}
	}
	return nil
}
func handleObjectUpdated(orgID string, objectType string, objectID string, destType string, destID string,
	instanceID int64, dataID int64) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling object updated of %s %s\n", objectType, objectID)
	}
	lockIndex := common.HashStrings(orgID, objectType, objectID)
	common.ObjectLocks.Lock(lockIndex)
	defer common.ObjectLocks.Unlock(lockIndex)
	if trace.IsLogging(logger.DEBUG) {
		trace.Trace("In notificationHandler.go, handleObjectUpdated: retrieve notification record for %s %s %s %s %s\n", orgID, objectType, objectID, destType, destID)
	}
	notification, err := Store.RetrieveNotificationRecord(orgID, objectType, objectID, destType, destID)
	if err != nil || notification == nil {
		return &notificationHandlerError{"Error in handleObjectUpdated: no notification to update."}
	}
	if notification.InstanceID != instanceID || (notification.Status != common.Update && notification.Status != common.UpdatePending) {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Ignoring object updated of %s %s %s %s, notification.InstanceID: %d, instanceID: %d, notification status: %s\n", objectType, objectID, destType, destID, notification.InstanceID, instanceID, notification.Status)
		}
		return &ignoredByHandler{"Ignore object updated"}
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Trace("In notificationHandler.go, handleObjectUpdated: update notification record to updated for %s %s %s %s %s\n", orgID, objectType, objectID, destType, destID)
	}
	Store.UpdateNotificationRecord(
		common.Notification{ObjectID: objectID, ObjectType: objectType,
			DestOrgID: orgID, DestID: destID, DestType: destType, Status: common.Updated, InstanceID: instanceID, DataID: dataID})
	return nil
}
func handleObjectConsumed(orgID string, objectType string, objectID string, destType string, destID string,
	instanceID int64, dataID int64) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling object consumed of %s %s\n", objectType, objectID)
	}
	if common.Configuration.NodeType == common.ESS {
		consumedLock.Lock()
		defer consumedLock.Unlock()
	}
	lockIndex := common.HashStrings(orgID, objectType, objectID)
	common.ObjectLocks.Lock(lockIndex)
	notification, err := Store.RetrieveNotificationRecord(orgID, objectType, objectID, destType, destID)
	if err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectConsumed: failed to retrieve notification record. Error: %s\n", err)}
	}
	metaData, err := Store.RetrieveObject(orgID, objectType, objectID)
	if err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectConsumed: failed to retrieve object. Error: %s\n", err)}
	}
	if notification == nil || metaData == nil || notification.InstanceID != instanceID ||
		(notification.Status != common.Data && notification.Status != common.Updated && notification.Status != common.ReceivedByDestination) {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Ignoring object consumed of %s %s %s %s\n", objectType, objectID, destType, destID)
		}
		common.ObjectLocks.Unlock(lockIndex)
		Comm.SendNotificationMessage(common.AckConsumed, destType, destID, instanceID, dataID,
			&common.MetaData{ObjectType: objectType, ObjectID: objectID, DestOrgID: orgID, DestType: destType, DestID: destID,
				OriginType: common.Configuration.DestinationType, OriginID: common.Configuration.DestinationID, InstanceID: instanceID, DataID: dataID})
		return &ignoredByHandler{"Ignore object consumed"}
	}
	if common.Configuration.NodeType == common.ESS {
		if err := Store.UpdateObjectStatus(orgID, objectType, objectID, common.ConsumedByDest); err != nil {
			common.ObjectLocks.Unlock(lockIndex)
			return err
		}
		consumedObjects, err := Store.RetrieveConsumedObjects()
		if err != nil {
			log.Error("Error in handleObjectConsumed: failed to retrieve consumed objects. Error: %s\n", err)
		} else {
			if float32(len(consumedObjects)) > 1.1*float32(common.Configuration.ESSConsumedObjectsKept) {
				sort.Slice(consumedObjects, func(i, j int) bool {
					return consumedObjects[i].Timestamp.Before(consumedObjects[j].Timestamp)
				})
				for i := 0; i < len(consumedObjects)-common.Configuration.ESSConsumedObjectsKept; i++ {
					objectToDelete := consumedObjects[i].MetaData
					index := common.HashStrings(objectToDelete.DestOrgID, objectToDelete.ObjectType, objectToDelete.ObjectID)
					common.ObjectLocks.ConditionalLock(index, lockIndex)
					stored, status, err := Store.RetrieveObjectAndStatus(objectToDelete.DestOrgID, objectToDelete.ObjectType, objectToDelete.ObjectID)
					if err == nil && status == common.ConsumedByDest && stored.InstanceID == objectToDelete.InstanceID {
						if err = storage.DeleteStoredObject(Store, objectToDelete); err != nil && log.IsLogging(logger.ERROR) {
							log.Error("Error in handleObjectConsumed: failed to delete stored object. Error: %s\n", err)
						}
					}
					common.ObjectLocks.ConditionalUnlock(index, lockIndex)
				}
			}
		}
		if err := storage.DeleteStoredData(Store, *metaData); err != nil && trace.IsLogging(logger.TRACE) {
			trace.Trace("Error in handleObjectConsumed: %s \n", err)
		}
		err = Store.DeleteNotificationRecords(orgID, objectType, objectID, "", "")
		if err != nil && log.IsLogging(logger.ERROR) {
			log.Error("Error in handleObjectConsumed: failed to delete notification records. Error: %s\n", err)
		}
		removeNotificationChunksInfo(*metaData, metaData.OriginType, metaData.OriginID)
	} else {
		_, err = Store.UpdateObjectDeliveryStatus(common.Consumed, "", orgID, objectType, objectID, destType, destID)
		if err != nil && log.IsLogging(logger.ERROR) {
			log.Error("Error in handleObjectConsumed: failed to mark object as delivered to the destination. Error: %s\n", err)
		}
		if err := Store.UpdateNotificationRecord(
			common.Notification{ObjectID: objectID, ObjectType: objectType,
				DestOrgID: orgID, DestID: destID, DestType: destType, Status: common.ConsumedByDestination,
				InstanceID: instanceID, DataID: dataID},
		); err != nil {
			common.ObjectLocks.Unlock(lockIndex)
			return &notificationHandlerError{fmt.Sprintf("Error in handleObjectConsumed: failed to update notification record. Error: %s\n", err)}
		}
	}
	common.ObjectLocks.Unlock(lockIndex)
	if err := Comm.SendNotificationMessage(common.AckConsumed, destType, destID, instanceID, dataID, metaData); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectConsumed: failed to send notification. Error: %s\n",
			err)}
	}
	return nil
}
func handleAckConsumed(orgID string, objectType string, objectID string, destType string, destID string, instanceID int64, dataID int64) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling ack consumed of %s %s\n", objectType, objectID)
	}
	lockIndex := common.HashStrings(orgID, objectType, objectID)
	common.ObjectLocks.Lock(lockIndex)
	defer common.ObjectLocks.Unlock(lockIndex)
	notification, err := Store.RetrieveNotificationRecord(orgID, objectType, objectID, destType, destID)
	if err != nil || notification == nil {
		return &notificationHandlerError{"Error in handleAckConsumed: no notification to update."}
	}
	if notification.InstanceID != instanceID || (notification.Status != common.Consumed && notification.Status != common.ConsumedPending) {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Ignoring ack consumed of %s %s %s %s\n", objectType, objectID, destType, destID)
		}
		return &ignoredByHandler{"Ignore Ack consumed"}
	}
	if err := Store.UpdateNotificationRecord(
		common.Notification{ObjectID: objectID, ObjectType: objectType,
			DestOrgID: orgID, DestID: destID, DestType: destType, Status: common.AckConsumed, InstanceID: instanceID, DataID: dataID},
	); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleAckConsumed: failed to update notification record. Error: %s\n", err)}
	}
	metaData, err := Store.RetrieveObject(orgID, objectType, objectID)
	if err == nil && metaData != nil {
		err = storage.DeleteStoredObject(Store, *metaData)
		if err != nil && log.IsLogging(logger.ERROR) {
			log.Error("Error in handleAckConsumed: failed to delete stored object. Error: %s\n", err)
		}
	}
	err = Store.DeleteNotificationRecords(orgID, objectType, objectID, "", "")
	if err != nil && log.IsLogging(logger.ERROR) {
		log.Error("Error in handleAckConsumed: failed to delete notification records. Error: %s\n", err)
	}
	return nil
}
func handleObjectReceived(orgID string, objectType string, objectID string, destType string, destID string,
	instanceID int64, dataID int64) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling object received of %s %s\n", objectType, objectID)
	}
	lockIndex := common.HashStrings(orgID, objectType, objectID)
	common.ObjectLocks.Lock(lockIndex)
	notification, err := Store.RetrieveNotificationRecord(orgID, objectType, objectID, destType, destID)
	if err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectReceived: failed to retrieve notification record. Error: %s\n", err)}
	}
	metaData, err := Store.RetrieveObject(orgID, objectType, objectID)
	if err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectReceived: failed to retrieve object. Error: %s\n", err)}
	}
	if notification == nil || metaData == nil || notification.InstanceID != instanceID ||
		(notification.Status != common.Data && notification.Status != common.Updated &&
			notification.Status != common.Update && notification.Status != common.UpdatePending) {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Ignoring object received of %s %s for destination %s %s\n", objectType, objectID, destType, destID)
		}
		common.ObjectLocks.Unlock(lockIndex)
		if notification != nil && metaData != nil && notification.InstanceID == instanceID && notification.Status == common.Error {
			if trace.IsLogging(logger.TRACE) {
				trace.Trace("Ignoring object received of %s %s for destination %s %s because notification.Status %s is not error\n", objectType, objectID, destType, destID, notification.Status)
			}
		} else {
			if trace.IsLogging(logger.TRACE) {
				trace.Trace("Ignoring object received of %s %s for destination %s %s, send %s notification to other side\n", objectType, objectID, destType, destID, common.AckReceived)
			}
			Comm.SendNotificationMessage(common.AckReceived, destType, destID, instanceID, dataID,
				&common.MetaData{ObjectType: objectType, ObjectID: objectID, DestOrgID: orgID, DestType: destType, DestID: destID,
					OriginType: common.Configuration.DestinationType, OriginID: common.Configuration.DestinationID, InstanceID: instanceID, DataID: dataID})
		}
		return &ignoredByHandler{"Ignore object received"}
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("Update object status to delivery for %s %s %s %s %s\n", orgID, objectType, objectID, destType, destID)
	}
	_, err = Store.UpdateObjectDeliveryStatus(common.Delivered, "", orgID, objectType, objectID, destType, destID)
	if err != nil && log.IsLogging(logger.ERROR) {
		log.Error("Error in handleObjectReceived: failed to mark object (%s %s %s) as delivered to the destination(%s %s). Error: %s. Sending destination update request to destRequestQueue", orgID, objectType, objectID, destType, destID, err)
		destinationUpdateRequestInQueue := common.DestinationRequestInQueue{
			Action:      common.Update,
			Status:      common.Delivered,
			Object:      *metaData,
			Destination: common.Destination{DestType: destType, DestID: destID},
		}
		DestReqQueue.SendDestReqToQueue(destinationUpdateRequestInQueue)
	}
	if err := Store.UpdateNotificationRecord(
		common.Notification{ObjectID: objectID, ObjectType: objectType,
			DestOrgID: orgID, DestID: destID, DestType: destType, Status: common.ReceivedByDestination, InstanceID: instanceID, DataID: dataID},
	); err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectReceived: failed to update notification record. Error: %s\n", err)}
	}
	common.ObjectLocks.Unlock(lockIndex)
	if err := Comm.SendNotificationMessage(common.AckReceived, destType, destID, instanceID, dataID, metaData); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleObjectReceived: failed to send notification. Error: %s\n",
			err)}
	}
	return nil
}
func handleAckObjectReceived(orgID string, objectType string, objectID string, destType string, destID string, instanceID int64, dataID int64) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling ack received of %s %s\n", objectType, objectID)
	}
	lockIndex := common.HashStrings(orgID, objectType, objectID)
	common.ObjectLocks.Lock(lockIndex)
	defer common.ObjectLocks.Unlock(lockIndex)
	notification, err := Store.RetrieveNotificationRecord(orgID, objectType, objectID, destType, destID)
	if err != nil || notification == nil {
		return &notificationHandlerError{"Error in handleAckObjectReceived: no notification to update."}
	}
	if notification.InstanceID != instanceID || (notification.Status != common.Received && notification.Status != common.ReceivedPending) {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Ignoring ack received of %s %s %s %s, notification InstanceID: %d, InstanceID: %d, notification status: %s\n", objectType, objectID, destType, destID, notification.InstanceID, instanceID, notification.Status)
		}
		return &ignoredByHandler{"Ignore object ack received"}
	}
	if err := Store.UpdateNotificationRecord(
		common.Notification{ObjectID: objectID, ObjectType: objectType,
			DestOrgID: orgID, DestID: destID, DestType: destType, Status: common.AckReceived, InstanceID: instanceID, DataID: dataID},
	); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleAckObjectReceived: failed to update notification record. Error: %s\n", err)}
	}
	return nil
}
func handleDelete(metaData common.MetaData) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling delete of %s %s\n", metaData.ObjectType, metaData.ObjectID)
	}
	lockIndex := common.HashStrings(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
	common.ObjectLocks.Lock(lockIndex)
	sendDeleted := false
	if err := Store.MarkObjectDeleted(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID); err != nil {
		if common.Configuration.NodeType == common.ESS && storage.IsNotFound(err) {
			metaData.Deleted = true
			if _, err := Store.StoreObject(metaData, nil, common.ObjDeleted); err != nil {
				common.ObjectLocks.Unlock(lockIndex)
				return &notificationHandlerError{fmt.Sprintf("Error in handleDelete: failed to recreate deleted object. Error: %s\n", err)}
			}
		} else {
			if trace.IsLogging(logger.TRACE) {
				trace.Trace("In handleDelete: failed to update status of %s %s. Error: %s\n", metaData.ObjectType, metaData.ObjectID, err)
			}
			if common.Configuration.NodeType == common.CSS {
				sendDeleted = true
			}
		}
	} else {
		err = storage.DeleteStoredData(Store, metaData)
		if err != nil && trace.IsLogging(logger.TRACE) {
			trace.Trace("Error in handleDelete: %s \n", err)
		}
		err = Store.ResetObjectRemainingConsumers(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
		if err != nil && trace.IsLogging(logger.TRACE) {
			trace.Trace("Error in handleDelete: %s \n", err)
		}
	}
	Store.DeleteNotificationRecords(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, "", "")
	removeNotificationChunksInfo(metaData, metaData.OriginType, metaData.OriginID)
	common.ObjectLocks.Unlock(lockIndex)
	if sendDeleted {
		if err := Comm.SendNotificationMessage(common.Deleted, metaData.OriginType, metaData.OriginID,
			metaData.InstanceID, metaData.DataID, &metaData); err != nil {
			return &notificationHandlerError{fmt.Sprintf("Error in handleDelete: failed to send notification. Error: %s\n", err)}
		}
	}
	if err := Comm.SendNotificationMessage(common.AckDelete, metaData.OriginType, metaData.OriginID, metaData.InstanceID, metaData.DataID,
		&metaData); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleDelete: failed to send notification. Error: %s\n", err)}
	}
	return nil
}
func handleAckDelete(orgID string, objectType string, objectID string, destType string, destID string, instanceID int64, dataID int64) common.SyncServiceError {
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Handling ack delete of %s %s\n", objectType, objectID)
	}
	lockIndex := common.HashStrings(orgID, objectType, objectID)
	common.ObjectLocks.Lock(lockIndex)
	defer common.ObjectLocks.Unlock(lockIndex)
	notification, err := Store.RetrieveNotificationRecord(orgID, objectType, objectID, destType, destID)
	if err != nil || notification == nil {
		return &notificationHandlerError{"Error in handleAckDelete: no notification to update."}
	}
	if notification.InstanceID != instanceID ||
		(notification.Status != common.Delete && notification.Status != common.DeletePending && notification.Status != common.Deleted) {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Ignoring ack delete of %s %s %s %s\n", objectType, objectID, destType, destID)
		}
		return &ignoredByHandler{"Ignore object ack delete"}
	}
	if err := Store.UpdateNotificationRecord(
		common.Notification{ObjectID: objectID, ObjectType: objectType,
			DestOrgID: orgID, DestID: destID, DestType: destType, Status: common.AckDelete, InstanceID: instanceID, DataID: dataID},
	); err != nil {
		return &notificationHandlerError{fmt.Sprintf("Error in handleAckDelete: failed to update notification record. Error: %s\n", err)}
	}
	deleteObject, err := Store.UpdateObjectDeliveryStatus(common.Deleted, "", orgID, objectType, objectID, destType, destID)
	if err == nil && deleteObject {
		if trace.IsLogging(logger.TRACE) {
			trace.Trace("Deleting object %s:%s:%s\n", orgID, objectType, objectID)
		}
		metaData, err := Store.RetrieveObject(orgID, objectType, objectID)
		if err == nil && metaData != nil {
			return storage.DeleteStoredObject(Store, *metaData)
		}
		return &notificationHandlerError{fmt.Sprintf("Error in handleAckDelete: failed to find object. Error: %s\n", err)}
	}
	return nil
}