inmemoryprovider

Documentation

Index

• func NewStorageProvider(logger *zap.Logger, conf *config.Config) (storageprovider.IStorageProvider, error)
• type InMemoryRecord
• type InMemoryStorage
  • func (s *InMemoryStorage) BuildIndex(streamUUID types.StreamUUID) (interface{}, error)
  • func (s *InMemoryStorage) ClearStreams() error
  • func (s *InMemoryStorage) DeleteStream(streamUUID types.StreamUUID) error
  • func (s *InMemoryStorage) GenerateNewStreamUuid() types.StreamUUID
  • func (s *InMemoryStorage) Init() error
  • func (s *InMemoryStorage) LoadStreamFromUUID(streamUUID types.StreamUUID) (*types.StreamInfo, error)
  • func (s *InMemoryStorage) LoadStreams() (types.StreamInfoList, error)
  • func (s *InMemoryStorage) LoadStreamsFromUUIDs(streamUUIDs types.StreamUUIDList) (types.StreamInfoList, error)
  • func (s *InMemoryStorage) NewStreamIteratorHandler(streamUUID types.StreamUUID, iteratorUUID types.StreamIteratorUUID) (types.IStreamIteratorHandler, error)
  • func (s *InMemoryStorage) NewStreamWriter(info *types.StreamInfo) (buffering.IStreamWriter, error)
  • func (s *InMemoryStorage) OnCreateStream(info *types.StreamInfo) error
  • func (s *InMemoryStorage) SaveStreamCatalog(streamUUIDs types.StreamUUIDList) error
  • func (s *InMemoryStorage) Stop() error
  • func (s *InMemoryStorage) StreamExists(streamUUID types.StreamUUID) bool
• type InMemoryStream
  • func NewInMemoryStream(info *types.StreamInfo, maxRecordsByStream uint64, maxSizeInBytes uint64) (*InMemoryStream, error)
  • func (s *InMemoryStream) AddRecord(record *types.DeferedStreamRecord) error
  • func (s *InMemoryStream) GetIndexAfterMessageId(messageId types.MessageId) (uint64, error)
  • func (s *InMemoryStream) GetIndexAtMessageId(messageId types.MessageId) (uint64, error)
  • func (s *InMemoryStream) GetIndexAtTimestamp(timestamp *time.Time) (uint64, error)
  • func (s *InMemoryStream) GetRecordAtIndex(index uint64) (*InMemoryRecord, bool, bool)
  • func (s *InMemoryStream) GetRecordsCount() uint64
• type StreamCatalogInMemory
  • func NewStreamCatalogInMemory(logger *zap.Logger) *StreamCatalogInMemory
  • func (s *StreamCatalogInMemory) Init() error
  • func (s *StreamCatalogInMemory) LoadStreamCatalog() (types.StreamUUIDList, error)
  • func (s *StreamCatalogInMemory) SaveStreamCatalog(streamUUIDs types.StreamUUIDList) error
  • func (s *StreamCatalogInMemory) Stop() error
• type StreamIteratorHandlerInMemory
  • func NewStreamIteratorHandlerInMemory(streamUUID types.StreamUUID, iteratorUUID types.StreamIteratorUUID, ...) *StreamIteratorHandlerInMemory
  • func (h *StreamIteratorHandlerInMemory) Close() error
  • func (h *StreamIteratorHandlerInMemory) GetNextRecord() (types.MessageId, interface{}, bool, bool, error)
  • func (h *StreamIteratorHandlerInMemory) Open() error
  • func (h *StreamIteratorHandlerInMemory) SaveSeek() error
  • func (h *StreamIteratorHandlerInMemory) Seek(request *types.StreamIteratorRequest) error
• type StreamWriterInMemory
  • func NewStreamWriterInMemory(info *types.StreamInfo, inMemoryStream *InMemoryStream, logger *zap.Logger, ...) *StreamWriterInMemory
  • func (w *StreamWriterInMemory) Close() error
  • func (w *StreamWriterInMemory) Init() error
  • func (w *StreamWriterInMemory) Open() error
  • func (w *StreamWriterInMemory) Write(records *[]types.DeferedStreamRecord) error

Functions

func NewStorageProvider

func NewStorageProvider(logger *zap.Logger, conf *config.Config) (storageprovider.IStorageProvider, error)

Types

type InMemoryRecord

type InMemoryRecord struct {
	Id           types.MessageId `json:"i"`
	CreationDate time.Time       `json:"d"`
	Msg          interface{}     `json:"m"`
}

type InMemoryStorage

type InMemoryStorage struct {
	// contains filtered or unexported fields
}

func (*InMemoryStorage) BuildIndex

func (s *InMemoryStorage) BuildIndex(streamUUID types.StreamUUID) (interface{}, error)

func (*InMemoryStorage) ClearStreams

func (s *InMemoryStorage) ClearStreams() error

func (*InMemoryStorage) DeleteStream

func (s *InMemoryStorage) DeleteStream(streamUUID types.StreamUUID) error

func (*InMemoryStorage) GenerateNewStreamUuid

func (s *InMemoryStorage) GenerateNewStreamUuid() types.StreamUUID

func (*InMemoryStorage) Init

func (s *InMemoryStorage) Init() error

func (*InMemoryStorage) LoadStreamFromUUID

func (s *InMemoryStorage) LoadStreamFromUUID(streamUUID types.StreamUUID) (*types.StreamInfo, error)

func (*InMemoryStorage) LoadStreams

func (s *InMemoryStorage) LoadStreams() (types.StreamInfoList, error)

func (*InMemoryStorage) LoadStreamsFromUUIDs

func (s *InMemoryStorage) LoadStreamsFromUUIDs(streamUUIDs types.StreamUUIDList) (types.StreamInfoList, error)

func (*InMemoryStorage) NewStreamIteratorHandler

func (s *InMemoryStorage) NewStreamIteratorHandler(streamUUID types.StreamUUID, iteratorUUID types.StreamIteratorUUID) (types.IStreamIteratorHandler, error)

func (*InMemoryStorage) NewStreamWriter

func (s *InMemoryStorage) NewStreamWriter(info *types.StreamInfo) (buffering.IStreamWriter, error)

func (*InMemoryStorage) OnCreateStream

func (s *InMemoryStorage) OnCreateStream(info *types.StreamInfo) error

func (*InMemoryStorage) SaveStreamCatalog

func (s *InMemoryStorage) SaveStreamCatalog(streamUUIDs types.StreamUUIDList) error

func (*InMemoryStorage) Stop

func (s *InMemoryStorage) Stop() error

func (*InMemoryStorage) StreamExists

func (s *InMemoryStorage) StreamExists(streamUUID types.StreamUUID) bool

type InMemoryStream

type InMemoryStream struct {
	// contains filtered or unexported fields
}

func NewInMemoryStream

func NewInMemoryStream(info *types.StreamInfo, maxRecordsByStream uint64, maxSizeInBytes uint64) (*InMemoryStream, error)

func (*InMemoryStream) AddRecord

func (s *InMemoryStream) AddRecord(record *types.DeferedStreamRecord) error

func (*InMemoryStream) GetIndexAfterMessageId

func (s *InMemoryStream) GetIndexAfterMessageId(messageId types.MessageId) (uint64, error)

func (*InMemoryStream) GetIndexAtMessageId

func (s *InMemoryStream) GetIndexAtMessageId(messageId types.MessageId) (uint64, error)

func (*InMemoryStream) GetIndexAtTimestamp

func (s *InMemoryStream) GetIndexAtTimestamp(timestamp *time.Time) (uint64, error)

func (*InMemoryStream) GetRecordAtIndex

func (s *InMemoryStream) GetRecordAtIndex(index uint64) (*InMemoryRecord, bool, bool)

func (*InMemoryStream) GetRecordsCount

func (s *InMemoryStream) GetRecordsCount() uint64

type StreamCatalogInMemory

type StreamCatalogInMemory struct {
	// contains filtered or unexported fields
}

func NewStreamCatalogInMemory

func NewStreamCatalogInMemory(logger *zap.Logger) *StreamCatalogInMemory

func (*StreamCatalogInMemory) Init

func (s *StreamCatalogInMemory) Init() error

func (*StreamCatalogInMemory) LoadStreamCatalog

func (s *StreamCatalogInMemory) LoadStreamCatalog() (types.StreamUUIDList, error)

func (*StreamCatalogInMemory) SaveStreamCatalog

func (s *StreamCatalogInMemory) SaveStreamCatalog(streamUUIDs types.StreamUUIDList) error

func (*StreamCatalogInMemory) Stop

func (s *StreamCatalogInMemory) Stop() error

type StreamIteratorHandlerInMemory

type StreamIteratorHandlerInMemory struct {
	// contains filtered or unexported fields
}

func NewStreamIteratorHandlerInMemory

func NewStreamIteratorHandlerInMemory(streamUUID types.StreamUUID, iteratorUUID types.StreamIteratorUUID, inMemoryStream *InMemoryStream, logger *zap.Logger) *StreamIteratorHandlerInMemory

func (*StreamIteratorHandlerInMemory) Close

func (h *StreamIteratorHandlerInMemory) Close() error

func (*StreamIteratorHandlerInMemory) GetNextRecord

func (h *StreamIteratorHandlerInMemory) GetNextRecord() (types.MessageId, interface{}, bool, bool, error)

func (*StreamIteratorHandlerInMemory) Open

func (h *StreamIteratorHandlerInMemory) Open() error

func (*StreamIteratorHandlerInMemory) SaveSeek

func (h *StreamIteratorHandlerInMemory) SaveSeek() error

func (*StreamIteratorHandlerInMemory) Seek

func (h *StreamIteratorHandlerInMemory) Seek(request *types.StreamIteratorRequest) error

type StreamWriterInMemory

type StreamWriterInMemory struct {
	// contains filtered or unexported fields
}

func NewStreamWriterInMemory

func NewStreamWriterInMemory(info *types.StreamInfo, inMemoryStream *InMemoryStream, logger *zap.Logger, logVerbosity int) *StreamWriterInMemory

func (*StreamWriterInMemory) Close

func (w *StreamWriterInMemory) Close() error

func (*StreamWriterInMemory) Init

func (w *StreamWriterInMemory) Init() error

func (*StreamWriterInMemory) Open

func (w *StreamWriterInMemory) Open() error

func (*StreamWriterInMemory) Write

func (w *StreamWriterInMemory) Write(records *[]types.DeferedStreamRecord) error