关于ddd:聊聊eventhorizon的Aggregate

序

本文次要钻研一下 eventhorizon 的 Aggregate

Aggregate

eventhorizon/aggregate.go

type Aggregate interface {
    // Entity provides the ID of the aggregate.
    Entity

    // AggregateType returns the type name of the aggregate.
    // AggregateType() string
    AggregateType() AggregateType

    // CommandHandler is used to handle commands.
    CommandHandler
}

type Entity interface {
    // EntityID returns the ID of the entity.
    EntityID() uuid.UUID}

type CommandHandler interface {HandleCommand(context.Context, Command) error
}

// AggregateType is the type of an aggregate.
type AggregateType string

// String returns the string representation of an aggregate type.
func (at AggregateType) String() string {return string(at)
}

Aggregate 接口内嵌了 Entity 及 CommandHandler，定义了 AggregateType 办法

AggregateStore

eventhorizon/aggregate.go

// AggregateStore is responsible for loading and saving aggregates.
type AggregateStore interface {
    // Load loads the most recent version of an aggregate with a type and id.
    Load(context.Context, AggregateType, uuid.UUID) (Aggregate, error)

    // Save saves the uncommittend events for an aggregate.
    Save(context.Context, Aggregate) error
}

AggregateStore 定义了 Load、Save 办法

AggregateStore

eventhorizon/aggregatestore/model/aggregatestore.go

// AggregateStore is an aggregate store that uses a read write repo for
// loading and saving aggregates.
type AggregateStore struct {
    repo         eh.ReadWriteRepo
    eventHandler eh.EventHandler
}

var (
    // ErrInvalidRepo is when a dispatcher is created with a nil repo.
    ErrInvalidRepo = errors.New("invalid repo")
    // ErrInvalidAggregate occurs when a loaded aggregate is not an aggregate.
    ErrInvalidAggregate = errors.New("invalid aggregate")
)

// NewAggregateStore creates an aggregate store with a read write repo and an
// event handler that can handle any resulting events (for example by publishing
// them on an event bus).
func NewAggregateStore(repo eh.ReadWriteRepo, eventHandler eh.EventHandler) (*AggregateStore, error) {
    if repo == nil {return nil, ErrInvalidRepo}

    d := &AggregateStore{
        repo:         repo,
        eventHandler: eventHandler,
    }
    return d, nil
}

// Load implements the Load method of the eventhorizon.AggregateStore interface.
func (r *AggregateStore) Load(ctx context.Context, aggregateType eh.AggregateType, id uuid.UUID) (eh.Aggregate, error) {item, err := r.repo.Find(ctx, id)
    if errors.Is(err, eh.ErrEntityNotFound) {
        // Create the aggregate.
        if item, err = eh.CreateAggregate(aggregateType, id); err != nil {return nil, err}
    } else if err != nil {return nil, err}

    aggregate, ok := item.(eh.Aggregate)
    if !ok {return nil, ErrInvalidAggregate}

    return aggregate, nil
}

// Save implements the Save method of the eventhorizon.AggregateStore interface.
func (r *AggregateStore) Save(ctx context.Context, aggregate eh.Aggregate) error {if err := r.repo.Save(ctx, aggregate); err != nil {return err}

    // Handle any events optionally provided by the aggregate.
    if a, ok := aggregate.(eh.EventSource); ok && r.eventHandler != nil {for _, e := range a.Events() {if err := r.eventHandler.HandleEvent(ctx, e); err != nil {return err}
        }
    }

    return nil
}

AggregateStore 定义了 eh.ReadWriteRepo、eh.EventHandler 属性；其 Load、Save 办法均委托给了 eh.ReadWriteRepo

ReadWriteRepo

eventhorizon/repo.go

type ReadWriteRepo interface {
    ReadRepo
    WriteRepo
}

type ReadRepo interface {
    // Parent returns the parent read repository, if there is one.
    // Useful for iterating a wrapped set of repositories to get a specific one.
    Parent() ReadRepo

    // Find returns an entity for an ID.
    Find(context.Context, uuid.UUID) (Entity, error)

    // FindAll returns all entities in the repository.
    FindAll(context.Context) ([]Entity, error)
}

type WriteRepo interface {
    // Save saves a entity in the storage.
    Save(context.Context, Entity) error

    // Remove removes a entity by ID from the storage.
    Remove(context.Context, uuid.UUID) error
}

ReadWriteRepo 接口组合了 ReadRepo、WriteRepo 接口；其中 ReadRepo 接口定义了 Parent、Find、FindAll 办法；WriteRepo 接口定义了 Save、Remove 办法

小结

eventhorizon 的 Aggregate 接口内嵌了 Entity 及 CommandHandler，定义了 AggregateType 办法；AggregateStore 定义了 Load、Save 办法；AggregateStore 的实现类定义了 eh.ReadWriteRepo、eh.EventHandler 属性，其 Load、Save 办法均委托给了 eh.ReadWriteRepo。

doc

• eventhorizon