Merq 2.0.0

Mercury: 罗马神话中的使者

Mercury > Merq-ry > Merq

Merq 将 消息总线 模式与 面向命令的接口 结合起来，用于可扩展并解耦的进程内应用程序架构。

这些模式在微服务和面向服务的架构中已经很好地建立，但其益处也可以应用于应用程序，尤其是可扩展的应用程序，其中多个团队可以贡献扩展，这些扩展在运行时进行组合。

组件之间经过优化解耦，使得它们可以更独立地发展，同时提高了可发现现有命令和事件的能力。您可以在VSCode命令和<弓 href="https://vscode.js.cn/api/references/vscode-api#window" rel="noopener noreferrer nofollow">窗口事件等方面看到这种方法的实际应用。显然，在VSCode的情况下，一切都是进程内的，但干净、可预见的API带来的好处是显而易见的。

Merq为.NET应用程序提供了相同的特性。

事件

事件可以是任何类型，没有限制或必须实现的接口。如今，C#记录类型非常适合作为事件数据类型。一个示例事件可以是以下一行：

public record ItemShipped(string Id, DateTimeOffset Date);

消息总线上的基于事件的API界面足够简单

public interface IMessageBus
{
    void Notify<TEvent>(TEvent e);
    IObservable<TEvent> Observe<TEvent>();
}

通过依赖IObservable<TEvent>，Merq可以无缝地集成通过System.Reactive或更轻量级的RxFree实现更强的基于事件的处理。使用这些包中的任何一个订阅事件都是微不足道的。

IDisposable subscription;

// constructor may use DI to get the dependency
public CustomerViewModel(IMessageBus bus)
{
    subscription = bus.Observe<ItemShipped>().Subscribe(OnItemShipped);
}

void OnItemShipped(ItemShipped e) => // Refresh item status

public void Dispose() => subscription.Dispose();

除了事件生产者只是调用Notify之外，它们还可以直接作为IObservable<TEvent>实现，这对生产者是观察序列本身非常有用。

这两个特性都可以无缝集成，并充分利用Reactive Extensions的全部功能。

命令

命令也可以是任何类型，而C#记录使得定义变得简洁。

record CancelOrder(string OrderId) : IAsyncCommand;

与事件不同，命令消息需要表明它们在执行时需要的调用样式。

前面的示例命令可以使用以下代码执行

// perhaps a method invoked when a user 
// clicks/taps a Cancel button next to an order
async Task OnCancel(string orderId)
{
    await bus.ExecuteAsync(new CancelOrder(orderId), CancellationToken.None);
    // refresh UI for new state.
}

同步命令的示例可以是

// Command declaration
record SignOut() : ICommand;

// Command invocation
void OnSignOut() => bus.Execute(new SignOut());

// or alternatively, for void commands that have no additional data:
void OnSignOut() => bus.Execute<SignOut>();

命令消息上的标记接口驱使编译器仅允许在消息总线上的正确调用样式，如命令作者所定义

public interface IMessageBus
{
    // sync void
    void Execute(ICommand command);
    // sync value-returning
    TResult Execute<TResult>(ICommand<TResult> command);
    // async void
    Task ExecuteAsync(IAsyncCommand command, CancellationToken cancellation);
    // async value-returning
    Task<TResult> ExecuteAsync<TResult>(IAsyncCommand<TResult> command, CancellationToken cancellation);
    // async stream
    IAsyncEnumerable<TResult> ExecuteStream<TResult>(IStreamCommand<TResult> command, CancellationToken cancellation);
}

例如，要创建一个返回值的异步命令以检索某些值，则需要

record FindDocuments(string Filter) : IAsyncCommand<IEnumerable<string>>;

class FindDocumentsHandler : IAsyncCommandHandler<FindDocument, IEnumerable<string>>
{
    public bool CanExecute(FindDocument command) => !string.IsNullOrEmpty(command.Filter);
    
    public Task<IEnumerable<string>> ExecuteAsync(FindDocument command, CancellationToken cancellation)
        => // evaluate command.Filter across all documents and return matches
}

为了执行此类命令，编译器允许的唯一执行方法是

IEnumerable<string> files = await bus.ExecuteAsync(new FindDocuments("*.json"));

如果消费者试图使用Execute，编译器将报错称命令没有实现ICommand<TResult>，这将是标记接口的同步版本。

虽然这些命令消息上的标记接口可能看似不必要，但实际上它们非常重要。它们解决了执行抽象的关键问题：命令执行是同步还是异步（以及void或返回值）不应被抽象，否则您可能会陷入两种常见的反模式，即ASP.NET的异步指南中所称的同步覆盖异步和异步覆盖同步。

同样，在实现处理器时不会犯错误，因为处理器接口定义了对命令必须实施的限制。

// sync
public interface ICommandHandler<in TCommand> : ... where TCommand : ICommand;
public interface ICommandHandler<in TCommand, out TResult> : ... where TCommand : ICommand<TResult>;

// async
public interface IAsyncCommandHandler<in TCommand> : ... where TCommand : IAsyncCommand;
public interface IAsyncCommandHandler<in TCommand, TResult> : ... where TCommand : IAsyncCommand<TResult>

// async stream
public interface IStreamCommandHandler<in TCommand, out TResult>: ... where TCommand : IStreamCommand<TResult>

这种设计选择还使得不可能以错误的方式执行命令实现。

除了执行外，IMessageBus 还提供了一个机制，通过 CanHandle<T> 方法来判断一个命令是否已注册了处理程序，以及通过 CanExecute<T> 提供的验证机制，如上 FindDocumentsHandler 示例中所示。

命令可以通知新的事件，事件观察者/订阅者随后可以执行命令。

异步流

对于 .NET6+ 应用，Merq 还支持将 异步流 作为命令调用方式。这在命令执行产生大量结果，而消费者希望边生成边处理它们的场景下非常有用。

例如，上面的过滤文档命令可以实现为异步流命令

record FindDocuments(string Filter) : IStreamCommand<string>;

class FindDocumentsHandler : IStreamCommandHandler<FindDocument, string>
{
    public bool CanExecute(FindDocument command) => !string.IsNullOrEmpty(command.Filter);
    
    public async IAsyncEnumerable<string> ExecuteAsync(FindDocument command, [EnumeratorCancellation] CancellationToken cancellation)
    {
        await foreach (var file in FindFilesAsync(command.Filter, cancellation))
            yield return file;
    }
}

为了执行此类命令，编译器允许的唯一执行方法是

await foreach (var file in bus.ExecuteStream(new FindDocuments("*.json")))
    Console.WriteLine(file);

分析器和代码修复

除了编译器的抱怨外，Merq 还提供了一组分析器和代码修复来学习模式和避免常见错误。例如，如果你创建了一个简单的记录作为命令使用，如下所示：

public record Echo(string Message);

然后尝试为它实现命令处理程序

public class EchoHandler : ICommandHandler<Echo>
{
}

编译器会立即抱怨由于对 Echo 类型本身的依赖，导致各种约束和接口没有得到满足。对于经验丰富的 Merq 开发者来说，这是不言自明的，但对于新开发者来说，这可能会有些困惑。

在这种情况提供了自动实现所需接口的代码修复

同样，如果消费者尝试以异步方式调用上面的 Echo 命令（称为同步/异步反模式），他们将会遇到一个有些不明智的编译器错误。

但第二个错误更有帮助，因为它指向了真正的问题，并且可以应用代码修复来解决问题。

同样，对于 同步/异步 反模式的分析器，即同步命令异步执行，也提供了相同的分析器和代码修复。

赞助商

了解更多关于 GitHub Sponsors