MASA Framework 事件总线 - 跨进程事件总线

dotnet new web -o Assignment.IntegrationEventBuscd Assignment.IntegrationEventBus
dotnet add package Masa.Contrib.Dispatcher.IntegrationEvents.Dapr --version 0.7.0-preview.8 // 使用dapr提供的pubsub能力dotnet add package Masa.Contrib.Dispatcher.IntegrationEvents.EventLogs.EFCore --version 0.7.0-preview.8 //本地消息表dotnet add package Masa.Contrib.Data.EFCore.Sqlite --version 0.7.0-preview.8 //使用EfCore.Sqlitedotnet add package Masa.Contrib.Data.UoW.EFCore --version 0.7.0-preview.8 //使用工作单元dotnet add package Masa.Contrib.Development.DaprStarter.AspNetCore --version 0.7.0-preview.8 //开发环境使用DaprStarter协助管理Dapr Sidecardotnet add package Microsoft.EntityFrameworkCore.Design --version 6.0.6 //方便后续通过CodeFirst迁移数据库

public class UserDbContext : MasaDbContext{    public UserDbContext(MasaDbContextOptions<UserDbContext> options) : base(options)    {    }}

if (builder.Environment.IsDevelopment()){    builder.Services.AddDaprStarter();}

builder.Services.AddIntegrationEventBus(option =>{    option.UseDapr()        .UseEventLog<UserDbContext>()        .UseUoW<UserDbContext>(optionBuilder => optionBuilder.UseSqlite($"Data Source=./Db/{Guid.NewGuid():N}.db;"));});var app = builder.Build();
#region dapr 订阅集成事件使用app.UseRouting();
app.UseCloudEvents();app.UseEndpoints(endpoints =>{    endpoints.MapSubscribeHandler();});#endregion

public record RegisterUserEvent : IntegrationEvent{    public override string Topic { get; set; } = nameof(RegisterUserEvent);
    public string Account { get; set; }
    public string Mobile { get; set; }}

dotnet ef migrations add init //创建迁移dotnet ef database update //更新数据库

app.MapPost("/register", async (IIntegrationEventBus eventBus) =>{    //todo: 模拟注册用户并发布注册用户事件    await eventBus.PublishAsync(new RegisterUserEvent()    {        Account = "Tom",        Mobile = "19999999999"    });});

app.MapPost("/IntegrationEvent/RegisterUser", [Topic("pubsub", nameof(RegisterUserEvent))](RegisterUserEvent @event) =>{    Console.WriteLine($"注册用户成功: {@event.Account}");});

dotnet add package DotNetCore.CAP

public class IntegrationEventBus : IIntegrationEventBus{    private readonly ICapPublisher _publisher;    private readonly ICapTransaction _capTransaction;    private readonly IUnitOfWork? _unitOfWork;    public IntegrationEventBus(ICapPublisher publisher, ICapTransaction capTransaction, IUnitOfWork? unitOfWork = null)    {        _publisher = publisher;        _capTransaction = capTransaction;        _unitOfWork = unitOfWork;    }        public Task PublishAsync<TEvent>(TEvent @event) where TEvent : IEvent    {        // 如果使用事务        // _publisher.Transaction.Value.DbTransaction = unitOfWork.Transaction;        // _publisher.Publish(@event.Topic, @event);        throw new NotImplementedException();    }
    public IEnumerable<Type> GetAllEventTypes()    {        throw new NotImplementedException();    }
    public Task CommitAsync(CancellationToken cancellationToken = default)    {        throw new NotImplementedException();    }}

public static class ServiceCollectionExtensions{    public static DispatcherOptions UseRabbitMq(this IServiceCollection services)    {         //todo: 注册RabbitMq信息         services.TryAddScoped<IIntegrationEventBus, IntegrationEventBus>();         return dispatcherOptions;    }}

dotnet add package RabbitMQ.Client //使用RabbitMq

public class Publisher : IPublisher{    public async Task PublishAsync<T>(string topicName, T @event, CancellationToken stoppingToken = default) where T : IIntegrationEvent    {        //todo: 通过 RabbitMQ.Client 发送消息到RabbitMq        throw new NotImplementedException();    }}

public static class DispatcherOptionsExtensions{    public static DispatcherOptions UseRabbitMq(this Masa.Contrib.Dispatcher.IntegrationEvents.Options.DispatcherOptions options)    {         //todo: 注册RabbitMq信息         dispatcherOptions.Services.TryAddSingleton<IPublisher, Publisher>();         return dispatcherOptions;    }}

builder.Services.AddIntegrationEventBus(option =>{    option.UseRabbitMq();//修改为使用RabbitMq    option.UseUoW<UserDbContext>(optionBuilder => optionBuilder.UseSqlite($"Data Source=./Db/{Guid.NewGuid():N}.db;"));    option.UseEventLog<UserDbContext>();});