ENode框架Conference案例分析系列之——快速入门篇

git clone https://github.com/tangxuehua/enode-samples.git
cd enode-samples/ConferenceSample

# 启动RabbitMQ容器（带管理UI）
docker run -d --name enode-rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

sequenceDiagram
    Client->>WebAPI: POST /api/attendees/register（提交注册命令）
    WebAPI->>CommandBus: 发送RegisterAttendeeCommand
    CommandBus->>CommandHandler: 路由到对应处理器
    CommandHandler->>AggregateRepository: 加载ConferenceAggregate
    AggregateRepository->>EventStore: 重放事件重建聚合状态
    CommandHandler->>ConferenceAggregate: 调用RegisterAttendee方法
    ConferenceAggregate->>EventBus: 发布AttendeeRegisteredEvent
    EventBus->>Projection: 触发读模型更新
    Projection->>ReadModelDB: 插入AttendeeReadModel记录

{
  "name": "2024全球技术大会",
  "totalSeats": 100,
  "startTime": "2024-10-01T09:00:00Z",
  "endTime": "2024-10-03T18:00:00Z"
}

{
  "conferenceId": "替换为步骤1返回的会议ID",
  "attendeeEmail": "[email protected]",
  "attendeeName": "张三",
  "idempotencyKey": "unique-key-12345"
}

using ENode.Commanding;
 
public class RegisterAttendeeCommand : Command<Guid>
{
    /// <summary>
    /// 会议ID
    /// </summary>
    public Guid ConferenceId { get; set; }
    
    /// <summary>
    /// 参会者邮箱
    /// </summary>
    public string AttendeeEmail { get; set; }
    
    /// <summary>
    /// 参会者姓名
    /// </summary>
    public string AttendeeName { get; set; }
    
    /// <summary>
    /// 幂等键（防止重复提交）
    /// </summary>
    public string IdempotencyKey { get; set; }
}

using ENode.Commanding;
using ENode.Domain;
 
[CommandHandler]
public class RegisterAttendeeCommandHandler : ICommandHandler<RegisterAttendeeCommand>
{
    private readonly IAggregateRepository _aggregateRepository;
 
    public RegisterAttendeeCommandHandler(IAggregateRepository aggregateRepository)
    {
        _aggregateRepository = aggregateRepository;
    }
 
    public async Task HandleAsync(RegisterAttendeeCommand command, CancellationToken cancellationToken)
    {
        // 从事件存储加载会议聚合根
        var conference = await _aggregateRepository.GetAsync<ConferenceAggregate>(command.ConferenceId, cancellationToken);
        
        // 调用聚合根业务方法，触发领域事件
        conference.RegisterAttendee(command.AttendeeEmail, command.AttendeeName, command.IdempotencyKey);
        
        // 保存聚合根的未提交事件到事件存储
        await _aggregateRepository.SaveAsync(conference, cancellationToken);
    }
}

using ENode.Domain;
 
[AggregateRoot]
public class ConferenceAggregate : AggregateRoot<Guid>
{
    private int _remainingSeats;
    private readonly HashSet<string> _registeredKeys = new(); // 存储邮箱和幂等键
 
    // 事件溯源重建聚合根时使用的无参构造函数
    public ConferenceAggregate() {}
 
    // 创建会议的构造函数
    public ConferenceAggregate(Guid id, string name, int totalSeats)
    {
        if (totalSeats <= 0)
            throw new ArgumentException("总座位数必须大于0", nameof(totalSeats));
            
        ApplyEvent(new ConferenceCreatedEvent(id, name, totalSeats));
    }
 
    // 注册参会者的业务方法
    public void RegisterAttendee(string email, string name, string idempotencyKey)
    {
        // 业务规则校验
        if (_registeredKeys.Contains(idempotencyKey))
            throw new BusinessException("该请求已处理，请勿重复提交");
        if (_remainingSeats <= 0)
            throw new BusinessException("会议座位已售罄");
        if (_registeredKeys.Contains(email))
            throw new BusinessException("该邮箱已注册参会");
 
        // 生成领域事件
        ApplyEvent(new AttendeeRegisteredEvent(
            Id, Guid.NewGuid(), email, name, idempotencyKey));
    }
 
    // 处理会议创建事件，更新聚合状态
    private void Apply(ConferenceCreatedEvent e)
    {
        Id = e.ConferenceId;
        _remainingSeats = e.TotalSeats;
    }
 
    // 处理参会者注册事件，更新聚合状态
    private void Apply(AttendeeRegisteredEvent e)
    {
        _registeredKeys.Add(e.Email);
        _registeredKeys.Add(e.IdempotencyKey);
        _remainingSeats--;
    }
}

using Dapper;
using ENode.Eventing;
using System.Data;
 
public class AttendeeProjection : IEventHandler<AttendeeRegisteredEvent>
{
    private readonly IDbConnectionFactory _dbConnectionFactory;
 
    public AttendeeProjection(IDbConnectionFactory dbConnectionFactory)
    {
        _dbConnectionFactory = dbConnectionFactory;
    }
 
    public async Task HandleAsync(AttendeeRegisteredEvent e, CancellationToken cancellationToken)
    {
        using var connection = await _dbConnectionFactory.OpenAsync(cancellationToken);
        await connection.ExecuteAsync(
            @"INSERT INTO AttendeeReadModel (Id, ConferenceId, Email, Name, RegisteredTime) 
              VALUES (@Id, @ConferenceId, @Email, @Name, @RegisteredTime)",
            new 
            { 
                Id = e.AttendeeId, 
                ConferenceId = e.ConferenceId, 
                Email = e.Email, 
                Name = e.Name, 
                RegisteredTime = DateTime.UtcNow 
            },
            cancellationToken: cancellationToken);
    }
}