golangでcqrs+esにチャレンジしてみる 実装編

はじめに

　前回、下準備したものの続きです。
　今回は下準備したものを使ってTodoアプリっぽく実装してみます。ついでにAkkaやProto.Actorをヒントにアクターシステムっぽくしてみたいと思います。コマンド側とクエリー側それぞれのアクターがコマンド処理やメッセージ処理を行います。

前回からの変更

　前回のを使ってと言いながらいきなり変更します。だって前回のFake永続化、Fakeメッセージングでは実装できなかったんですもの・・・。

persistence.go

package common

import (
    "encoding/json"
    "sort"

    "go.uber.org/zap"
)

type InMemoryDB struct {
    data   map[string][]*StoredEvent
    logger *zap.SugaredLogger
}

func NewInMemoryDB(logger *zap.SugaredLogger) *InMemoryDB {
    return &InMemoryDB{
        data:   make(map[string][]*StoredEvent),
        logger: logger,
    }
}

func (db *InMemoryDB) GetByID(id string) []*StoredEvent {
    results := make([]*StoredEvent, 0)
    results = append(results, db.data[id]...)
    sort.SliceStable(results, func(i, j int) bool {
        switch {
        case results[i].StreamVersion < results[j].StreamVersion:
            return true
        case results[i].StreamVersion > results[j].StreamVersion:
            return false
        default:
            return results[i].OccurredOn < results[j].OccurredOn
        }
    })
    return results
}

func (db *InMemoryDB) Save(e *StoredEvent) {
    db.data[e.AggregateID] = append(db.data[e.AggregateID], e)
}

type PersistenceContext interface {
    ReplayAggregate(a AggregateContext) error
    Save(a AggregateContext) error
}

type FakePersistence struct {
    db     *InMemoryDB
    logger *zap.SugaredLogger
}

func NewFakePersistence(db *InMemoryDB, logger *zap.SugaredLogger) *FakePersistence {
    return &FakePersistence{
        db:     db,
        logger: logger,
    }
}

func (p *FakePersistence) ReplayAggregate(a AggregateContext) error {
    storedEvents := p.db.GetByID(a.AggregateID())
    if err := a.Replay(storedEvents); err != nil {
        return err
    }
    return nil
}

func (p *FakePersistence) Save(a AggregateContext) error {
    for _, e := range a.UncommittedEvents() {
        d, err := json.Marshal(e)
        if err != nil {
            return err
        }
        storedEvent := &StoredEvent{
            AggregateID:   a.AggregateID(),
            StreamVersion: a.StreamVersion(),
            OccurredOn:    e.GetOccurredOn(),
            EventType:     e.GetEventType(),
            Data:          d,
        }
        p.db.Save(storedEvent)
        a.CommitEvent(e)
    }
    a.SetStreamVersion(a.StreamVersion() + 1)
    return nil
}

type PersistenceQueryContext interface {
    QueryEvents(id string, base, limit int64) ([]*StoredEvent, error)
}

type FakePersistenceQuery struct {
    db     *InMemoryDB
    logger *zap.SugaredLogger
}

func NewFakePersistenceQuery(db *InMemoryDB, logger *zap.SugaredLogger) *FakePersistenceQuery {
    return &FakePersistenceQuery{
        db:     db,
        logger: logger,
    }
}

func (p *FakePersistenceQuery) QueryEvents(id string, base, limit int64) ([]*StoredEvent, error) {
    results := make([]*StoredEvent, 0)
    storedEvents := p.db.GetByID(id)
    for _, storedEvent := range storedEvents {
        if storedEvent.StreamVersion >= base && storedEvent.StreamVersion <= limit {
            results = append(results, storedEvent)
        }
    }
    return results, nil
}

type StoredEvent struct {
    AggregateID   string
    StreamVersion int64
    OccurredOn    int64
    EventType     string
    Data          []byte
}

messaging.go

package common

import (
    "context"
    "encoding/json"

    "github.com/google/uuid"
    "github.com/pkg/errors"
    "go.uber.org/zap"
)

type InMemoryMessage struct {
    Header string
    Data   []byte
}

type MessageContext interface {
    GetMessageID() string
    GetMessageType() string
}

func NewMessageID() (string, error) {
    id, err := uuid.NewUUID()
    if err != nil {
        return "", errors.Wrap(err, "IDの生成に失敗しました")
    }

    return id.String(), nil
}

type MessagingProducerContext interface {
    Publish(m MessageContext) error
}

type FakeMessagingProducer struct {
    channel chan<- *InMemoryMessage
    logger  *zap.SugaredLogger
}

func NewFakeMessagingProducer(ch chan<- *InMemoryMessage, logger *zap.SugaredLogger) *FakeMessagingProducer {
    return &FakeMessagingProducer{
        channel: ch,
        logger:  logger,
    }
}

func (p *FakeMessagingProducer) Publish(m MessageContext) error {
    d, err := json.Marshal(m)
    if err != nil {
        return err
    }
    msg := &InMemoryMessage{
        Header: m.GetMessageType(),
        Data:   d,
    }
    p.channel <- msg
    p.logger.Infow("publish message", "message", msg)
    return nil
}

type MessagingConsumerContext interface {
    Consume(ctx context.Context, msg chan<- *InMemoryMessage) error
}

type FakeMessagingConsumer struct {
    channel <-chan *InMemoryMessage
    logger  *zap.SugaredLogger
}

func NewFakeMessagingConsumer(ch <-chan *InMemoryMessage, logger *zap.SugaredLogger) *FakeMessagingConsumer {
    return &FakeMessagingConsumer{
        channel: ch,
        logger:  logger,
    }
}

func (c *FakeMessagingConsumer) Consume(ctx context.Context, msg chan<- *InMemoryMessage) error {
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case m := <-c.channel:
            msg <- m
            c.logger.Infow("consume message", "message", m)
        }
    }
}

　
　デバッグ用にロガーを突っ込んでますが、気にしないでください。Mapを利用したインメモリーデータベースを用意し、そこにイベントを集積します。メッセージングはチャネルを使ってインメモリーで通信します。PubSub（1対多送信）ではなくなってますが今回はこれで通します。

コマンド側

　cqrsのコマンド側をcommandパッケージに準備します。
　まずは集約ルートとしてTodoを定義します。

todo.go

package command

import (
    "errors"

    "github.com/lightstaff/go-dddcqrses/common"
    "github.com/lightstaff/go-dddcqrses/events"
)

type Todo struct {
    *common.AggregateBase
    message   string
    completed bool
}

func NewTodo(id string) *Todo {
    return &Todo{
        AggregateBase: common.NewAggregateBase(id),
    }
}

func (t *Todo) Message() string {
    return t.message
}

func (t *Todo) Completed() bool {
    return t.completed
}

func (t *Todo) RaiseEvent(e common.EventContext, n bool) error {
    switch e := e.(type) {
    case *events.TodoRegistered:
        t.message = e.Message
        t.completed = e.Completed
    case *events.TodoMessageChanged:
        t.message = e.Message
    case *events.TodoCompleted:
        t.completed = e.Completed
    default:
        return errors.New("unknown event")
    }

    if n {
        t.AppendUncommittedEvent(e)
    }

    return nil
}

func (t *Todo) Replay(storedEvents []*common.StoredEvent) error {
    for _, storedEvent := range storedEvents {
        e, err := events.EventConverter(storedEvent.EventType, storedEvent.Data)
        if err != nil {
            return err
        }

        if err := t.RaiseEvent(e, false); err != nil {
            return err
        }

        t.CommitEvent(e)
        t.SetStreamVersion(storedEvent.StreamVersion)
    }

    return nil
}

　ドメイン貧血症感満載ですが、サンプルだからよしとします。
　前回定義した「AggregateBase」を埋め込み、そちらに定義していなかった「Repaly」メソッドを実装しています。「RaiseEvent」でイベントが反映される感じです。

　続いてTodoを操作するアクターを定義します。

todo_actor.go

package command

import (
    "errors"

    "go.uber.org/zap"

    "github.com/lightstaff/go-dddcqrses/common"
    "github.com/lightstaff/go-dddcqrses/events"
    "github.com/lightstaff/go-dddcqrses/messages"
)

type (
    TodoRegistry struct {
        Message   string
        Completed bool
    }

    TodoMessageChange struct {
        AggregateID string
        Message     string
    }

    TodoComplete struct {
        AggregateID string
        Completed   bool
    }
)

type TodoActor struct {
    persistence common.PersistenceContext
    producer    common.MessagingProducerContext
    logger      *zap.SugaredLogger
}

func NewTodoActor(persistence common.PersistenceContext, producer common.MessagingProducerContext, logger *zap.SugaredLogger) *TodoActor {
    return &TodoActor{
        persistence: persistence,
        producer:    producer,
        logger:      logger,
    }
}

func (t *TodoActor) Act(command interface{}) error {
    switch command := command.(type) {
    case *TodoRegistry:
        aggregateID, err := common.NewAggregateID()
        if err != nil {
            return err
        }
        entity := NewTodo(aggregateID)
        e, err := events.NewTodoRegistered(aggregateID, command.Message, command.Completed)
        if err != nil {
            return err
        }
        if err := entity.RaiseEvent(e, true); err != nil {
            return err
        }
        if err := t.persistence.Save(entity); err != nil {
            return err
        }
        m, err := messages.NewTodoEventOccurred(entity.AggregateID(), entity.StreamVersion())
        if err != nil {
            return err
        }
        if err := t.producer.Publish(m); err != nil {
            return err
        }
    case *TodoMessageChange:
        entity := NewTodo(command.AggregateID)
        e, err := events.NewTodoMessageChanged(command.AggregateID, command.Message)
        if err != nil {
            return err
        }
        if err := entity.RaiseEvent(e, true); err != nil {
            return err
        }
        if err := t.persistence.Save(entity); err != nil {
            return err
        }
        m, err := messages.NewTodoEventOccurred(entity.AggregateID(), entity.StreamVersion())
        if err != nil {
            return err
        }
        if err := t.producer.Publish(m); err != nil {
            return err
        }
    case *TodoComplete:
        entity := NewTodo(command.AggregateID)
        e, err := events.NewTodoCompleted(command.AggregateID, command.Completed)
        if err != nil {
            return err
        }
        if err := entity.RaiseEvent(e, true); err != nil {
            return err
        }
        if err := t.persistence.Save(entity); err != nil {
            return err
        }
        m, err := messages.NewTodoEventOccurred(entity.AggregateID(), entity.StreamVersion())
        if err != nil {
            return err
        }
        if err := t.producer.Publish(m); err != nil {
            return err
        }
    }
    return errors.New("unknown command")
}

　頭で定義された各コマンドによって処理を分岐しています。「TodoRegistry」コマンドで作成、「TodoMessageChange」コマンでTodoのメッセージを変更。「TodoComplete」コマンドでTodoを完了させます。Proto.Actorのアクターはもっといろいろ管理させたりもできますが、今回はそこまでやりません。

クエリー側

　cqrsのクエリー側をqueryパッケージに準備します。
　DTOの永続化もインメモリーで定義します。

query_db.go

package query

import "go.uber.org/zap"

type QueryDBContext interface {
    FindByID(id string) *TodoQuery
    Save(entity *TodoQuery)
}

type FakeQueryDB struct {
    data   map[string]*TodoQuery
    logger *zap.SugaredLogger
}

func NewFakeQueryDB(logger *zap.SugaredLogger) *FakeQueryDB {
    return &FakeQueryDB{
        data:   make(map[string]*TodoQuery),
        logger: logger,
    }
}

func (db *FakeQueryDB) FindByID(id string) *TodoQuery {
    if entity, ok := db.data[id]; ok {
        return entity
    }
    return nil
}

func (db *FakeQueryDB) Save(entity *TodoQuery) {
    db.data[entity.AggregateID] = entity
}

　特にコメント無し。
　次にDTOとして「TodoQuery」モデルを定義します。

todo_query.go

package query

import (
    "github.com/lightstaff/go-dddcqrses/common"
    "github.com/lightstaff/go-dddcqrses/events"
)

type TodoQuery struct {
    AggregateID   string
    Message       string
    Completed     bool
    StreamVersion int64
}

func (t *TodoQuery) ApplyEvent(e common.EventContext) error {
    switch e := e.(type) {
    case *events.TodoRegistered:
        t.Message = e.Message
        t.Completed = e.Completed
    case *events.TodoMessageChanged:
        t.Message = e.Message
    case *events.TodoCompleted:
        t.Completed = e.Completed
    }
    return nil
}

　「ApplyEvent」メソッドは下のアクターによって呼ばれます。
　こちらはクエリー側のアクターです。

todo_actor.go

package query

import (
    "github.com/lightstaff/go-dddcqrses/common"
    "github.com/lightstaff/go-dddcqrses/events"
    "github.com/lightstaff/go-dddcqrses/messages"
    "go.uber.org/zap"
)

type TodoActor struct {
    persistenceQuery common.PersistenceQueryContext
    queryDB          QueryDBContext
    logger           *zap.SugaredLogger
}

func NewTodoActor(persistenceQuery common.PersistenceQueryContext, queryDB QueryDBContext, logger *zap.SugaredLogger) *TodoActor {
    return &TodoActor{
        persistenceQuery: persistenceQuery,
        queryDB:          queryDB,
        logger:           logger,
    }
}

func (t *TodoActor) Act(msg *messages.TodoEventOccurred) error {
    target := t.queryDB.FindByID(msg.AggregateID)
    if target == nil {
        target = &TodoQuery{
            AggregateID: msg.AggregateID,
        }
    }
    storedEvents, err := t.persistenceQuery.QueryEvents(msg.AggregateID, target.StreamVersion, msg.StreamVersion)
    if err != nil {
        return err
    }
    for _, storedEvent := range storedEvents {
        e, err := events.EventConverter(storedEvent.EventType, storedEvent.Data)
        if err != nil {
            return err
        }
        target.ApplyEvent(e)
        target.StreamVersion = storedEvent.StreamVersion
        t.logger.Infow("apply event", "event", e)
    }
    t.queryDB.Save(target)
    return nil
}

　アクターはメッセージを受け取り、永続化ストアのクエリから取得したイベントをDTOに流し、保存します。

メイン

　最後にそれぞれを動作させるエントリポイントとしてmainを定義します。

main.go

package main

import (
    "context"
    "os"
    "os/signal"

    "github.com/lightstaff/go-dddcqrses/command"
    "github.com/lightstaff/go-dddcqrses/common"
    "github.com/lightstaff/go-dddcqrses/messages"
    "github.com/lightstaff/go-dddcqrses/query"
    "go.uber.org/zap"
)

func main() {
    logger, err := zap.NewProduction()
    if err != nil {
        panic(err)
    }
    defer logger.Sync()

    sugar := logger.Sugar()

    inMemoryDB := common.NewInMemoryDB(sugar)

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    delivery := make(chan *common.InMemoryMessage)
    errc := make(chan error)

    go func() {
        persistence := common.NewFakePersistence(inMemoryDB, sugar)
        producer := common.NewFakeMessagingProducer(delivery, sugar)
        commandActor := command.NewTodoActor(persistence, producer, sugar)
        commandActor.Act(&command.TodoRegistry{
            Message:   "test message",
            Completed: false,
        })
        sugar.Info("command actor action completed")
    }()

    go func() {
        defer close(delivery)

        persistenceQuery := common.NewFakePersistenceQuery(inMemoryDB, sugar)
        consumer := common.NewFakeMessagingConsumer(delivery, sugar)
        queryDB := query.NewFakeQueryDB(sugar)
        queryActor := query.NewTodoActor(persistenceQuery, queryDB, sugar)

        go func() {
            if err := consumer.Consume(ctx, delivery); err != nil {
                errc <- err
            }
        }()

        for {
            select {
            case <-ctx.Done():
                sugar.Info("call context cancel")
                return
            case m := <-delivery:
                msg, err := messages.MessageConveter(m.Header, m.Data)
                if err != nil {
                    errc <- err
                }
                sugar.Infow("receive message", "message", msg)
                switch msg := msg.(type) {
                case *messages.TodoEventOccurred:
                    if err := queryActor.Act(msg); err != nil {
                        errc <- err
                    }
                    sugar.Info("query actor action completed")
                    sugar.Infow("now todo", "todo", queryDB.FindByID(msg.AggregateID))
                }
            }
        }
    }()

    go func() {
        if err := <-errc; err != nil {
            sugar.Errorw("error happend", "error", err)
            cancel()
        }
    }()

    quit := make(chan os.Signal, 1)
    signal.Notify(quit, os.Interrupt)
    <-quit
}

　今回はインメモリーで永続化やメッセージングをしているのでコマンドとクエリーを分割できませんが、そのへんを別の機構で実装すればそれぞれを独立することが可能なはずです。
　Proto.Actorのリモート周りを使うのも手ですが、protocol-bufferの知識があった方が良いのとホスト側でリモートのインスタンスを立てなきゃいけないので今回は見送りました。

おわりに

　何とかかんとか実装してみたけど、cqrs（DDD含めて）+esとgolangの相性が良いかと言われると微妙な気がする。C#やらJava（Scala)に比べて抽象化はしずらいし（埋め込みでは解決できないこともある）、諸々の命名がいまいち長ったらしくなってしまってgolang感が・・・。

　成果物:https://github.com/lightstaff/go-dddcqrses

参考資料

• エヴァンス本
• 実践ドメイン駆動設計
• .NETのエンタープライズアプリケーションアーキテクチャ第2版
• Akka
• Proto.Actor