Part 4: Commands

...

• commands, command handlers, command bus, ...
• show how to add new commands + tests

Since CommandHandlers usually access the AggregateRepository, we first create a structure to facilitate the implementation of subsequent CommandHandlers.

What is a Command?

In the CQRS pattern, a command is a message that represents an intention to change the state of the system. A command is sent to the system and is handled by a CommandHandler. The CommandHandler is responsible for validating the command and emitting events as a result of the command being executed.

Create Command Handler

// domain/aqua/command/handler.go
package command

import "github.com/gradientzero/comby/repo"

type cmdDataHandler struct {
	AggregateRepository *repo.AggregateRepository
}

func NewCommandDataHandler(
	AggregateRepository *repo.AggregateRepository,
) *cmdDataHandler {
	cdh := &cmdDataHandler{}
	cdh.AggregateRepository = AggregateRepository
	return cdh
}

Create New Command

// domain/aqua/command/create.go
package command

import (
	"context"
	"fmt"

	"aqua/predict/domain/aqua/aggregate"

	"github.com/gradientzero/comby/event"
	"github.com/gradientzero/comby/utils"
)

type AddCommand struct {
	AquaUuid string `json:"aquaUuid"`
	Name     string `json:"name"`
}

func (cdh *cmdDataHandler) Add(ctx context.Context, cmdData *AddCommand) ([]event.Event, error) {
	var err error


	// validate uuid
	err = utils.ValidateUuid(cmdData.AquaUuid)
	if err != nil {
		return nil, fmt.Errorf("%s failed - aquaUuid is invalid", utils.GetTypeName(cmdData))
	}

	// retrieve existing aggregate from store
	_agg, _ := cdh.AggregateRepository.GetAggregate(cmdData.AquaUuid, aggregate.NewAggregate)
	if _agg != nil {
		return nil, fmt.Errorf("%s failed - aquaUuid already exist", utils.GetTypeName(cmdData))
	}

	// create new aggregate
	_agg = aggregate.NewAggregate()

	// cast into a concrete type
	agg, ok := _agg.(*aggregate.Aqua)
	if !ok {
		return nil, fmt.Errorf("%s failed - conversion to concrete aggregate failed", utils.GetTypeName(cmdData))
	}
	agg.AggregateUuid = cmdData.AquaUuid

	// execute logic
	err = agg.Add(cmdData.Name)
	if err != nil {
		return nil, err
	}

	// return new events
	return agg.GetUncommittedEvents(), nil
}