cqrs简单入门(golang)(代码片段)

 1 func handles(command string) gin.HandlerFunc 
 2     return func(c *gin.Context) 
 3         data, err := c.GetRawData()
 4         if err != nil 
 5             c.JSON(http.StatusBadRequest, "")
 6             return
 7         
 8         result := application.HandCommand(data, command)
 9         c.JSON(http.StatusOK, result)
10     
11 

//api2Cmd  路由到领域CMD的映射
var api2Cmd map[string]eh.CommandType

type Result struct 
    Succ bool        `json:"success"`
    Code int         `json:"code"`
    Msg  string      `json:"msg"`  // message
    Data interface `json:"data"` // data object


func HandCommand(postBody []byte, commandKey string) (result Result) 
    cmd, err := eh.CreateCommand(eh.CommandType(commandKey))
    if err != nil 
        result.Msg = "could not create command: " + err.Error()
        return
    
    if err := json.Unmarshal(postBody, &cmd); err != nil 
        result.Msg = "could not decode Json" + err.Error()
        return
    
    ctx := context.Background()
    if err := bus.HandleCommand(ctx, cmd); err != nil 
        result.Msg = "could not handle command: " + err.Error()
        return
    

    result.Succ = true
    result.Msg = "ok"

    return

//commandBus 命令总线
var commandBus = bus.NewCommandHandler()

//eventBus 事件总线
var eventBus = eventbus.NewEventBus(nil)

//
var eventStore eh.EventStore

//aggregateStore 领域事件存储与发布
//var AggregateStore *events.AggregateStore

func InitBus() 
    eventStore, _ = eventstore.NewEventStore("127.0.0.1:27017", "EventStore")
    //AggregateStore, _ = events.NewAggregateStore(eventStore, eventBus)


//RegisterHandler 注册命令的处理
func RegisterHandler(cmd eh.CommandType, cmdHandler eh.Aggregate) 
    err := commandBus.SetHandler(cmdHandler, cmd)
    if err != nil 
        panic(err)
    


//HandleCommand 命令的执行
func HandleCommand(ctx context.Context, cmd eh.Command) error 
    return commandBus.HandleCommand(ctx, cmd)


//RegisterEventHandler 注册事件的处理
func RegisterEventHandler(evtMatcher eh.EventMatcher, evtHandler eh.EventHandler) 
    eventBus.AddHandler(evtMatcher, evtHandler)


//RaiseEvents 异步进行事件的存储 和 发布
func RaiseEvents(ctx context.Context, events []eh.Event, originalVersion int) error 
    go eventStore.Save(ctx, events, originalVersion)
    for _, event := range events 
        err := eventBus.PublishEvent(ctx, event)
        if err != nil 
            return err
        
    

    return nil

    signProtocolAPI := "/api/sign_protocol"
    router.POST(signProtocolAPI, handles(signProtocolAPI))

func init() 
    eh.RegisterCommand(func() eh.Command  return &SignProtocol )


const (
    SignCommand eh.CommandType = "/api/sign_protocol"
)

type SignProtocol struct 
    ID uuid.UUID
    //通道号
    AisleType string `json:"AisleType"`
    //银行code，各平台不一样
    BankCode string `json:"BankCode"`
    //账户类型
    AccountType string `json:"AccountType"`
    //账户属性
    AccountProp string `json:"AccountProp"`
    //银行卡号
    BankCardNo string `json:"BankCardNo"`
    //预留手机号
    ReservePhone string `json:"Tel"`
    //银行卡预留的证件类型
    IDCardType string `json:"IDType"`
    //银行卡开户姓名
    CardName string `json:"CardName"`
    //银行卡预留的证件号码
    IDCardNo string `json:"IDCardNo"`
    //提示标识
    Merrem string `json:"Merrem"`
    //备注
    Remark string `json:"Remark"`


func (c SignProtocol) AggregateID() uuid.UUID           return c.ID 
func (c SignProtocol) CommandType() eh.CommandType      return SignCommand 
func (c SignProtocol) AggregateType() eh.AggregateType  return ""  //Command需要知道具体Aggregate的存在，貌似不甚合理呀！！！

func init() 
    prdctAgg := &ProtocolAggregate
        AggregateBase: events.NewAggregateBase("ProtocolAggregate", uuid.New()),
    
    bus.RegisterHandler(SignCommand, prdctAgg)


type ProtocolAggregate struct 
    *events.AggregateBase


var _ = eh.Aggregate(&ProtocolAggregate)

func (a *ProtocolAggregate) HandleCommand(ctx context.Context, cmd eh.Command) error 
    switch cmd := cmd.(type) 
    case *SignProtocol:
        //命令只需要确定输入参数满足业务校验即可
        err := a.CheckSign()
        if err != nil 
            return err
        
        //实际的业务可以异步进行处理
        go a.CfrmSign(cmd)

        return nil
    
    return fmt.Errorf("couldn‘t handle command")


func (a *ProtocolAggregate) CheckSign() error 
    //校验输入参数是否有效，
    return nil


func (a *ProtocolAggregate) CfrmSign(cmd *SignProtocol) error 

    protocolOrm := &interfaces.ProtocolOrm
        ProtocolNo:   uuid.New().String(),
        AisleType:    cmd.AisleType,
        BankCode:     cmd.BankCode,
        BankCardNo:   cmd.BankCardNo,
        ReservePhone: cmd.ReservePhone,
        CardName:     cmd.CardName,
        IDCardNo:     cmd.IDCardNo,
        Merrem:       cmd.Merrem,
        Remark:       cmd.Remark,
        Status:       1,
    
    protocolOrm.AccountType, _ = strconv.Atoi(cmd.AccountType)
    protocolOrm.AccountProp, _ = strconv.Atoi(cmd.AccountProp)
    protocolOrm.IDCardType, _ = strconv.Atoi(cmd.IDCardType)

    //这里本应该的业务逻辑请求，通过领域服务
    //result := domainser.Allinpay.SignGetCode(protocolOrm)

    protocolRepo := infrastructure.RepoFac.ProtocolRepo
    _, err := protocolRepo.Add(protocolOrm)

    if err != nil 
        return err
    
    ctx := context.Background()
    //业务处理成功后，激活领域事件
    bus.RaiseEvents(ctx, a.Events(), 0)

    return nil