您好,登录后才能下订单哦!
go.cqrs中DomainRepository的作用是什么,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
// DomainRepository is the interface that all domain repositories should implement. type DomainRepository interface { //Loads an aggregate of the given type and ID Load(aggregateTypeName string, aggregateID string) (AggregateRoot, error) //Saves the aggregate. Save(aggregate AggregateRoot, expectedVersion *int) error }
DomainRepository定义了Load、Save方法
// GetEventStoreCommonDomainRepo is an implementation of the DomainRepository // that uses GetEventStore for persistence type GetEventStoreCommonDomainRepo struct { eventStore *goes.Client eventBus EventBus streamNameDelegate StreamNamer aggregateFactory AggregateFactory eventFactory EventFactory } // Load will load all events from a stream and apply those events to an aggregate // of the type specified. // // The aggregate type and id will be passed to the configured StreamNamer to // get the stream name. func (r *GetEventStoreCommonDomainRepo) Load(aggregateType, id string) (AggregateRoot, error) { if r.aggregateFactory == nil { return nil, fmt.Errorf("The common domain repository has no Aggregate Factory.") } if r.streamNameDelegate == nil { return nil, fmt.Errorf("The common domain repository has no stream name delegate.") } if r.eventFactory == nil { return nil, fmt.Errorf("The common domain has no Event Factory.") } aggregate := r.aggregateFactory.GetAggregate(aggregateType, id) if aggregate == nil { return nil, fmt.Errorf("The repository has no aggregate factory registered for aggregate type: %s", aggregateType) } streamName, err := r.streamNameDelegate.GetStreamName(aggregateType, id) if err != nil { return nil, err } stream := r.eventStore.NewStreamReader(streamName) for stream.Next() { switch err := stream.Err().(type) { case nil: break case *url.Error, *goes.ErrTemporarilyUnavailable: return nil, &ErrRepositoryUnavailable{} case *goes.ErrNoMoreEvents: return aggregate, nil case *goes.ErrUnauthorized: return nil, &ErrUnauthorized{} case *goes.ErrNotFound: return nil, &ErrAggregateNotFound{AggregateType: aggregateType, AggregateID: id} default: return nil, &ErrUnexpected{Err: err} } event := r.eventFactory.GetEvent(stream.EventResponse().Event.EventType) //TODO: No test for meta meta := make(map[string]string) stream.Scan(event, &meta) if stream.Err() != nil { return nil, stream.Err() } em := NewEventMessage(id, event, Int(stream.EventResponse().Event.EventNumber)) for k, v := range meta { em.SetHeader(k, v) } aggregate.Apply(em, false) aggregate.IncrementVersion() } return aggregate, nil } // Save persists an aggregate func (r *GetEventStoreCommonDomainRepo) Save(aggregate AggregateRoot, expectedVersion *int) error { if r.streamNameDelegate == nil { return fmt.Errorf("The common domain repository has no stream name delagate.") } resultEvents := aggregate.GetChanges() streamName, err := r.streamNameDelegate.GetStreamName(typeOf(aggregate), aggregate.AggregateID()) if err != nil { return err } if len(resultEvents) > 0 { evs := make([]*goes.Event, len(resultEvents)) for k, v := range resultEvents { //TODO: There is no test for this code v.SetHeader("AggregateID", aggregate.AggregateID()) evs[k] = goes.NewEvent("", v.EventType(), v.Event(), v.GetHeaders()) } streamWriter := r.eventStore.NewStreamWriter(streamName) err := streamWriter.Append(expectedVersion, evs...) switch e := err.(type) { case nil: break case *goes.ErrConcurrencyViolation: return &ErrConcurrencyViolation{Aggregate: aggregate, ExpectedVersion: expectedVersion, StreamName: streamName} case *goes.ErrUnauthorized: return &ErrUnauthorized{} case *goes.ErrTemporarilyUnavailable: return &ErrRepositoryUnavailable{} default: return &ErrUnexpected{Err: e} } } aggregate.ClearChanges() for k, v := range resultEvents { if expectedVersion == nil { r.eventBus.PublishEvent(v) } else { em := NewEventMessage(v.AggregateID(), v.Event(), Int(*expectedVersion+k+1)) r.eventBus.PublishEvent(em) } } return nil }
GetEventStoreCommonDomainRepo定义了eventStore、eventBus、streamNameDelegate、aggregateFactory、eventFactory属性,其Load方法先通过r.aggregateFactory.GetAggregate获取aggregate,再通过r.streamNameDelegate.GetStreamName(aggregateType, id)获取streamName,然后通过r.eventStore.NewStreamReader去遍历event,挨个执行aggregate.Apply(em, false)及aggregate.IncrementVersion();其Save方法先通过aggregate.GetChanges()获取resultEvents,再遍历resultEvents构造goes.Event,之后通过streamWriter.Append写入,然后执行aggregate.ClearChanges(),最后执行r.eventBus.PublishEvent
go.cqrs的DomainRepository定义了Load、Save方法;GetEventStoreCommonDomainRepo实现了DomainRepository接口,其Load方法主要是读取event,然后挨个执行aggregate.Apply;其Save方法主要是将aggregate.GetChanges()转换为event,然后通过streamWriter.Append写入,然后执行aggregate.ClearChanges(),最后执行r.eventBus.PublishEvent。
看完上述内容,你们掌握go.cqrs中DomainRepository的作用是什么的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注亿速云行业资讯频道,感谢各位的阅读!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。