From e11821842f22b1a9461ab2209c0be7cd47f5c25e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 27 Jan 2016 17:22:39 -0500 Subject: [PATCH] AMQP-566: Add String Version of CorrelationId JIRA: https://jira.spring.io/browse/AMQP-566 --- .../amqp/core/MessageProperties.java | 12 +++- .../DefaultMessagePropertiesConverter.java | 70 ++++++++++++++----- ...efaultMessagePropertiesConverterTests.java | 41 ++++++++++- src/reference/asciidoc/amqp.adoc | 32 ++++++++- src/reference/asciidoc/whats-new.adoc | 5 ++ 5 files changed, 135 insertions(+), 25 deletions(-) diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java b/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java index 7ae2aaf9..a1a828b9 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -71,6 +71,8 @@ public class MessageProperties implements Serializable { private volatile byte[] correlationId; + private volatile String correlationIdString; + private volatile String replyTo; private volatile String contentType = DEFAULT_CONTENT_TYPE; @@ -181,6 +183,14 @@ public class MessageProperties implements Serializable { return this.correlationId;//NOSONAR } + public String getCorrelationIdString() { + return correlationIdString; + } + + public void setCorrelationIdString(String correlationIdString) { + this.correlationIdString = correlationIdString; + } + public void setReplyTo(String replyTo) { this.replyTo = replyTo; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java index 5a30e7ab..f9181b78 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -26,6 +26,7 @@ import org.springframework.amqp.AmqpUnsupportedEncodingException; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; import org.springframework.util.CollectionUtils; +import org.springframework.util.StringUtils; import com.rabbitmq.client.AMQP.BasicProperties; import com.rabbitmq.client.Envelope; @@ -41,10 +42,15 @@ import com.rabbitmq.client.LongString; */ public class DefaultMessagePropertiesConverter implements MessagePropertiesConverter { + public enum CorrelationIdPolicy { + STRING, BYTES, BOTH + } + private static final int DEFAULT_LONG_STRING_LIMIT = 1024; private final int longStringLimit; + private volatile CorrelationIdPolicy correlationIdPolicy = CorrelationIdPolicy.BYTES; /** * Construct an instance where {@link LongString}s will be returned as a @@ -64,6 +70,18 @@ public class DefaultMessagePropertiesConverter implements MessagePropertiesConve this.longStringLimit = longStringLimit; } + /** + * For inbound, determine whether correlationId, correlationIdString or + * both are populated. For outbound, determine whether correlationIdString + * or correlationId is used when mapping; if {@code CorrelationIdPolicy.BOTH} + * is set for outbound, String takes priority and we fallback to bytes. + * Default {@code CorrelationIdPolicy.BYTES}. + * @param correlationIPolicy true to use. + */ + public void setCorrelationIdAsString(CorrelationIdPolicy correlationIPolicy) { + this.correlationIdPolicy = correlationIPolicy; + } + public MessageProperties toMessageProperties(final BasicProperties source, final Envelope envelope, final String charset) { MessageProperties target = new MessageProperties(); @@ -88,11 +106,17 @@ public class DefaultMessagePropertiesConverter implements MessagePropertiesConve target.setContentType(source.getContentType()); target.setContentEncoding(source.getContentEncoding()); String correlationId = source.getCorrelationId(); - if (correlationId != null) { - try { - target.setCorrelationId(source.getCorrelationId().getBytes(charset)); - } catch (UnsupportedEncodingException ex) { - throw new AmqpUnsupportedEncodingException(ex); + if (!CorrelationIdPolicy.BYTES.equals(this.correlationIdPolicy) && correlationId != null) { + target.setCorrelationIdString(correlationId); + } + if (!CorrelationIdPolicy.STRING.equals(this.correlationIdPolicy)) { + if (correlationId != null) { + try { + target.setCorrelationId(source.getCorrelationId().getBytes(charset)); + } + catch (UnsupportedEncodingException ex) { + throw new AmqpUnsupportedEncodingException(ex); + } } } String replyTo = source.getReplyTo(); @@ -110,26 +134,34 @@ public class DefaultMessagePropertiesConverter implements MessagePropertiesConve public BasicProperties fromMessageProperties(final MessageProperties source, final String charset) { BasicProperties.Builder target = new BasicProperties.Builder(); - target.headers(this.convertHeadersIfNecessary(source.getHeaders())); - target.timestamp(source.getTimestamp()); - target.messageId(source.getMessageId()); - target.userId(source.getUserId()); - target.appId(source.getAppId()); - target.clusterId(source.getClusterId()); - target.type(source.getType()); + target.headers(this.convertHeadersIfNecessary(source.getHeaders())) + .timestamp(source.getTimestamp()) + .messageId(source.getMessageId()) + .userId(source.getUserId()) + .appId(source.getAppId()) + .clusterId(source.getClusterId()) + .type(source.getType()); MessageDeliveryMode deliveryMode = source.getDeliveryMode(); if (deliveryMode != null) { target.deliveryMode(MessageDeliveryMode.toInt(deliveryMode)); } - target.expiration(source.getExpiration()); - target.priority(source.getPriority()); - target.contentType(source.getContentType()); - target.contentEncoding(source.getContentEncoding()); + target.expiration(source.getExpiration()) + .priority(source.getPriority()) + .contentType(source.getContentType()) + .contentEncoding(source.getContentEncoding()); byte[] correlationId = source.getCorrelationId(); - if (correlationId != null && correlationId.length > 0) { + String correlationIdString = source.getCorrelationIdString(); + if (!CorrelationIdPolicy.BYTES.equals(this.correlationIdPolicy) + && StringUtils.hasText(correlationIdString)) { + target.correlationId(correlationIdString); + correlationId = null; + } + if (!CorrelationIdPolicy.STRING.equals(this.correlationIdPolicy) + && correlationId != null && correlationId.length > 0) { try { target.correlationId(new String(correlationId, charset)); - } catch (UnsupportedEncodingException ex) { + } + catch (UnsupportedEncodingException ex) { throw new AmqpUnsupportedEncodingException(ex); } } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverterTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverterTests.java index c01a1161..63ac2b54 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverterTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at @@ -13,6 +13,7 @@ package org.springframework.amqp.rabbit.support; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -36,11 +37,13 @@ import com.rabbitmq.client.impl.LongStringHelper; /** * @author Soeren Unruh + * @author Gary Russell * @since 1.3 */ public class DefaultMessagePropertiesConverterTests { - private final MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter(); + private final DefaultMessagePropertiesConverter messagePropertiesConverter = + new DefaultMessagePropertiesConverter(); private final Envelope envelope = new Envelope(0, false, null, null); @@ -167,4 +170,38 @@ public class DefaultMessagePropertiesConverterTests { ((Map) basicProps.getHeaders().get("map")).get("unsupported") instanceof String); } + @Test + public void testCorrelationIdAsString() { + MessageProperties messageProperties = new MessageProperties(); + this.messagePropertiesConverter + .setCorrelationIdAsString(DefaultMessagePropertiesConverter.CorrelationIdPolicy.BOTH); + messageProperties.setCorrelationIdString("foo"); + messageProperties.setCorrelationId("bar".getBytes()); // foo should win + BasicProperties basicProps = this.messagePropertiesConverter.fromMessageProperties(messageProperties, "UTF-8"); + assertEquals("foo", basicProps.getCorrelationId()); + messageProperties = this.messagePropertiesConverter.toMessageProperties(basicProps, null, "UTF-8"); + assertEquals("foo", messageProperties.getCorrelationIdString()); + assertEquals("foo", new String(messageProperties.getCorrelationId())); + + this.messagePropertiesConverter + .setCorrelationIdAsString(DefaultMessagePropertiesConverter.CorrelationIdPolicy.STRING); + messageProperties.setCorrelationIdString("foo"); + messageProperties.setCorrelationId("bar".getBytes()); // foo should win + basicProps = this.messagePropertiesConverter.fromMessageProperties(messageProperties, "UTF-8"); + assertEquals("foo", basicProps.getCorrelationId()); + messageProperties = this.messagePropertiesConverter.toMessageProperties(basicProps, null, "UTF-8"); + assertEquals("foo", messageProperties.getCorrelationIdString()); + assertNull(messageProperties.getCorrelationId()); + + this.messagePropertiesConverter + .setCorrelationIdAsString(DefaultMessagePropertiesConverter.CorrelationIdPolicy.BYTES); + messageProperties.setCorrelationIdString("foo"); + messageProperties.setCorrelationId("bar".getBytes()); // bar should win + basicProps = this.messagePropertiesConverter.fromMessageProperties(messageProperties, "UTF-8"); + assertEquals("bar", basicProps.getCorrelationId()); + messageProperties = this.messagePropertiesConverter.toMessageProperties(basicProps, null, "UTF-8"); + assertNull(messageProperties.getCorrelationIdString()); + assertEquals("bar", new String(messageProperties.getCorrelationId())); + } + } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index dbfa3671..f119ad11 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -1884,6 +1884,30 @@ The default properties converter will convert `BasicProperties` elements of type when the size is not greater than `1024` bytes. Larger `LongString` s are returned as a `DataInputStream. This limit can be overridden with a constructor argument. +Starting with _version 1.6_, a new property `correlationIdString` has been added to `MessageProperties`. +Previously, when converting to/from `BasicProperties` used by the RabbitMQ client, an unnecessary `byte[] <-> String` +conversion was performed because `MessageProperties.correlationId` is a `byte[]` but `BasicProperties` uses a +`String`. (Utimately, the RabbitMQ client uses UTF-8 to convert the String to bytes to put in the protocol message). + +To provide maximum backwards compatibility, a new property `correlationIdPolicy` has been adde to the +`DefaultMessagePropertiesConverter`. +This takes an `DefaultMessagePropertiesConverter.CorrelationIdPolicy` enum argument. +By default it is set to `BYTES` which replicates the previous behavior. + +For inbound messages: + +- `STRING` - just the `correlationIdString` property is mapped +- `BYTES` - just the `correlationId` property is mapped +- `BOTH` - both properties are mapped + +For outbound messages: + +- `STRING` - just the `correlationIdString` property is mapped +- `BYTES` - just the `correlationId` property is mapped +- `BOTH` - Both properties will be considered, with the String property taking precedence + + + [[post-processing]] ==== Modifying Messages - Compression and More @@ -2291,7 +2315,7 @@ This functionality can only be used programmatically by invoking the `RabbitAdmi It is not supported for auto-declaration by the admin by defining a queue declaratively in the application context. This is in contrast to an `AnonymousQueue` where the framework generates a unique (`UUID`) name and sets `durable` to -`false` and `exlusive`, `autoDelete` to `true`. +`false` and `exclusive`, `autoDelete` to `true`. A `` with an empty, or missing, `name` attribute will always create an `AnonymousQueue`. See <> to understand why `AnonymousQueue` is preferred over broker-generated queue names, as well as @@ -2432,8 +2456,10 @@ Starting with _version 1.3_ the HeadersExchange can be configured to match on mu ---- -To see how to use Java to configure the AMQP infrastructure, look at the Stock sample application, where there is the `@Configuration` class `AbstractStockRabbitConfiguration` which in turn has RabbitClientConfiguration and RabbitServerConfiguration subclasses. -The code for AbstractStockRabbitConfiguration is shown below +To see how to use Java to configure the AMQP infrastructure, look at the Stock sample application, +where there is the `@Configuration` class `AbstractStockRabbitConfiguration` which in turn has +`RabbitClientConfiguration` and `RabbitServerConfiguration` subclasses. +The code for `AbstractStockRabbitConfiguration` is shown below [source,java] ---- diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index aeea17a0..f8715283 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -62,6 +62,11 @@ This version allows you to override this default behavior and use a temporary qu `useTemporaryReplyQueues` property to `true`. See <> for more information. +===== Message Properties and CorrelationId + +The `correlationId` message property can now be a `String`. +See <> for more information. + ==== Changes in 1.5 Since 1.4 ===== spring-erlang is No Longer Supported