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

Merge pull request #777 from garyrussell/INT-2970

Browse Source 
* garyrussell-INT-2970:
  INT-2970 Update to Spring-AMQP 1.2.0.M1
This commit is contained in:
Gunnar Hillert
2013-04-02 18:08:25 -04:00
parent 110957f975 4449595520
commit 93211942cf
5 changed files with 48 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
build.gradle
View File

@@ -47,7 +47,7 @@ subprojects { subproject ->
springVersionDefault = '3.1.3.RELEASE'
springVersion = project.hasProperty('springVersion') ? getProperty('springVersion') : springVersionDefault
springAmqpVersion = '1.1.4.RELEASE'
springAmqpVersion = '1.2.0.M1'
springDataMongoVersion = '1.1.1.RELEASE'
springDataRedisVersion = '1.0.2.RELEASE'
springGemfireVersion = '1.2.2.RELEASE'
@@ -157,6 +157,9 @@ project('spring-integration-amqp') {
testCompile project(":spring-integration-test")
testCompile project(":spring-integration-http") // need to test INT-2713
}
// suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac)
compileJava.options.compilerArgs = ["${xLintArg},-deprecation"]
}
project('spring-integration-core') {

18
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/AmqpHeaders.java
View File

@@ -22,7 +22,7 @@ import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
/**
* Pre-defined names and prefixes to be used for setting and/or retrieving AMQP
* MessageProperties from/to integration Message Headers.
*
*
* @author Mark Fisher
*/
public abstract class AmqpHeaders {
@@ -87,4 +87,20 @@ public abstract class AmqpHeaders {
public static final String RETURN_ROUTING_KEY = PREFIX + "returnRoutingKey";
/**
* Compatibility with Spring-AMQP 1.1
* This was previously in RabbitTemplate
* @deprecated Use the standard Rabbit CorrelationId header (mapped to {@link #CORRELATION_ID}).
*/
@Deprecated
public static final String STACKED_CORRELATION_HEADER = "spring_reply_correlation";
/**
* Compatibility with Spring-AMQP 1.1
* This was previously in RabbitTemplate
* @deprecated No longer used by Spring-AMQP 1.2 and above.
*/
@Deprecated
public static final String STACKED_REPLY_TO_HEADER = "spring_reply_to";
}

9
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
View File

@@ -24,7 +24,6 @@ import java.util.Map;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.mapping.AbstractHeaderMapper;
@@ -158,13 +157,13 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
if (StringUtils.hasText(userId)) {
headers.put(AmqpHeaders.USER_ID, userId);
}
Object replyCorrelation = amqpMessageProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER);
Object replyCorrelation = amqpMessageProperties.getHeaders().get(AmqpHeaders.STACKED_CORRELATION_HEADER);
if (replyCorrelation instanceof String) {
if (StringUtils.hasText((String) replyCorrelation)) {
headers.put(AmqpHeaders.SPRING_REPLY_CORRELATION, replyCorrelation);
}
}
Object replyToStack = amqpMessageProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER);
Object replyToStack = amqpMessageProperties.getHeaders().get(AmqpHeaders.STACKED_REPLY_TO_HEADER);
if (replyToStack instanceof String) {
if (StringUtils.hasText((String) replyToStack)) {
headers.put(AmqpHeaders.SPRING_REPLY_TO_STACK, replyToStack);
@@ -185,8 +184,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
@Override
protected Map<String, Object> extractUserDefinedHeaders(MessageProperties amqpMessageProperties) {
Map<String, Object> headers = amqpMessageProperties.getHeaders();
headers.remove(RabbitTemplate.STACKED_CORRELATION_HEADER);
headers.remove(RabbitTemplate.STACKED_REPLY_TO_HEADER);
headers.remove(AmqpHeaders.STACKED_CORRELATION_HEADER);
headers.remove(AmqpHeaders.STACKED_REPLY_TO_HEADER);
return headers;
}

30
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java
View File

@@ -25,6 +25,17 @@ import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionListener;
import com.rabbitmq.client.AMQP.Basic.RecoverOk;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.Channel.FlowOk;
import com.rabbitmq.client.AMQP.Confirm.SelectOk;
import com.rabbitmq.client.AMQP.Exchange.BindOk;
import com.rabbitmq.client.AMQP.Exchange.DeclareOk;
import com.rabbitmq.client.AMQP.Exchange.DeleteOk;
import com.rabbitmq.client.AMQP.Exchange.UnbindOk;
import com.rabbitmq.client.AMQP.Queue.PurgeOk;
import com.rabbitmq.client.AMQP.Tx.CommitOk;
import com.rabbitmq.client.AMQP.Tx.RollbackOk;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Command;
import com.rabbitmq.client.ConfirmListener;
@@ -35,17 +46,6 @@ import com.rabbitmq.client.Method;
import com.rabbitmq.client.ReturnListener;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AMQP.Basic.RecoverOk;
import com.rabbitmq.client.AMQP.Channel.FlowOk;
import com.rabbitmq.client.AMQP.Confirm.SelectOk;
import com.rabbitmq.client.AMQP.Exchange.BindOk;
import com.rabbitmq.client.AMQP.Exchange.DeclareOk;
import com.rabbitmq.client.AMQP.Exchange.DeleteOk;
import com.rabbitmq.client.AMQP.Exchange.UnbindOk;
import com.rabbitmq.client.AMQP.Queue.PurgeOk;
import com.rabbitmq.client.AMQP.Tx.CommitOk;
import com.rabbitmq.client.AMQP.Tx.RollbackOk;
/**
* @author Mark Fisher
@@ -87,7 +87,7 @@ public class StubRabbitConnectionFactory implements ConnectionFactory {
return false;
}
}
private static class StubChannel implements Channel {
public void addShutdownListener(ShutdownListener listener) {
@@ -334,7 +334,7 @@ public class StubRabbitConnectionFactory implements ConnectionFactory {
public RecoverOk basicRecover(boolean requeue) throws IOException {
return null;
}
@Deprecated
public void basicRecoverAsync(boolean requeue) throws IOException {
}
@@ -409,6 +409,10 @@ public class StubRabbitConnectionFactory implements ConnectionFactory {
public void waitForConfirmsOrDie(long timeout) throws IOException,
InterruptedException, TimeoutException {
}
public void basicPublish(String arg0, String arg1, boolean arg2, BasicProperties arg3, byte[] arg4)
throws IOException {
}
}
}

12
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
View File

@@ -25,11 +25,9 @@ import java.util.Map;
import java.util.Set;
import org.junit.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.JsonMessageConverter;
import org.springframework.http.MediaType;
import org.springframework.integration.MessageHeaders;
@@ -43,6 +41,7 @@ import org.springframework.integration.amqp.AmqpHeaders;
*/
public class DefaultAmqpHeaderMapperTests {
@SuppressWarnings("deprecation")
@Test
public void fromHeaders() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
@@ -94,8 +93,8 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals(testTimestamp, amqpProperties.getTimestamp());
assertEquals("test.type", amqpProperties.getType());
assertEquals("test.userId", amqpProperties.getUserId());
assertEquals("test.correlation", amqpProperties.getHeaders().get(RabbitTemplate.STACKED_CORRELATION_HEADER));
assertEquals("test.replyTo2", amqpProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER));
assertEquals("test.correlation", amqpProperties.getHeaders().get(AmqpHeaders.STACKED_CORRELATION_HEADER));
assertEquals("test.replyTo2", amqpProperties.getHeaders().get(AmqpHeaders.STACKED_REPLY_TO_HEADER));
}
@Test
@@ -113,6 +112,7 @@ public class DefaultAmqpHeaderMapperTests {
assertEquals("text/html", amqpProperties.getContentType());
}
@SuppressWarnings("deprecation")
@Test
public void toHeaders() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
@@ -138,8 +138,8 @@ public class DefaultAmqpHeaderMapperTests {
amqpProperties.setTimestamp(testTimestamp);
amqpProperties.setType("test.type");
amqpProperties.setUserId("test.userId");
amqpProperties.setHeader(RabbitTemplate.STACKED_CORRELATION_HEADER, "test.correlation");
amqpProperties.setHeader(RabbitTemplate.STACKED_REPLY_TO_HEADER, "test.replyTo2");
amqpProperties.setHeader(AmqpHeaders.STACKED_CORRELATION_HEADER, "test.correlation");
amqpProperties.setHeader(AmqpHeaders.STACKED_REPLY_TO_HEADER, "test.replyTo2");
Map<String, Object> headerMap = headerMapper.toHeadersFromReply(amqpProperties);
assertEquals("test.appId", headerMap.get(AmqpHeaders.APP_ID));
assertEquals("test.clusterId", headerMap.get(AmqpHeaders.CLUSTER_ID));