event_store_adapter_go

package module
v0.3.4 Latest Latest
Warning

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

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

README

event-store-adapter-go

CI Release

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) error {
    return r.eventStore.PersistEvent(event, version)
}

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

func (r *userAccountRepository) findById(id esag.AggregateId) (*userAccount, error) {
	result, err := r.eventStore.GetLatestSnapshotById(id, r.aggregateConverter)
	if err != nil {
		return nil, err
	}
	if result.Empty() {
		return nil, fmt.Errorf("not found")
	} else {
		events, err := r.eventStore.GetEventsByIdSinceSeqNr(id, result.Aggregate().GetSeqNr()+1, r.eventConverter)
		if err != nil {
			return nil, err
		}
		return replayUserAccount(events, result.Aggregate().(*userAccount)), 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, userAccountRenamed := userAccount2.ChangeName("test2")

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

Table Specifications

See docs/DATABASE_SCHEMA.md.

License.

MIT License. See LICENSE for details.

Other language implementations

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
	WithVersion(version uint64) Aggregate
}

type AggregateConverter 

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

type AggregateId 

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

type AggregateResult added in v0.3.4 

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

func (*AggregateResult) Aggregate added in v0.3.4 

func (awsv *AggregateResult) Aggregate() Aggregate

func (*AggregateResult) Empty added in v0.3.4 

func (awsv *AggregateResult) Empty() bool

func (*AggregateResult) Present added in v0.3.4 

func (awsv *AggregateResult) Present() bool

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) (*AggregateResult, 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) PersistEvent added in v0.3.3 

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

func (*EventStore) PersistEventAndSnapshot added in v0.3.3 

func (es *EventStore) PersistEventAndSnapshot(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 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.