mongodb

Documentation

Index

• func IsDup(err error) bool
• type Config
  • func (c *Config) Validate() error
• type DeviceIdFilter
  • func GetNormalizedGetEventsFilter(queries []eventstore.GetEventsQuery) DeviceIdFilter
• type EventStore
  • func New(ctx context.Context, config Config, fileWatcher *fsnotify.Watcher, ...) (*EventStore, error)
  • func (s *EventStore) AddCloseFunc(f func())
  • func (s *EventStore) Clear(ctx context.Context) error
  • func (s *EventStore) ClearCollections(ctx context.Context) error
  • func (s *EventStore) Close(ctx context.Context) error
  • func (s *EventStore) DBName() string
  • func (s *EventStore) Delete(ctx context.Context, queries []eventstore.DeleteQuery) error
  • func (s *EventStore) GetEvents(ctx context.Context, queries []eventstore.GetEventsQuery, timestamp int64, ...) error
  • func (s *EventStore) Insert(ctx context.Context, task maintenance.Task) error
  • func (s *EventStore) LoadFromSnapshot(ctx context.Context, queries []eventstore.SnapshotQuery, ...) error
  • func (s *EventStore) LoadFromVersion(ctx context.Context, queries []eventstore.VersionQuery, eh eventstore.Handler) error
  • func (s *EventStore) LoadUpToVersion(ctx context.Context, queries []eventstore.VersionQuery, eh eventstore.Handler) error
  • func (s *EventStore) Query(ctx context.Context, limit int, taskHandler maintenance.TaskHandler) error
  • func (s *EventStore) Remove(ctx context.Context, task maintenance.Task) error
  • func (s *EventStore) RemoveUpToVersion(ctx context.Context, versionQueries []eventstore.VersionQuery) error
  • func (s *EventStore) Save(ctx context.Context, events ...eventstore.Event) (eventstore.SaveStatus, error)
• type LogDebugfFunc
• type MarshalerFunc
• type MarshalerOpt
  • func WithMarshaler(f MarshalerFunc) MarshalerOpt
• type Option
• type ResourceIdFilter
• type UnmarshalerFunc
• type UnmarshalerOpt
  • func WithUnmarshaler(f UnmarshalerFunc) UnmarshalerOpt

Functions

func IsDup

func IsDup(err error) bool

IsDup check it error is duplicate

Types

type Config

type Config struct {
	Embedded pkgMongo.Config `yaml:",inline" json:",inline"`
	// contains filtered or unexported fields
}

Config provides Mongo DB configuration options

func (*Config) Validate

func (c *Config) Validate() error

type DeviceIdFilter

type DeviceIdFilter struct {
	All       bool
	DeviceIds map[string]ResourceIdFilter
}

func GetNormalizedGetEventsFilter

func GetNormalizedGetEventsFilter(queries []eventstore.GetEventsQuery) DeviceIdFilter

type EventStore

type EventStore struct {
	LogDebugfFunc LogDebugfFunc
	// contains filtered or unexported fields
}

EventStore implements an EventStore for MongoDB.

func New

func New(ctx context.Context, config Config, fileWatcher *fsnotify.Watcher, logger log.Logger, tracerProvider trace.TracerProvider, opts ...Option) (*EventStore, error)

func (*EventStore) AddCloseFunc

func (s *EventStore) AddCloseFunc(f func())

func (*EventStore) Clear

func (s *EventStore) Clear(ctx context.Context) error

Clear clears the event storage.

func (*EventStore) ClearCollections

func (s *EventStore) ClearCollections(ctx context.Context) error

Clear documents in collections, but don't drop the database or the collections

func (*EventStore) Close

func (s *EventStore) Close(ctx context.Context) error

Close closes the database session.

func (*EventStore) DBName

func (s *EventStore) DBName() string

DBName returns db name

func (*EventStore) Delete

func (s *EventStore) Delete(ctx context.Context, queries []eventstore.DeleteQuery) error

Delete documents with given group ids

func (*EventStore) GetEvents

func (s *EventStore) GetEvents(ctx context.Context, queries []eventstore.GetEventsQuery, timestamp int64, eventHandler eventstore.Handler) error

Get events from the eventstore.

func (*EventStore) Insert

func (s *EventStore) Insert(ctx context.Context, task maintenance.Task) error

Insert stores (or updates) the information about the latest snapshot version per aggregate into the DB

func (*EventStore) LoadFromSnapshot

func (s *EventStore) LoadFromSnapshot(ctx context.Context, queries []eventstore.SnapshotQuery, eventHandler eventstore.Handler) error

LoadFromSnapshot loads events from the last snapshot eventstore.

func (*EventStore) LoadFromVersion

func (s *EventStore) LoadFromVersion(ctx context.Context, queries []eventstore.VersionQuery, eh eventstore.Handler) error

LoadFromVersion loads aggregates events from version.

func (*EventStore) LoadUpToVersion

func (s *EventStore) LoadUpToVersion(ctx context.Context, queries []eventstore.VersionQuery, eh eventstore.Handler) error

LoadUpToVersion loads aggregates events up to a specific version.

func (*EventStore) Query

func (s *EventStore) Query(ctx context.Context, limit int, taskHandler maintenance.TaskHandler) error

Query retrieves the latest snapshot version per aggregate for thw number of aggregates specified by 'limit'

func (*EventStore) Remove

func (s *EventStore) Remove(ctx context.Context, task maintenance.Task) error

Remove deletes (the latest snapshot version) database record for a given aggregate ID

func (*EventStore) RemoveUpToVersion

func (s *EventStore) RemoveUpToVersion(ctx context.Context, versionQueries []eventstore.VersionQuery) error

RemoveUpToVersion deletes the aggregated events up to a specific version.

func (*EventStore) Save

func (s *EventStore) Save(ctx context.Context, events ...eventstore.Event) (eventstore.SaveStatus, error)

Save save events to eventstore. AggregateID, GroupID and EventType are required. All events within one Save operation shall have the same AggregateID and GroupID. Versions shall be unique and ascend continually. Only first event can be a snapshot.

type LogDebugfFunc

type LogDebugfFunc = func(fmt string, args ...interface{})

type MarshalerFunc

type MarshalerFunc = func(v interface{}) ([]byte, error)

MarshalerFunc marshal struct to bytes.

type MarshalerOpt

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

func WithMarshaler

func WithMarshaler(f MarshalerFunc) MarshalerOpt

WithMarshaler provides the possibility to set an marshaling function for the config

type Option

type Option interface {
	// contains filtered or unexported methods
}

Option provides the means to use function call chaining

type ResourceIdFilter

type ResourceIdFilter struct {
	All         bool
	ResourceIds strings.Set
}

type UnmarshalerFunc

type UnmarshalerFunc = func(b []byte, v interface{}) error

UnmarshalerFunc unmarshal bytes to pointer of struct.

type UnmarshalerOpt

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

func WithUnmarshaler

func WithUnmarshaler(f UnmarshalerFunc) UnmarshalerOpt

WithUnmarshaler provides the possibility to set an unmarshaling function for the config