event_store_adapter_go

package module
v0.3.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 13, 2023 License: MIT Imports: 10 Imported by: 0

README

event-store-adapter-go

CI Release tokei

This library is designed to turn DynamoDB into an Event Store for Event Sourcing.

日本語

Usage

You can easily implement an Event Sourcing-enabled repository using EventStore.

type UserAccountRepository struct {
    eventStore         *EventStore
    aggregateConverter AggregateConverter
    eventConverter     EventConverter
}

func (r *UserAccountRepository) Store(event Event, version uint64, aggregate Aggregate) error {
    return r.eventStore.StoreEventAndSnapshotOpt(event, version, aggregate)
}

func (r *UserAccountRepository) FindById(id AggregateId) (*UserAccount, error) {
    result, err := r.eventStore.GetLatestSnapshotById(id, r.aggregateConverter)
    if err != nil {
        return nil, err
    }
    events, err := r.eventStore.GetEventsByIdSinceSeqNr(id, result.SeqNr()+1, r.eventConverter)
    if err != nil {
        return nil, err
    }
    return ReplayUserAccount(events, result.Aggregate().(*UserAccount), result.Version()), nil
}

The following is an example of the repository usage

eventStore := NewEventStore(dynamodbClient, "journal", "snapshot", "journal-aid-index", "snapshot-aid-index", 1)
repository := NewUserAccountRepository(eventStore)

userAccount1, userAccountCreated := NewUserAccount(UserAccountId{Value: "1"}, "test")
// Store an aggregate with a create event
err = repository.Store(userAccountCreated, userAccount1.Version, userAccount1)
if err != nil {
    return err
}

// Replay the aggregate from the event store
userAccount2, err := repository.FindById(&initial.Id)
if err != nil {
	return err
}

// Execute a command on the aggregate
userAccountUpdated, userAccountReanmed := userAccount2.ChangeName("test2")

// Store the new event without a snapshot
err = repository.Store(userAccountReanmed, userAccountUpdated.Version, nil)
// Store the new event with a snapshot
// err = repository.Store(userAccountReanmed, userAccountUpdated.Version, userAccountUpdated)
if err != nil {
    return err
}

Table Specifications

See docs/DATABASE_SCHEMA.md.

License.

MIT License. See LICENSE for details.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Aggregate 

type Aggregate interface {
	fmt.Stringer
	GetId() AggregateId
	GetSeqNr() uint64
	GetVersion() uint64
}

type AggregateConverter 

type AggregateConverter func(map[string]interface{}) (Aggregate, error)

type AggregateId 

type AggregateId interface {
	fmt.Stringer
	GetTypeName() string
	GetValue() string
	AsString() string
}

type AggregateWithSeqNrWithVersion 

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

func (*AggregateWithSeqNrWithVersion) Aggregate 

func (awsv *AggregateWithSeqNrWithVersion) Aggregate() Aggregate

func (*AggregateWithSeqNrWithVersion) Empty added in v0.3.2 

func (awsv *AggregateWithSeqNrWithVersion) Empty() bool

func (*AggregateWithSeqNrWithVersion) Present added in v0.3.2 

func (awsv *AggregateWithSeqNrWithVersion) Present() bool

func (*AggregateWithSeqNrWithVersion) SeqNr 

func (awsv *AggregateWithSeqNrWithVersion) SeqNr() uint64

func (*AggregateWithSeqNrWithVersion) Version 

func (awsv *AggregateWithSeqNrWithVersion) Version() uint64

type DefaultKeyResolver 

type DefaultKeyResolver struct {
}

func (*DefaultKeyResolver) ResolvePkey 

func (kr *DefaultKeyResolver) ResolvePkey(aggregateId AggregateId, shardCount uint64) string

func (*DefaultKeyResolver) ResolveSkey 

func (kr *DefaultKeyResolver) ResolveSkey(aggregateId AggregateId, seqNr uint64) string

type Event 

type Event interface {
	fmt.Stringer
	GetId() string
	GetTypeName() string
	GetAggregateId() AggregateId
	GetSeqNr() uint64
	IsCreated() bool
	GetOccurredAt() uint64
}

type EventConverter 

type EventConverter func(map[string]interface{}) (Event, error)

type EventSerializer 

type EventSerializer interface {
	Serialize(event Event) ([]byte, error)
	Deserialize(data []byte, eventMap *map[string]interface{}) error
}

type EventStore 

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

EventStore for DynamoDB.

func NewEventStore 

func NewEventStore(
	client *dynamodb.Client,
	journalTableName string,
	snapshotTableName string,
	journalAidIndexName string,
	snapshotAidIndexName string,
	shardCount uint64,
	options ...EventStoreOption,
) (*EventStore, error)

NewEventStore returns a new EventStore.

func (*EventStore) GetEventsByIdSinceSeqNr added in v0.3.0 

func (es *EventStore) GetEventsByIdSinceSeqNr(aggregateId AggregateId, seqNr uint64, converter EventConverter) ([]Event, error)

GetEventsByIdSinceSeqNr returns events by aggregateId and seqNr.

- aggregateId is an aggregateId to get events. - seqNr is a seqNr to get events. - converter is a converter to convert a map to an event.

Returns events and an error.

func (*EventStore) GetLatestSnapshotById added in v0.3.0 

func (es *EventStore) GetLatestSnapshotById(aggregateId AggregateId, converter AggregateConverter) (*AggregateWithSeqNrWithVersion, error)

GetLatestSnapshotById returns a snapshot by aggregateId.

- aggregateId is an aggregateId to get a snapshot. - converter is a converter to convert a map to an aggregate.

Returns a snapshot and an error.

func (*EventStore) StoreEvent added in v0.3.2 

func (es *EventStore) StoreEvent(event Event, version uint64) error

func (*EventStore) StoreEventAndSnapshot added in v0.3.2 

func (es *EventStore) StoreEventAndSnapshot(event Event, aggregate Aggregate) error

type EventStoreOption 

type EventStoreOption func(*EventStore) error

EventStoreOption is an option for EventStore.

func WithDeleteTtl added in v0.3.0 

func WithDeleteTtl(deleteTtl time.Duration) EventStoreOption

WithDeleteTtl sets a delete ttl.

- If you want to delete snapshots, specify a time.Duration. - If you do not want to delete snapshots, specify math.MaxInt64. - The default is math.MaxInt64.

Returns an EventStoreOption.

func WithEventSerializer 

func WithEventSerializer(eventSerializer EventSerializer) EventStoreOption

WithEventSerializer sets an event serializer.

- If you want to change the event serializer, specify an EventSerializer. - The default is JsonEventSerializer.

Returns an EventStoreOption.

func WithKeepSnapshot added in v0.3.0 

func WithKeepSnapshot(keepSnapshot bool) EventStoreOption

WithKeepSnapshot sets whether or not to keep snapshots.

- If you want to keep snapshots, specify true. - If you do not want to keep snapshots, specify false. - The default is false.

Returns an EventStoreOption.

func WithKeepSnapshotCount added in v0.3.0 

func WithKeepSnapshotCount(keepSnapshotCount uint32) EventStoreOption

WithKeepSnapshotCount sets a keep snapshot count.

- If you want to keep snapshots, specify a keep snapshot count. - If you do not want to keep snapshots, specify math.MaxInt64. - The default is math.MaxInt64.

Returns an EventStoreOption.

func WithKeyResolver 

func WithKeyResolver(keyResolver KeyResolver) EventStoreOption

WithKeyResolver sets a key resolver.

- If you want to change the key resolver, specify a KeyResolver. - The default is DefaultKeyResolver.

Returns an EventStoreOption.

func WithSnapshotSerializer 

func WithSnapshotSerializer(snapshotSerializer SnapshotSerializer) EventStoreOption

WithSnapshotSerializer sets a snapshot serializer.

- If you want to change the snapshot serializer, specify a SnapshotSerializer. - The default is JsonSnapshotSerializer.

Returns an EventStoreOption.

type JsonEventSerializer 

type JsonEventSerializer struct{}

func (*JsonEventSerializer) Deserialize 

func (s *JsonEventSerializer) Deserialize(data []byte, eventMap *map[string]interface{}) error

func (*JsonEventSerializer) Serialize 

func (s *JsonEventSerializer) Serialize(event Event) ([]byte, error)

type JsonSnapshotSerializer 

type JsonSnapshotSerializer struct{}

func (*JsonSnapshotSerializer) Deserialize 

func (s *JsonSnapshotSerializer) Deserialize(data []byte, aggregateMap *map[string]interface{}) error

func (*JsonSnapshotSerializer) Serialize 

func (s *JsonSnapshotSerializer) Serialize(aggregate Aggregate) ([]byte, error)

type KeyResolver 

type KeyResolver interface {
	ResolvePkey(aggregateId AggregateId, shardCount uint64) string
	ResolveSkey(aggregateId AggregateId, seqNr uint64) string
}

type PkeySkey added in v0.3.0 

type PkeySkey struct {
	Pkey string
	Skey string
}

type SnapshotSerializer 

type SnapshotSerializer interface {
	Serialize(aggregate Aggregate) ([]byte, error)
	Deserialize(data []byte, aggregateMap *map[string]interface{}) error
}

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.