Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-566: Add String Version of CorrelationId

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-566
This commit is contained in:
Gary Russell
2016-01-27 17:22:39 -05:00
committed by Artem Bilan
parent 59710724e5
commit e11821842f
5 changed files with 135 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-amqp/src/main/java/org/springframework/amqp/core/MessageProperties.java
View File

@@ -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;
}

70
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverter.java
View File

@@ -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);
}
}

41
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/support/DefaultMessagePropertiesConverterTests.java
View File

@@ -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<String, Object>) 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()));
}
}

32
src/reference/asciidoc/amqp.adoc
View File

@@ -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 `<rabbit:queue/>` with an empty, or missing, `name` attribute will always create an `AnonymousQueue`.
See <<anonymous-queue>> 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
</rabbit:headers-exchange>
----
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]
----

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -62,6 +62,11 @@ This version allows you to override this default behavior and use a temporary qu
`useTemporaryReplyQueues` property to `true`.
See <<direct-reply-to>> for more information.
===== Message Properties and CorrelationId
The `correlationId` message property can now be a `String`.
See <<message-properties-converters>> for more information.
==== Changes in 1.5 Since 1.4
===== spring-erlang is No Longer Supported