如何在cqrs模式中使用eventstore
本文关键字:eventstore 模式 cqrs | 更新日期: 2023-09-27 18:19:06
我正在研究CQRS
模式。我已经创建了一个与此方法相关的项目,我可以在其中插入和检索数据。我开始知道有两种不同的模型写模型(命令)和读模型(查询)。我只是想知道我写模型的方法是否正确。以及如何在多个用户执行相同操作时使用临时数据库进行事件溯源。
Command.cs
public class Command : Message
{
}
public class Insert : Command
{
public readonly Guid Id;
public readonly string Name;
public Insert(Guid id, string name)
{
Id = id;
Name = name;
}
}
public class Update : Command
{
public readonly Guid Id;
public readonly string NewName;
public readonly int OriginalVersion;
public Update(Guid id, string newName)
{
Id = id;
NewName = newName;
}
}
public class Delete : Command
{
public Guid Id;
public readonly int OriginalVersion;
public Delete(Guid id)
{
Id = id;
}
}
Event.cs
public class Event:Message
{
public int Version;
}
public class Inserted : Event
{
public readonly Guid Id;
public readonly string Name;
public Inserted(Guid id, string name)
{
Id = id;
Name = name;
}
}
public class Updated : Event
{
public readonly Guid Id;
public readonly string NewName;
public readonly int OriginalVersion;
public Updated(Guid id, string newName)
{
Id = id;
NewName = newName;
}
}
public class Deleted : Event
{
public Guid Id;
public Deleted(Guid id)
{
Id = id;
}
}
EventStore.cs
public interface IEventStore
{
void SaveEvents(Guid aggregateId, IEnumerable<Event> events, int expectedVersion);
List<Event> GetEventsForAggregate(Guid aggregateId);
}
public class EventStore : IEventStore
{
private readonly IEventPublisher _publisher;
private struct EventDescriptor
{
public readonly Event EventData;
public readonly Guid Id;
public readonly int Version;
public EventDescriptor(Guid id, Event eventData, int version)
{
EventData = eventData;
Version = version;
Id = id;
}
}
public EventStore(IEventPublisher publisher)
{
_publisher = publisher;
}
private readonly Dictionary<Guid, List<EventDescriptor>> _current = new Dictionary<Guid, List<EventDescriptor>>();
public void SaveEvents(Guid aggregateId, IEnumerable<Event> events, int expectedVersion)
{
List<EventDescriptor> eventDescriptors;
if (!_current.TryGetValue(aggregateId, out eventDescriptors))
{
eventDescriptors = new List<EventDescriptor>();
_current.Add(aggregateId, eventDescriptors);
}
else if (eventDescriptors[eventDescriptors.Count - 1].Version != expectedVersion && expectedVersion != -1)
{
throw new ConcurrencyException();
}
var i = expectedVersion;
foreach (var @event in events)
{
i++;
@event.Version = i;
eventDescriptors.Add(new EventDescriptor(aggregateId, @event, i));
_publisher.Publish(@event);
}
}
public List<Event> GetEventsForAggregate(Guid aggregateId)
{
List<EventDescriptor> eventDescriptors;
if (!_current.TryGetValue(aggregateId, out eventDescriptors))
{
throw new AggregateNotFoundException();
}
return eventDescriptors.Select(desc => desc.EventData).ToList();
}
}
public class AggregateNotFoundException : Exception
{
}
public class ConcurrencyException : Exception
{
}
ReadModel.cs
public interface IReadModelFacade
{
IEnumerable<InventoryItemListDto> GetInventoryItems();
InventoryItemDetailsDto GetInventoryItemDetails(Guid id);
}
public class InventoryItemDetailsDto
{
public Guid Id;
public string Name;
public int CurrentCount;
public int Version;
public InventoryItemDetailsDto(Guid id, string name, int currentCount, int version)
{
Id = id;
Name = name;
CurrentCount = currentCount;
Version = version;
}
}
public class InventoryItemListDto
{
public Guid Id;
public string Name;
public InventoryItemListDto(Guid id, string name)
{
Id = id;
Name = name;
}
}
public class InventoryListView : Handles<Inserted>, Handles<Updated>
{
public void Handle(Inserted message)
{
BullShitDatabase.list.Add(new InventoryItemListDto(message.Id, message.Name));
}
public void Handle(Updated message)
{
var item = BullShitDatabase.list.Find(x => x.Id == message.Id);
item.Name = message.NewName;
}
}
public class InvenotryItemDetailView : Handles<Inserted>, Handles<Updated>
{
public void Handle(Inserted message)
{
BullShitDatabase.details.Add(message.Id, new InventoryItemDetailsDto(message.Id, message.Name, 0, 0));
}
public void Handle(Updated message)
{
InventoryItemDetailsDto d = GetDetailsItem(message.Id);
d.Name = message.NewName;
d.Version = message.Version;
}
private InventoryItemDetailsDto GetDetailsItem(Guid id)
{
InventoryItemDetailsDto d;
if (!BullShitDatabase.details.TryGetValue(id, out d))
{
throw new InvalidOperationException("did not find the original inventory this shouldnt happen");
}
return d;
}
}
public class ReadModelFacade : IReadModelFacade
{
public IEnumerable<InventoryItemListDto> GetInventoryItems()
{
return BullShitDatabase.list;
}
public InventoryItemDetailsDto GetInventoryItemDetails(Guid id)
{
return BullShitDatabase.details[id];
}
}
public static class BullShitDatabase
{
public static Dictionary<Guid, InventoryItemDetailsDto> details = new Dictionary<Guid, InventoryItemDetailsDto>();
public static List<InventoryItemListDto> list = new List<InventoryItemListDto>();
}
不管你是使用EventStore还是任何其他存储机制,你都应该针对接口(契约)进行编码。
但是首先,您的命令IMO没有正确定义,它们应该是携带数据并表示域操作(CRUD或非CRUD)的不可变对象,那么为什么要在命令中定义方法?
将命令定义为类不是问题,您最终需要一个,但为什么不将接口作为所有命令的基本类型呢?(固体原则)
所有的类名(命令/事件)必须是有意义的,也就是说,更新,删除…不要说太多,真的。
我也没看到你的服务层在哪里。服务层应该负责处理命令,那么您打算如何做到这一点呢?
下面你有一个例子,我将如何做(有点抽象,但它给你一个想法):
// Message definitions
public interface IMessage
{
Guid ID {get; set;}
}
public interface IEvent : IMessage
{ }
public interface ICommand : IMessage
{ }
public class DeleteUserCommand : ICommand
{
public Guid ID {get; set;}
public Guid UserId {get; set;}
}
public class UserDeletedEvent : IEvent
{
public Guid ID {get; set;}
public Guid UserId {get; set;}
}
// Repository definitions
public interface IRepository
{ }
public interface IUserRepository : IRepository
{
void DeleteUser(Guid userId);
}
public UserRepository : IUserRepository
{
public void DeleteUser(Guid userId)
{}
}
// Service definitions
public interface IService
{ }
public class UserService : IService, IHandles<DeleteUserCommand>
{
public IUserRepository UserRepository {get; set;}
public void Handle(DeleteUserCommand deleteUserCommand)
{
UserRepository.DeleteUser(deleteUserCommand.Id)
//raise event
}
}