func (store *InMemoryStorage) RetrieveDestination(orgID string, destType string, destID string) (*common.Destination, common.SyncServiceError) {
	return &common.Destination{DestOrgID: orgID, DestType: destType, DestID: destID,
		Communication: common.Configuration.CommunicationProtocol}, nil
}
RetrieveDestination retrieves a destination
package storage
import (
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"strings"
	"time"
	"github.com/open-horizon/edge-sync-service/common"
	"github.com/open-horizon/edge-sync-service/core/dataURI"
	"github.com/open-horizon/edge-utilities/logger"
	"github.com/open-horizon/edge-utilities/logger/log"
)
type InMemoryStorage struct {
	lockChannel   chan int
	objects       map[string]inMemoryObject
	notifications map[string]common.Notification
	webhooks      map[string][]string
	timebase      int64
}
type inMemoryObject struct {
	meta                             common.MetaData
	data                             []byte
	tmpData                          []byte
	status                           string
	remainingConsumers               int
	remainingReceivers               int
	consumedTimestamp                time.Time
	removedDestinationPolicyServices []common.ServiceID
}
func (store *InMemoryStorage) Init() common.SyncServiceError {
	store.lockChannel = make(chan int, 1)
	store.lockChannel <- 1
	store.objects = make(map[string]inMemoryObject)
	store.notifications = make(map[string]common.Notification)
	store.webhooks = make(map[string][]string)
	currentTime := time.Now().UnixNano()
	store.timebase = currentTime
	currentTimeInSeconds := currentTime / 1e9
	persistedTimeBase := currentTimeInSeconds
	dir := common.Configuration.PersistenceRootPath + "/sync/local/"
	err := os.MkdirAll(dir, 0750)
	if err == nil {
		path := dir + "persisted-data"
		persisted := store.readPersistedTimebase(path)
		if persisted != 0 && currentTimeInSeconds <= persisted {
			persistedTimeBase++
			store.timebase = persistedTimeBase * 1e9
		}
		if err := store.writePersistedTimebase(path, persistedTimeBase); err != nil {
			if log.IsLogging(logger.ERROR) {
				log.Fatal("Failed to persist timebase. Error: %s\n", err.Error())
			}
		}
	}
	common.HealthStatus.ReconnectedToDatabase()
	return nil
}
func (store *InMemoryStorage) Stop() {
}
func (store *InMemoryStorage) PerformMaintenance() {
}
func (store *InMemoryStorage) Cleanup(isTest bool) common.SyncServiceError {
	return nil
}
func (store *InMemoryStorage) StoreObject(metaData common.MetaData, data []byte, status string) ([]common.StoreDestinationStatus, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := getObjectCollectionID(metaData)
	if status == common.NotReadyToSend || status == common.ReadyToSend {
		newID := store.getInstanceID()
		metaData.InstanceID = newID
		if data != nil && !metaData.NoData && !metaData.MetaOnly {
			metaData.DataID = newID
		}
	}
	if metaData.MetaOnly {
		if object, ok := store.objects[id]; ok {
			if object.status == common.ConsumedByDest {
				return nil, &Error{"Can't update only the meta data of consumed object"}
			}
			metaData.DataID = object.meta.DataID  
			object.meta = metaData
			object.status = status
			object.remainingConsumers = metaData.ExpectedConsumers
			object.remainingReceivers = metaData.ExpectedConsumers
			if metaData.NoData {
				object.data = nil
			}
			store.objects[id] = object
			return nil, nil
		}
	}
	if metaData.NoData {
		data = nil
	}
	store.objects[id] = inMemoryObject{meta: metaData, data: data, status: status,
		remainingConsumers: metaData.ExpectedConsumers, remainingReceivers: metaData.ExpectedConsumers}
	return nil, nil
}
func (store *InMemoryStorage) StoreObjectData(orgID string, objectType string, objectID string, dataReader io.Reader) (bool, common.SyncServiceError) {
	var data []byte
	var err error
	if data, err = ioutil.ReadAll(dataReader); err != nil {
		return false, err
	}
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		if object.status == common.NotReadyToSend {
			object.status = common.ReadyToSend
		}
		if object.status == common.NotReadyToSend || object.status == common.ReadyToSend {
			newID := store.getInstanceID()
			object.meta.InstanceID = newID
			object.meta.DataID = newID
		}
		object.data = data
		object.meta.ObjectSize = int64(len(object.data))
		store.objects[id] = object
		return true, nil
	}
	return false, nil
}
func (store *InMemoryStorage) StoreObjectTempData(orgID string, objectType string, objectID string, dataReader io.Reader) (bool, common.SyncServiceError) {
	var data []byte
	var err error
	if data, err = ioutil.ReadAll(dataReader); err != nil {
		return false, err
	}
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.tmpData = data
		store.objects[id] = object
		return true, nil
	}
	return false, nil
}
func (store *InMemoryStorage) RemoveObjectTempData(orgID string, objectType string, objectID string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.tmpData = nil
		store.objects[id] = object
		return nil
	}
	return notFound
}
func (store *InMemoryStorage) RetrieveTempObjectData(orgID string, objectType string, objectID string) (io.Reader, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		if object.tmpData != nil && len(object.tmpData) > 0 {
			return bytes.NewReader(object.tmpData), nil
		}
		return nil, nil
	}
	return nil, nil
}
func (store *InMemoryStorage) AppendObjectData(orgID string, objectType string, objectID string, dataReader io.Reader, dataLength uint32,
	offset int64, total int64, isFirstChunk bool, isLastChunk bool) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	object, ok := store.objects[id]
	if ok {
		var data []byte
		if dataLength == 0 {
			dt, err := ioutil.ReadAll(dataReader)
			if err != nil {
				return &Error{"Failed to read object data. Error: " + err.Error()}
			}
			data = dt
			dataLength = uint32(len(data))
		}
		if total < offset+int64(dataLength) {
			total = offset + int64(dataLength)
		}
		if isFirstChunk {
			object.data = make([]byte, total)
		} else {
			object.data = ensureArrayCapacity(object.data, total)
		}
		if data != nil {
			copy(object.data[offset:], data)
		} else {
			count, err := dataReader.Read(object.data[offset:])
			if err != nil {
				return &Error{"Failed to read object data. Error: " + err.Error()}
			}
			if count != int(dataLength) {
				return &Error{fmt.Sprintf("Read %d bytes for the object data, instead of %d", count, dataLength)}
			}
		}
		store.objects[id] = object
		return nil
	}
	return notFound
}
func (store *InMemoryStorage) UpdateObjectStatus(orgID string, objectType string, objectID string, status string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.status = status
		if status == common.ConsumedByDest {
			object.consumedTimestamp = time.Now()
		}
		store.objects[id] = object
		return nil
	}
	return &NotFound{"Object not found"}
}
func (store *InMemoryStorage) UpdateObjectSourceDataURI(orgID string, objectType string, objectID string, sourceDataURI string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.meta.SourceDataURI = sourceDataURI
		store.objects[id] = object
		return nil
	}
	return notFound
}
func (store *InMemoryStorage) RetrieveObjectStatus(orgID string, objectType string, objectID string) (string, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		return object.status, nil
	}
	return "", nil
}
func (store *InMemoryStorage) RetrieveObjectRemainingConsumers(orgID string, objectType string, objectID string) (int, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		return object.remainingConsumers, nil
	}
	return 0, notFound
}
func (store *InMemoryStorage) ResetObjectRemainingConsumers(orgID string, objectType string, objectID string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.remainingConsumers = object.meta.ExpectedConsumers
		store.objects[id] = object
		return nil
	}
	return notFound
}
func (store *InMemoryStorage) DecrementAndReturnRemainingConsumers(orgID string, objectType string, objectID string) (int,
	common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.remainingConsumers--
		store.objects[id] = object
		return object.remainingConsumers, nil
	}
	return 0, notFound
}
func (store *InMemoryStorage) DecrementAndReturnRemainingReceivers(orgID string, objectType string, objectID string) (int,
	common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.remainingReceivers--
		store.objects[id] = object
		return object.remainingReceivers, nil
	}
	return 0, notFound
}
func (store *InMemoryStorage) RetrieveUpdatedObjects(orgID string, objectType string, received bool) ([]common.MetaData, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	result := make([]common.MetaData, 0)
	for _, obj := range store.objects {
		if objectType == obj.meta.ObjectType &&
			(obj.status == common.CompletelyReceived || obj.status == common.ObjDeleted ||
				(obj.status == common.ObjReceived && received)) {
			result = append(result, obj.meta)
		}
	}
	return result, nil
}
func (store *InMemoryStorage) RetrieveObjectsWithDestinationPolicy(orgID string, received bool) ([]common.ObjectDestinationPolicy, common.SyncServiceError) {
	return nil, nil
}
func (store *InMemoryStorage) RetrieveObjectsWithDestinationPolicyByService(orgID, serviceOrgID, serviceName string) ([]common.ObjectDestinationPolicy, common.SyncServiceError) {
	return nil, nil
}
func (store *InMemoryStorage) RetrieveObjectsWithDestinationPolicyUpdatedSince(orgID string, since int64) ([]common.ObjectDestinationPolicy, common.SyncServiceError) {
	return nil, nil
}
func (store *InMemoryStorage) RetrieveObjectsWithFilters(orgID string, destinationPolicy *bool, dpServiceOrgID string, dpServiceName string, dpPropertyName string, since int64, objectType string, objectID string, destinationType string, destinationID string, noData *bool, expirationTimeBefore string, deleted *bool) ([]common.MetaData, common.SyncServiceError) {
	return nil, nil
}
func (store *InMemoryStorage) RetrieveAllObjects(orgID string, objectType string) ([]common.ObjectDestinationPolicy, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	result := make([]common.ObjectDestinationPolicy, 0)
	for _, obj := range store.objects {
		if objectType == obj.meta.ObjectType {
			object := common.ObjectDestinationPolicy{
				OrgID: orgID, ObjectType: objectType, ObjectID: obj.meta.ObjectID,
				DestinationPolicy: obj.meta.DestinationPolicy}
			result = append(result, object)
		}
	}
	return result, nil
}
func (store *InMemoryStorage) RetrieveObjects(orgID string, destType string, destID string, resend int) ([]common.MetaData, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	result := make([]common.MetaData, 0)
	for _, obj := range store.objects {
		if !obj.meta.Inactive && obj.status == common.ReadyToSend &&
			(obj.meta.DestType == "" || obj.meta.DestType == destType || destType == "") &&
			(obj.meta.DestID == "" || obj.meta.DestID == destID || destID == "") {
			result = append(result, obj.meta)
		}
	}
	return result, nil
}
func (store *InMemoryStorage) RetrieveConsumedObjects() ([]common.ConsumedObject, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	result := make([]common.ConsumedObject, 0)
	for _, obj := range store.objects {
		if obj.status == common.ConsumedByDest {
			result = append(result, common.ConsumedObject{MetaData: obj.meta, Timestamp: obj.consumedTimestamp})
		}
	}
	return result, nil
}
func (store *InMemoryStorage) RetrieveObject(orgID string, objectType string, objectID string) (*common.MetaData, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		return &object.meta, nil
	}
	return nil, nil
}
func (store *InMemoryStorage) RetrieveObjectAndStatus(orgID string, objectType string, objectID string) (*common.MetaData, string, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		return &object.meta, object.status, nil
	}
	return nil, "", nil
}
func (store *InMemoryStorage) RetrieveObjectData(orgID string, objectType string, objectID string) (io.Reader, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		if object.data != nil && len(object.data) > 0 {
			return bytes.NewReader(object.data), nil
		}
		return nil, nil
	}
	return nil, nil
}
func (store *InMemoryStorage) CloseDataReader(dataReader io.Reader) common.SyncServiceError {
	switch v := dataReader.(type) {
	case *os.File:
		return v.Close()
	}
	return nil
}
func (store *InMemoryStorage) ReadObjectData(orgID string, objectType string, objectID string, size int, offset int64) ([]byte, bool, int, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		lod := int64(len(object.data))
		if lod <= offset {
			return make([]byte, 0), true, 0, nil
		}
		s := int64(size)
		eof := false
		if s >= lod-offset {
			s = lod - offset
			eof = true
		}
		b := make([]byte, s)
		copy(b, object.data[offset:])
		return b, eof, int(s), nil
	}
	return nil, true, 0, &common.NotFound{}
}
func (store *InMemoryStorage) MarkObjectDeleted(orgID string, objectType string, objectID string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.meta.Deleted = true
		object.status = common.ObjDeleted
		store.objects[id] = object
		return nil
	}
	return notFound
}
func (store *InMemoryStorage) MarkDestinationPolicyReceived(orgID string, objectType string, objectID string) common.SyncServiceError {
	return nil
}
func (store *InMemoryStorage) ActivateObject(orgID string, objectType string, objectID string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.meta.Inactive = false
		store.objects[id] = object
		return nil
	}
	return notFound
}
func (store *InMemoryStorage) GetObjectsToActivate() ([]common.MetaData, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	currentTime := time.Now().UTC().Format(time.RFC3339)
	result := make([]common.MetaData, 0)
	for _, obj := range store.objects {
		if (obj.status == common.NotReadyToSend || obj.status == common.ReadyToSend) &&
			obj.meta.Inactive && obj.meta.ActivationTime != "" && obj.meta.ActivationTime <= currentTime {
			result = append(result, obj.meta)
		}
	}
	return result, nil
}
func (store *InMemoryStorage) DeleteStoredObject(orgID string, objectType string, objectID string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	delete(store.objects, id)
	return nil
}
func (store *InMemoryStorage) DeleteStoredData(orgID string, objectType string, objectID string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	id := createObjectCollectionID(orgID, objectType, objectID)
	if object, ok := store.objects[id]; ok {
		object.data = nil
		store.objects[id] = object
		return nil
	}
	return notFound
}
func (store *InMemoryStorage) CleanObjects() common.SyncServiceError {
	store.lock()
	defer store.unLock()
	for _, obj := range store.objects {
		if obj.status == common.PartiallyReceived || obj.status == common.CompletelyReceived {
			id := createObjectCollectionID(obj.meta.DestOrgID, obj.meta.ObjectType, obj.meta.ObjectID)
			delete(store.objects, id)
		}
	}
	return nil
}
func (store *InMemoryStorage) GetObjectDestinations(metaData common.MetaData) ([]common.Destination, common.SyncServiceError) {
	return []common.Destination{common.Destination{DestOrgID: metaData.DestOrgID, DestType: common.Configuration.DestinationType,
		DestID: common.Configuration.DestinationID, Communication: common.Configuration.CommunicationProtocol}}, nil
}
func (store *InMemoryStorage) UpdateObjectDeliveryStatus(status string, message string, orgID string, objectType string, objectID string,
	destType string, destID string) (bool, common.SyncServiceError) {
	return true, nil
}
func (store *InMemoryStorage) UpdateObjectDelivering(orgID string, objectType string, objectID string) common.SyncServiceError {
	return nil
}
func (store *InMemoryStorage) GetObjectDestinationsList(orgID string, objectType string,
	objectID string) ([]common.StoreDestinationStatus, common.SyncServiceError) {
	return nil, nil
}
func (store *InMemoryStorage) UpdateObjectDestinations(orgID string, objectType string, objectID string, destinationsList []string) (*common.MetaData, string,
	[]common.StoreDestinationStatus, []common.StoreDestinationStatus, common.SyncServiceError) {
	return nil, "", nil, nil, nil
}
func (store *InMemoryStorage) AddObjectDestinations(orgID string, objectType string, objectID string, destinationsList []string) (*common.MetaData, string,
	[]common.StoreDestinationStatus, common.SyncServiceError) {
	return nil, "", nil, nil
}
func (store *InMemoryStorage) DeleteObjectDestinations(orgID string, objectType string, objectID string, destinationsList []string) (*common.MetaData, string,
	[]common.StoreDestinationStatus, common.SyncServiceError) {
	return nil, "", nil, nil
}
func (store *InMemoryStorage) GetNumberOfStoredObjects() (uint32, common.SyncServiceError) {
	var count uint32
	for _, object := range store.objects {
		if object.status == common.ReadyToSend || object.status == common.NotReadyToSend {
			count++
		}
	}
	return count, nil
}
func (store *InMemoryStorage) AddWebhook(orgID string, objectType string, url string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	var hooks []string
	if h := store.webhooks[objectType]; h != nil {
		hooks = h
	} else {
		hooks = make([]string, 0)
	}
	for _, hook := range hooks {
		if url == hook {
			return nil
		}
	}
	hooks = append(hooks, url)
	store.webhooks[objectType] = hooks
	return nil
}
func (store *InMemoryStorage) DeleteWebhook(orgID string, objectType string, url string) common.SyncServiceError {
	store.lock()
	defer store.unLock()
	if hooks := store.webhooks[objectType]; hooks != nil {
		for i, hook := range hooks {
			if strings.EqualFold(hook, url) {
				hooks[i] = hooks[len(hooks)-1]
				store.webhooks[objectType] = hooks[:len(hooks)-1]
				return nil
			}
		}
	}
	return nil
}
func (store *InMemoryStorage) RetrieveWebhooks(orgID string, objectType string) ([]string, common.SyncServiceError) {
	store.lock()
	defer store.unLock()
	if hooks := store.webhooks[objectType]; hooks != nil {
		if len(hooks) == 0 {
			return nil, &NotFound{"No webhooks"}
		}
		return hooks, nil
	}
	return nil, &NotFound{"No webhooks"}
}
func (store *InMemoryStorage) RetrieveDestinations(orgID string, destType string) ([]common.Destination, common.SyncServiceError) {
	return nil, nil
}
func (store *InMemoryStorage) DestinationExists(orgID string, destType string, destID string) (bool, common.SyncServiceError) {
	return true, nil
}
func (store *InMemoryStorage) StoreDestination(destination common.Destination) common.SyncServiceError {
	return nil
}
func (store *InMemoryStorage) DeleteDestination(orgID string, destType string, destID string) common.SyncServiceError {
	return nil
}
func (store *InMemoryStorage) UpdateDestinationLastPingTime(destination common.Destination) common.SyncServiceError {
	return nil
}
func (store *InMemoryStorage) RemoveInactiveDestinations(lastTimestamp time.Time) {}
func (store *InMemoryStorage) GetNumberOfDestinations() (uint32, common.SyncServiceError) {
	return 0, nil
}