rgannu commented on a change in pull request #4346:
URL: https://github.com/apache/camel/pull/4346#discussion_r498766080
##########
File path:
components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQMessageConverter.java
##########
@@ -49,158 +52,196 @@
public void mergeAmqpProperties(Exchange exchange, AMQP.BasicProperties
properties) {
if (properties.getType() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.TYPE,
properties.getType());
+ exchange.getIn().setHeader(RabbitMQConstants.TYPE.key(),
properties.getType());
}
if (properties.getAppId() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.APP_ID,
properties.getAppId());
+ exchange.getIn().setHeader(RabbitMQConstants.APP_ID.key(),
properties.getAppId());
}
if (properties.getClusterId() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.CLUSTERID,
properties.getClusterId());
+ exchange.getIn().setHeader(RabbitMQConstants.CLUSTERID.key(),
properties.getClusterId());
}
if (properties.getContentEncoding() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.CONTENT_ENCODING,
properties.getContentEncoding());
+
exchange.getIn().setHeader(RabbitMQConstants.CONTENT_ENCODING.key(),
properties.getContentEncoding());
}
if (properties.getContentType() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.CONTENT_TYPE,
properties.getContentType());
+ exchange.getIn().setHeader(RabbitMQConstants.CONTENT_TYPE.key(),
properties.getContentType());
}
if (properties.getCorrelationId() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.CORRELATIONID,
properties.getCorrelationId());
+ exchange.getIn().setHeader(RabbitMQConstants.CORRELATIONID.key(),
properties.getCorrelationId());
}
if (properties.getExpiration() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.EXPIRATION,
properties.getExpiration());
+ exchange.getIn().setHeader(RabbitMQConstants.EXPIRATION.key(),
properties.getExpiration());
}
if (properties.getMessageId() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.MESSAGE_ID,
properties.getMessageId());
+ exchange.getIn().setHeader(RabbitMQConstants.MESSAGE_ID.key(),
properties.getMessageId());
}
if (properties.getPriority() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.PRIORITY,
properties.getPriority());
+ exchange.getIn().setHeader(RabbitMQConstants.PRIORITY.key(),
properties.getPriority());
}
if (properties.getReplyTo() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.REPLY_TO,
properties.getReplyTo());
+ exchange.getIn().setHeader(RabbitMQConstants.REPLY_TO.key(),
properties.getReplyTo());
}
if (properties.getTimestamp() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.TIMESTAMP,
properties.getTimestamp());
+ exchange.getIn().setHeader(RabbitMQConstants.TIMESTAMP.key(),
properties.getTimestamp());
}
if (properties.getUserId() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.USERID,
properties.getUserId());
+ exchange.getIn().setHeader(RabbitMQConstants.USERID.key(),
properties.getUserId());
}
if (properties.getDeliveryMode() != null) {
- exchange.getIn().setHeader(RabbitMQConstants.DELIVERY_MODE,
properties.getDeliveryMode());
+ exchange.getIn().setHeader(RabbitMQConstants.DELIVERY_MODE.key(),
properties.getDeliveryMode());
}
}
public AMQP.BasicProperties.Builder buildProperties(Exchange exchange) {
- AMQP.BasicProperties.Builder properties = new
AMQP.BasicProperties.Builder();
-
Message msg;
if (exchange.hasOut()) {
msg = exchange.getOut();
} else {
msg = exchange.getIn();
}
- final Object contentType =
msg.removeHeader(RabbitMQConstants.CONTENT_TYPE);
+ AMQP.BasicProperties.Builder properties =
buildBasicAmqpProperties(exchange.getProperties(), msg);
+
+ final Map<String, Object> headers = msg.getHeaders();
+ // Add additional headers (if any)
+ if (additionalHeaders != null) {
+ headers.putAll(additionalHeaders);
+ }
+ Map<String, Object> filteredHeaders = new HashMap<>();
+
+ for (Map.Entry<String, Object> header : headers.entrySet()) {
+ // filter header values.
+ Object value = getValidRabbitMQHeaderValue(header.getKey(),
header.getValue());
+
+ // additionally filter out the OVERRIDE header so it does not
+ // propagate
+ if ((value != null || isAllowNullHeaders()) && !header.getKey()
+ .equals(RabbitMQConstants.EXCHANGE_OVERRIDE_NAME.key())) {
+ boolean filteredHeader;
+ if (!allowCustomHeaders) {
+ filteredHeader = headerFilterStrategy
+ .applyFilterToCamelHeaders(header.getKey(),
header.getValue(), exchange);
+ if (filteredHeader) {
+ filteredHeaders.put(header.getKey(),
header.getValue());
+ }
+ } else {
+ filteredHeaders.put(header.getKey(), header.getValue());
+ }
+ } else if (LOG.isDebugEnabled()) {
+ if (header.getValue() == null) {
+ LOG.debug("Ignoring header: {} with null value",
header.getKey());
+ } else if
(header.getKey().equals(RabbitMQConstants.EXCHANGE_OVERRIDE_NAME.key())) {
+ LOG.debug("Preventing header propagation: {} with value
{}:", header.getKey(),
+ header.getValue());
+ } else {
+ LOG.debug("Ignoring header: {} of class: {} with value:
{}", header.getKey(),
+
ObjectHelper.classCanonicalName(header.getValue()), header.getValue());
+ }
+ }
+ }
+
+ properties.headers(filteredHeaders);
+
+ return properties;
+ }
+
+ private AMQP.BasicProperties.Builder buildBasicAmqpProperties(
+ Map<String, Object> exchangeProperties, Message msg) {
+ AMQP.BasicProperties.Builder properties = new
AMQP.BasicProperties.Builder();
+
+ Object contentType = getBasicAmqpProperty(exchangeProperties, msg,
RabbitMQConstants.CONTENT_TYPE);
if (contentType != null) {
properties.contentType(contentType.toString());
}
- final Object priority = msg.removeHeader(RabbitMQConstants.PRIORITY);
Review comment:
Removed it accidentally. It should be there. Accepted.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
