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
}
DecrementAndReturnRemainingConsumers decrements the number of remaining consumers of the object
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
}