在运行时为 Rebus 生成消息处理程序答案

【问题标题】：Generating message handlers for Rebus in runtime在运行时为 Rebus 生成消息处理程序
【发布时间】：2020-09-13 19:34:40
【问题描述】：

我遇到了一个问题，我想使用 Rebus 来订阅和处理rabbitmq 中的消息。在第三方程序集中定义了多种消息类型，并且会定期将新消息类型添加到该程序集中。

我需要以某种方式让 Rebus 订阅并处理所有这些消息类型并将它们转发（发布）到另一个 rabbitmq 实例。我的服务本质上是转发消息，并在这样做时添加自定义 rebus 标头。

问题是我不想为每种消息类型生成处理程序类（因为无论消息类型如何，功能都是相同的）。我也不想每次在第三方程序集中添加新的消息类型时都更新我的代码（编写新的处理程序类）。

我尝试使用 TypeBuilder 为通过反射找到的每种类型动态创建消息处理程序类，但感觉有点乱，所以我希望有另一种方法？

下面的代码概述了我希望实现的目标，即使代码无法编译。

public void SubscribeAndHandleMessages()
        {
            // These types will be determined runtime by using reflection but thats omitted for clarity
            var messageTypes = new List<Type>(){typeof(MessageA), typeof(MessageB)}; 

            var activator = new BuiltinHandlerActivator();

            Configure.With(activator)
                .Transport(t => t.UseRabbitMq(_rabbitConnectionString, "MyQueue"))
                .Start();

            //Subscribe and register handlers
            foreach (var type in messageTypes)
            {
                activator.Bus.Subscribe(type); //This works, I can see the queue subscribing to the correct topics
                activator.Handle<type>(async (bus, context, message) => //This doesnt work since type is not known at compile time
                {
                    //Forwarding to another rabbit instance, same handling for all types of messages
                });
            }
        }

【问题讨论】：

标签： c# rabbitmq rebus

【解决方案1】：

一旦您建立了必要的订阅，您只需要能够处理您收到的各种消息。

使用 Rebus 的最佳方法是避免使用普通的消息处理管道（反序列化 => 查找处理程序 => 分派），而是以原始形式（即“传输消息”形式）处理消息.

您可以使用 Rebus 的传输消息转发功能来做到这一点。有了它，一个 100% 通用的消息处理程序可能看起来像这样：

Configure.With(activator)
    .Transport(t => t.UseInMemoryTransport(new InMemNetwork(), "router-tjek"))
    .Routing(r => r.AddTransportMessageForwarder(async transportMessage =>
    {
        var headers = transportMessage.Headers; //< Dictionary<string, string>
        var body = transportMessage.Body;       //< byte[]

        // handle the message here, e.g.
        // by deserializing the body into a JObject,
        // storing the bytes in a database, or by
        // forwarding the message to another queue
        return // appropriate forward action here
    }))
    .Start();

您可以在此处阅读更多信息：Transport message forwarding

【讨论】：