func (objectStorage *ObjectStorage) ForEach(consumer func(key []byte, cachedObject CachedObject) bool, options ...IteratorOption) {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	opts := &IteratorOptions{}
	opts.apply(defaultIteratorOptions...)
	opts.apply(options...)
	if objectStorage.options.keyPartitions == nil && len(opts.optionalPrefix) > 0 {
		panic("prefix iterations are only allowed when the option PartitionKey(....) is set")
	}
	iterations := 0
	var seenElements map[string]types.Empty
	if !opts.skipCache {
		if len(opts.optionalPrefix) == 0 {
			if seenElements = objectStorage.forEachCachedElement(func(key []byte, cachedObject *CachedObjectImpl) bool {
				iterations++
				if (opts.maxIterations != 0) && (iterations > opts.maxIterations) {
					cachedObject.Release(true)
					return false
				}
				return consumer(key, wrapCachedObject(cachedObject, 0))
			}); seenElements == nil {
				return
			}
		} else {
			if seenElements = objectStorage.forEachCachedElementWithPrefix(func(key []byte, cachedObject *CachedObjectImpl) bool {
				iterations++
				if (opts.maxIterations != 0) && (iterations > opts.maxIterations) {
					cachedObject.Release(true)
					return false
				}
				return consumer(key, wrapCachedObject(cachedObject, 0))
			}, opts.optionalPrefix); seenElements == nil {
				return
			}
		}
	}
	if opts.skipStorage {
		return
	}
	consumeFunc := func(key kvstore.Key, value kvstore.Value) bool {
		iterations++
		if (opts.maxIterations != 0) && (iterations > opts.maxIterations) {
			return false
		}
		if _, elementSeen := seenElements[string(key)]; elementSeen {
			return true
		}
		cachedObject, cacheHit := objectStorage.accessCache(key, true)
		if !cacheHit {
			var storableObject StorableObject
			if objectStorage.options.keysOnly {
				var err error
				if storableObject, err = objectStorage.objectFactory(key, nil); err != nil {
					return true
				}
			} else {
				marshaledData := make([]byte, len(value))
				copy(marshaledData, value)
				storableObject = objectStorage.unmarshalObject(key, marshaledData)
			}
			if !typeutils.IsInterfaceNil(storableObject) {
				storableObject.Persist()
			}
			cachedObject.publishResult(storableObject)
		}
		cachedObject.waitForInitialResult()
		if cachedObject.Exists() && !consumer(key, wrapCachedObject(cachedObject, 0)) {
			return false
		}
		return true
	}
	if objectStorage.options.keysOnly {
		_ = objectStorage.options.store.IterateKeys(opts.optionalPrefix, func(key kvstore.Key) bool {
			return consumeFunc(key, []byte{})
		})
		return
	}
	_ = objectStorage.options.store.Iterate(opts.optionalPrefix, consumeFunc)
}
ForEach calls the consumer function on every object residing within the cache and the underlying persistence layer.
package objectstorage
import (
	"errors"
	"sync"
	"sync/atomic"
	"unsafe"
	"github.com/iotaledger/hive.go/events"
	"github.com/iotaledger/hive.go/kvstore"
	"github.com/iotaledger/hive.go/syncutils"
	"github.com/iotaledger/hive.go/timedexecutor"
	"github.com/iotaledger/hive.go/types"
	"github.com/iotaledger/hive.go/typeutils"
)
type ObjectStorage struct {
	objectFactory      StorableObjectFactory
	cachedObjects      map[string]interface{}
	cacheMutex         syncutils.RWMutex
	options            *Options
	size               int
	flushMutex         syncutils.RWMutex
	cachedObjectsEmpty sync.WaitGroup
	shutdown           typeutils.AtomicBool
	partitionsManager  *PartitionsManager
	releaseExecutor    unsafe.Pointer
	Events Events
}
type ConsumerFunc = func(key []byte, cachedObject *CachedObjectImpl) bool
func New(store kvstore.KVStore, objectFactory StorableObjectFactory, optionalOptions ...Option) *ObjectStorage {
	storageOptions := newOptions(store, optionalOptions)
	result := &ObjectStorage{
		objectFactory:     objectFactory,
		cachedObjects:     make(map[string]interface{}),
		partitionsManager: NewPartitionsManager(),
		options:           storageOptions,
		Events: Events{
			ObjectEvicted: events.NewEvent(evictionEvent),
		},
	}
	atomic.StorePointer(&result.releaseExecutor, unsafe.Pointer(timedexecutor.New(storageOptions.releaseExecutorWorkerCount)))
	return result
}
func (objectStorage *ObjectStorage) Put(object StorableObject) CachedObject {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	return objectStorage.putObjectInCache(object)
}
func (objectStorage *ObjectStorage) Store(object StorableObject) CachedObject {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	if !objectStorage.options.persistenceEnabled {
		panic("persistence is disabled - use Put(object StorableObject) instead of Store(object StorableObject)")
	}
	object.Persist()
	object.SetModified()
	return objectStorage.putObjectInCache(object)
}
func (objectStorage *ObjectStorage) GetSize() int {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	objectStorage.flushMutex.RLock()
	objectStorage.cacheMutex.RLock()
	size := objectStorage.size
	objectStorage.cacheMutex.RUnlock()
	objectStorage.flushMutex.RUnlock()
	return size
}
func (objectStorage *ObjectStorage) Get(key []byte) CachedObject {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	cachedObject, cacheHit := objectStorage.accessCache(key, true)
	if !cacheHit {
		cachedObject.publishResult(nil)
	}
	return wrapCachedObject(cachedObject.waitForInitialResult(), 0)
}
func (objectStorage *ObjectStorage) Load(key []byte) CachedObject {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	if !objectStorage.options.persistenceEnabled {
		panic("persistence is disabled - use Get(object StorableObject) instead of Load(object StorableObject)")
	}
	cachedObject, cacheHit := objectStorage.accessCache(key, true)
	if !cacheHit {
		loadedObject := objectStorage.LoadObjectFromStore(key)
		if !typeutils.IsInterfaceNil(loadedObject) {
			loadedObject.Persist()
		}
		cachedObject.publishResult(loadedObject)
	}
	return wrapCachedObject(cachedObject.waitForInitialResult(), 0)
}
func (objectStorage *ObjectStorage) Contains(key []byte, options ...ReadOption) (result bool) {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	opts := &ReadOptions{}
	opts.apply(defaultReadOptions...)
	opts.apply(options...)
	if !opts.skipCache {
		if cachedObject, cacheHit := objectStorage.accessCache(key, false); cacheHit {
			defer cachedObject.Release()
			return cachedObject.waitForInitialResult().Exists()
		}
	}
	if !opts.skipStorage {
		return objectStorage.ObjectExistsInStore(key)
	}
	return false
}
func (objectStorage *ObjectStorage) ComputeIfAbsent(key []byte, remappingFunction func(key []byte) StorableObject) CachedObject {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	cachedObject, cacheHit := objectStorage.accessCache(key, true)
	if cacheHit {
		cachedObject.wg.Wait()
		if cachedObject.updateEmptyResult(func() StorableObject {
			return remappingFunction(key)
		}) {
			cachedObject.storeOnCreation()
		}
	} else {
		loadedObject := objectStorage.LoadObjectFromStore(key)
		if !typeutils.IsInterfaceNil(loadedObject) {
			loadedObject.Persist()
			cachedObject.publishResult(loadedObject)
		} else {
			cachedObject.publishResult(remappingFunction(key))
			cachedObject.storeOnCreation()
		}
	}
	return wrapCachedObject(cachedObject.waitForInitialResult(), 0)
}
func (objectStorage *ObjectStorage) DeleteIfPresent(key []byte) bool {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	deleteExistingEntry := func(cachedObject *CachedObjectImpl) bool {
		cachedObject.wg.Wait()
		if storableObject := cachedObject.Get(); !typeutils.IsInterfaceNil(storableObject) {
			if !storableObject.IsDeleted() {
				storableObject.Delete()
				cachedObject.Release(true)
				return true
			}
		}
		cachedObject.Release(true)
		return false
	}
	cachedObject, cacheHit := objectStorage.accessCache(key, false)
	if cacheHit {
		return deleteExistingEntry(cachedObject)
	}
	cachedObject, cacheHit = objectStorage.accessCache(key, true)
	if cacheHit {
		return deleteExistingEntry(cachedObject)
	}
	objectExistsInStore := objectStorage.ObjectExistsInStore(key)
	if objectExistsInStore {
		cachedObject.blindDelete.Set()
	}
	cachedObject.publishResult(nil)
	cachedObject.Release(true)
	return objectExistsInStore
}
func (objectStorage *ObjectStorage) DeleteIfPresentAndReturn(key []byte) StorableObject {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	deleteExistingEntry := func(cachedObject *CachedObjectImpl) StorableObject {
		cachedObject.wg.Wait()
		if storableObject := cachedObject.Get(); !typeutils.IsInterfaceNil(storableObject) {
			if !storableObject.IsDeleted() {
				storableObject.Delete()
				cachedObject.Release(true)
				return storableObject
			}
		}
		cachedObject.Release(true)
		return nil
	}
	cachedObject, cacheHit := objectStorage.accessCache(key, false)
	if cacheHit {
		return deleteExistingEntry(cachedObject)
	}
	cachedObject, cacheHit = objectStorage.accessCache(key, true)
	if cacheHit {
		return deleteExistingEntry(cachedObject)
	}
	storableObject := objectStorage.LoadObjectFromStore(key)
	if !typeutils.IsInterfaceNil(storableObject) {
		storableObject.Delete()
	}
	cachedObject.publishResult(nil)
	cachedObject.Release(true)
	return storableObject
}
func (objectStorage *ObjectStorage) Delete(key []byte) {
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	deleteExistingEntry := func(cachedObject *CachedObjectImpl) {
		cachedObject.wg.Wait()
		if storableObject := cachedObject.Get(); !typeutils.IsInterfaceNil(storableObject) {
			if !storableObject.IsDeleted() {
				storableObject.Delete()
				cachedObject.Release(true)
				return
			}
		}
		cachedObject.Release(true)
	}
	cachedObject, cacheHit := objectStorage.accessCache(key, false)
	if cacheHit {
		deleteExistingEntry(cachedObject)
		return
	}
	cachedObject, cacheHit = objectStorage.accessCache(key, true)
	if cacheHit {
		deleteExistingEntry(cachedObject)
		return
	}
	cachedObject.blindDelete.Set()
	cachedObject.publishResult(nil)
	cachedObject.Release(true)
}
func (objectStorage *ObjectStorage) StoreIfAbsent(object StorableObject) (result CachedObject, stored bool) {
	if typeutils.IsInterfaceNil(object) {
		return
	}
	if objectStorage.shutdown.IsSet() {
		panic("trying to access shutdown object storage")
	}
	key := object.ObjectStorageKey()
	existingCachedObject, cacheHit := objectStorage.accessCache(key, false)
	if cacheHit {
		cachedObject, updated := objectStorage.updateEmptyCachedObject(existingCachedObject, object)
		if !updated {
			cachedObject.Release()
			return nil, false
		}
		return cachedObject, updated
	}
	objectExists := objectStorage.ObjectExistsInStore(key)
	if objectExists {
		return
	}
	existingCachedObject, cacheHit = objectStorage.accessCache(key, true)
	if cacheHit {
		cachedObject, updated := objectStorage.updateEmptyCachedObject(existingCachedObject, object)
		if !updated {
			cachedObject.Release()
			return nil, false
		}
		return cachedObject, updated
	}
	if loadedObject := objectStorage.LoadObjectFromStore(key); !typeutils.IsInterfaceNil(loadedObject) {
		existingCachedObject.publishResult(loadedObject)
		existingCachedObject.Release()
		return
	}
	object.Persist()
	object.SetModified()
	existingCachedObject.publishResult(object)
	existingCachedObject.storeOnCreation()
	stored = true
	result = wrapCachedObject(existingCachedObject, 0)
	return
}