Publish events with MassTransit

Question

I'm trying to publish message in one microservice and get it in another one, but cannot implement this using MassTransit 5.5.3 with RabbitMQ.

As far as I know we don't have to create a ReceiveEndpoint to be able to publish event so I'm just creating the same message interface in both services and publish a message, but as I can see in RabbitMQ it either goes to nowhere (if doesn't mapped to a queue) or goes to "_skipped" queue.

Publisher:

namespace Publisher
{
    class Program
    {
        static async Task Main(string[] args)
        {
            var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                IRabbitMqHost host = cfg.Host("host", "vhost", h =>
                {
                    h.Username("xxx");
                    h.Password("yyy");
                });
            });

            bus.Start();

            await bus.Publish<Message>(new { Text = "Hello World" });

            Console.ReadKey();

            bus.Stop();
        }
    }

    public interface Message
    {
        string Text { get; set; }
    }
}

Consumer:

namespace Consumer
{
    class Program
    {
        static async Task Main(string[] args)
        {
            var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                IRabbitMqHost host = cfg.Host("host", "vhost", h =>
                {
                    h.Username("xxx");
                    h.Password("yyy");
                });

                cfg.ReceiveEndpoint(host, e =>
                {
                    e.Consumer<MbConsumer>();
                });
            });

            bus.Start();

            bool finish = false;

            while(!finish)
            {
                await Task.Delay(1000);
            }

            bus.Stop();
        }
    }

    public interface Message
    {
        string Text { get; set; }
    }

    public class MbConsumer : IConsumer<Message>
    {
        public async Task Consume(ConsumeContext<Message> context)
        {
            await Console.Out.WriteLineAsync(context.Message.Text);
        }
    }
}

I'm expeting the consumer to get the message once it's been published but it doesn't get it. I think this is because the full message types are different ("Publisher.Message" vs. "Consumer.Message") so message contract is different. How should I fix this code to get the event in the consumer? Looks like I'm missing some fundamental thing about RabbitMQ or MassTransit.

Alexey Zimarev Alexey Zimarev · Accepted Answer · 2019-07-19T13:49:54

Your guess is correct. MassTransit uses the fully qualified class name as the message contract name. MassTransit also uses type-based routing, so FQCNs are used to create exchanges and bindings.

So, if you move your message class to a separate namespace like:

namespace Messages
{
    public interface Message
    {
        string Text { get; set; }
    }
}

You then can reference this type when you publish a message

await bus.Publish<Messages.Message>(new { Text = "Hello World" });

and define your consumer

public class MbConsumer : IConsumer<Messages.Message>
{
    public async Task Consume(ConsumeContext<Message> context)
    {
        await Console.Out.WriteLineAsync(context.Message.Text);
    }
}

it will work.

You might want also to look at RMQ management UI to find out about MassTransit topology. With your code, you will see two exchanges, one Publisher.Message and another one Consumer.Message where your consumer queue is bound to the Consumer.Message exchange, but you publish messages to the Publisher.Message exchange and they just vanish.

I would also suggest specifying a meaningful endpoint name for your receive endpoint:

cfg.ReceiveEndpoint(host, "MyConsumer", e =>
{
    e.Consumer<MbConsumer>();
});

Publish events with MassTransit

1 Answers