mongodb

Documentation

Index

• type Config
  • func (c Config) String() string
• type EventStore
  • func NewEventStore(config Config, goroutinePoolGo eventstore.GoroutinePoolGoFunc, opts ...Option) (*EventStore, error)
  • func (s *EventStore) Clear(ctx context.Context) error
  • func (s *EventStore) Close(ctx context.Context) error
  • func (s *EventStore) GetInstanceId(ctx context.Context, aggregateId string) (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, ...) error
  • func (s *EventStore) LoadUpToVersion(ctx context.Context, queries []eventstore.VersionQuery, ...) 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) RemoveInstanceId(ctx context.Context, instanceId int64) error
  • func (s *EventStore) RemoveUpToVersion(ctx context.Context, queries []eventstore.VersionQuery) error
  • func (s *EventStore) Save(ctx context.Context, groupId, aggregateId string, events []event.Event) (concurrencyException bool, err error)
  • func (s *EventStore) SaveSnapshot(ctx context.Context, groupId, aggregateId string, event event.Event) (concurrencyException bool, err error)
• type Option
  • func WithMarshaler(f event.MarshalerFunc) Option
  • func WithTLS(cfg *tls.Config) Option
  • func WithUnmarshaler(f event.UnmarshalerFunc) Option

Types

type Config

type Config struct {
	URI             string        `long:"uri" env:"URI" envconfig:"URI" default:"mongodb://localhost:27017"`
	DatabaseName    string        `long:"dbName" env:"DATABASE" envconfig:"DATABASE" default:"eventStore"`
	BatchSize       int           `long:"batchSize" env:"BATCH_SIZE" envconfig:"BATCH_SIZE" default:"16"`
	MaxPoolSize     uint64        `long:"maxPoolSize" env:"MAX_POOL_SIZE" envconfig:"MAX_POOL_SIZE" default:"16"`
	MaxConnIdleTime time.Duration `long:"maxConnIdleTime" env:"MAX_CONN_IDLE_TIME" envconfig:"MAX_CONN_IDLE_TIME" default:"240s"`
	// contains filtered or unexported fields
}

Config provides Mongo DB configuration options

func (Config) String

func (c Config) String() string

String return string representation of Config

type EventStore

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

EventStore implements an EventStore for MongoDB.

func NewEventStore

func NewEventStore(config Config, goroutinePoolGo eventstore.GoroutinePoolGoFunc, opts ...Option) (*EventStore, error)

NewEventStore create a event store from configuration

func (*EventStore) Clear

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

Clear clears the event storage.

func (*EventStore) Close

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

Close closes the database session.

func (*EventStore) GetInstanceId

func (s *EventStore) GetInstanceId(ctx context.Context, aggregateId string) (int64, error)

GetInstanceId returns int64 that is unique

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 event.Handler) error

LoadFromSnapshot loads events from beginning.

func (*EventStore) LoadFromVersion

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

LoadFromVersion loads aggragate events from a specific version.

func (*EventStore) LoadUpToVersion

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

LoadUpToVersion loads aggragate 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) RemoveInstanceId

func (s *EventStore) RemoveInstanceId(ctx context.Context, instanceId int64) error

func (*EventStore) RemoveUpToVersion

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

RemoveUpToVersion deletes the aggragates events up to a specific version.

func (*EventStore) Save

func (s *EventStore) Save(ctx context.Context, groupId, aggregateId string, events []event.Event) (concurrencyException bool, err error)

Save saves events to a path.

func (*EventStore) SaveSnapshot

func (s *EventStore) SaveSnapshot(ctx context.Context, groupId, aggregateId string, event event.Event) (concurrencyException bool, err error)

SaveSnapshot saves snapshots to a path.

type Option

type Option func(Config) Config

Option provides the means to use function call chaining

func WithMarshaler

func WithMarshaler(f event.MarshalerFunc) Option

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

func WithTLS

func WithTLS(cfg *tls.Config) Option

WithTLS configures connection to use TLS

func WithUnmarshaler

func WithUnmarshaler(f event.UnmarshalerFunc) Option

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