Sending messages to an Apache Camel route using the RabbitMQ Provider

Architecture & Design
1

I work with .NET and Camel Quarkus.
For the first time I need to send a message via fanout to some camel routes.

This is the setup in .NET

 var cnnStr = config.ConnectionString;
 var transport = epc.UseTransport<RabbitMQTransport>().Transactions(TransportTransactionMode.ReceiveOnly)
     .ConnectionString(cnnStr).UseConventionalRoutingTopology();
      epc.UseSerialization<NewtonsoftSerializer>();
        
 // any name will do, as long there are a queue attached to it
 transport.Routing().RouteToEndpoint(typeof(ProcessCertificateTrades), "tradingservice.certificate.trades.out.fx");

This is the queue configuration in Camel

rabbitmq:tradingservice.certificate.trades.out.fx?queue=certificate.trade.aladdin.q&autoAck=false&autoDelete=false&skipExchangeDeclare=true&skipDlqDeclare=true&reQueue=true&prefetchCount=1&prefetchEnabled=true&prefetchGlobal=false

The Camel route fails on the reply_to header

image
image1121×285 21.4 KB

Can I remove the reply_to header? and why is it there.
Maybe someone in this forum knows how to make Camel ignore the reply_to header

2

The NServiceBus.ReplyToAddress header is used for allowing downstream handlers to know where to send message replies to.

If you’re not using context.Reply or sagas, then I think it should be safe to remove.

It looks like Camel is trying to convert the NServiceBus.ReplyToAddress to it’s own understanding of a reply_to header. The NServiceBus.ReplyToAddress header is a very specific NServiceBus concept, so I don’t think it’s wise to try and shoehorn it into a Camel header.

Is there a way to exclude that header from what looks like Camel’s mapping so that the header still remains in the message but isn’t inspected by Camel at all?

Here are all the NServiceBus headers: Message Headers • NServiceBus • Particular Docs

3

Thanks. Ended up removing the header for that specific message type

Code for removing the header

public class MutateOutgoingTransportMessages : IMutateOutgoingTransportMessages
{
    public Task MutateOutgoing(MutateOutgoingTransportMessageContext context)
    {
        var msg = context.OutgoingMessage;
        if(msg is not ProcessCertificateTrades)
            return Task.CompletedTask;
        
        if(context.OutgoingHeaders.ContainsKey("NServiceBus.ReplyToAddress"))
            context.OutgoingHeaders.Remove("NServiceBus.ReplyToAddress");
        
        return Task.CompletedTask;
    }
}