mongodb

Documentation

Index

• type ChangeEvent
• type EsRepository
  • func NewStore(connString, database string, opts ...StoreOption) (*EsRepository, error)
  • func (r *EsRepository) Close(ctx context.Context)
  • func (r *EsRepository) Forget(ctx context.Context, request eventsourcing.ForgetRequest, ...) error
  • func (r *EsRepository) GetAggregateEvents(ctx context.Context, aggregateID string, snapVersion int) ([]eventsourcing.Event, error)
  • func (r *EsRepository) GetEvents(ctx context.Context, afterEventID eventid.EventID, batchSize int, ...) ([]eventsourcing.Event, error)
  • func (r *EsRepository) GetLastEventID(ctx context.Context, trailingLag time.Duration, filter store.Filter) (eventid.EventID, error)
  • func (r *EsRepository) GetSnapshot(ctx context.Context, aggregateID string) (eventsourcing.Snapshot, error)
  • func (r *EsRepository) HasIdempotencyKey(ctx context.Context, idempotencyKey string) (bool, error)
  • func (r *EsRepository) MigrateInPlaceCopyReplace(ctx context.Context, revision int, snapshotThreshold uint32, ...) error
  • func (r *EsRepository) SaveEvent(ctx context.Context, eRec eventsourcing.EventRecord) (eventid.EventID, uint32, error)
  • func (r *EsRepository) SaveSnapshot(ctx context.Context, snapshot eventsourcing.Snapshot) error
• type Event
• type Feed
  • func NewFeed(logger log.Logger, connString, database string, opts ...FeedOption) Feed
  • func (m Feed) Feed(ctx context.Context, sinker sink.Sinker) error
• type FeedOption
  • func WithFeedEventsCollection(eventsCollection string) FeedOption
  • func WithPartitions(partitions, partitionsLow, partitionsHi uint32) FeedOption
• type Projector
• type Snapshot
• type StoreOption
  • func WithEventsCollection(eventsCollection string) StoreOption
  • func WithProjector(fn Projector) StoreOption
  • func WithSnapshotsCollection(snapshotsCollection string) StoreOption

Types

type ChangeEvent

type ChangeEvent struct {
	FullDocument Event `bson:"fullDocument,omitempty"`
}

type EsRepository

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

func NewStore

func NewStore(connString, database string, opts ...StoreOption) (*EsRepository, error)

NewStore creates a new instance of MongoEsRepository

func (*EsRepository) Close

func (r *EsRepository) Close(ctx context.Context)

func (*EsRepository) Forget

func (r *EsRepository) Forget(ctx context.Context, request eventsourcing.ForgetRequest, forget func(kind string, body []byte, snapshot bool) ([]byte, error)) error

func (*EsRepository) GetAggregateEvents

func (r *EsRepository) GetAggregateEvents(ctx context.Context, aggregateID string, snapVersion int) ([]eventsourcing.Event, error)

func (*EsRepository) GetEvents

func (r *EsRepository) GetEvents(ctx context.Context, afterEventID eventid.EventID, batchSize int, trailingLag time.Duration, filter store.Filter) ([]eventsourcing.Event, error)

func (*EsRepository) GetLastEventID

func (r *EsRepository) GetLastEventID(ctx context.Context, trailingLag time.Duration, filter store.Filter) (eventid.EventID, error)

func (*EsRepository) GetSnapshot

func (r *EsRepository) GetSnapshot(ctx context.Context, aggregateID string) (eventsourcing.Snapshot, error)

func (*EsRepository) HasIdempotencyKey

func (r *EsRepository) HasIdempotencyKey(ctx context.Context, idempotencyKey string) (bool, error)

func (*EsRepository) MigrateInPlaceCopyReplace

func (r *EsRepository) MigrateInPlaceCopyReplace(
	ctx context.Context,
	revision int,
	snapshotThreshold uint32,
	aggregateFactory func() (eventsourcing.Aggregater, error),
	rehydrateFunc func(eventsourcing.Aggregater, eventsourcing.Event) error,
	encoder eventsourcing.Encoder,
	handler eventsourcing.MigrationHandler,
	aggregateType eventsourcing.AggregateType,
	eventTypeCriteria ...eventsourcing.EventKind,
) error

func (*EsRepository) SaveEvent

func (r *EsRepository) SaveEvent(ctx context.Context, eRec eventsourcing.EventRecord) (eventid.EventID, uint32, error)

func (*EsRepository) SaveSnapshot

func (r *EsRepository) SaveSnapshot(ctx context.Context, snapshot eventsourcing.Snapshot) error

type Event

type Event struct {
	ID               string                      `bson:"_id,omitempty"`
	AggregateID      string                      `bson:"aggregate_id,omitempty"`
	AggregateIDHash  uint32                      `bson:"aggregate_id_hash,omitempty"`
	AggregateVersion uint32                      `bson:"aggregate_version,omitempty"`
	AggregateType    eventsourcing.AggregateType `bson:"aggregate_type,omitempty"`
	Kind             eventsourcing.EventKind     `bson:"kind,omitempty"`
	Body             []byte                      `bson:"body,omitempty"`
	IdempotencyKey   string                      `bson:"idempotency_key,omitempty"`
	Metadata         bson.M                      `bson:"metadata,omitempty"`
	CreatedAt        time.Time                   `bson:"created_at,omitempty"`
	Migrated         int                         `bson:"migrated"`
}

Event is the event data stored in the database

type Feed

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

func NewFeed

func NewFeed(logger log.Logger, connString, database string, opts ...FeedOption) Feed

func (Feed) Feed

func (m Feed) Feed(ctx context.Context, sinker sink.Sinker) error

type FeedOption

type FeedOption func(*Feed)

func WithFeedEventsCollection

func WithFeedEventsCollection(eventsCollection string) FeedOption

func WithPartitions

func WithPartitions(partitions, partitionsLow, partitionsHi uint32) FeedOption

type Projector

type Projector func(mongo.SessionContext, eventsourcing.Event) error

type Snapshot

type Snapshot struct {
	ID               string                      `bson:"_id,omitempty"`
	AggregateID      string                      `bson:"aggregate_id,omitempty"`
	AggregateVersion uint32                      `bson:"aggregate_version,omitempty"`
	AggregateType    eventsourcing.AggregateType `bson:"aggregate_type,omitempty"`
	Body             []byte                      `bson:"body,omitempty"`
	CreatedAt        time.Time                   `bson:"created_at,omitempty"`
}

type StoreOption

type StoreOption func(*EsRepository)

func WithEventsCollection

func WithEventsCollection(eventsCollection string) StoreOption

func WithProjector

func WithProjector(fn Projector) StoreOption

func WithSnapshotsCollection

func WithSnapshotsCollection(snapshotsCollection string) StoreOption