diff --git a/build.gradle b/build.gradle index c714415cbc..1fe57fcef0 100644 --- a/build.gradle +++ b/build.gradle @@ -62,7 +62,7 @@ subprojects { subproject -> springVersionDefault = '4.0.0.RELEASE' springVersion = project.hasProperty('springVersion') ? getProperty('springVersion') : springVersionDefault - springAmqpVersion = '1.2.0.RELEASE' + springAmqpVersion = '1.3.0.BUILD-SNAPSHOT' springDataMongoVersion = '1.1.1.RELEASE' springDataRedisVersion = '1.1.0.RELEASE' lettuceVersion = '2.3.3' diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java index f9f9b0418d..18f5446adb 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/StubRabbitConnectionFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2014 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. @@ -54,35 +54,43 @@ import com.rabbitmq.client.ShutdownSignalException; */ public class StubRabbitConnectionFactory implements ConnectionFactory { + @Override public Connection createConnection() throws AmqpException { return new StubConnection(); } + @Override public String getHost() { return null; } + @Override public int getPort() { return 0; } + @Override public String getVirtualHost() { return null; } + @Override public void addConnectionListener(ConnectionListener listener) { } private static class StubConnection implements Connection { + @Override public Channel createChannel(boolean transactional) throws AmqpException { return new StubChannel(); } + @Override public void close() throws AmqpException { } + @Override public boolean isOpen() { return false; } @@ -90,48 +98,61 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { private static class StubChannel implements Channel { + @Override public void addShutdownListener(ShutdownListener listener) { } + @Override public void removeShutdownListener(ShutdownListener listener) { } + @Override public ShutdownSignalException getCloseReason() { return null; } + @Override public void notifyListeners() { } + @Override public boolean isOpen() { return false; } + @Override public int getChannelNumber() { return 0; } + @Override public com.rabbitmq.client.Connection getConnection() { return null; } + @Override public void close() throws IOException { } + @Override public void close(int closeCode, String closeMessage) throws IOException { } + @Override public FlowOk flow(boolean active) throws IOException { return null; } + @Override public FlowOk getFlow() { return null; } + @Override public void abort() throws IOException { } + @Override public void abort(int closeCode, String closeMessage) throws IOException { } @@ -140,6 +161,7 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { return null; } + @Override public void addReturnListener(ReturnListener listener) { } @@ -161,162 +183,227 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { public void setConfirmListener(ConfirmListener listener) { } + @Override public Consumer getDefaultConsumer() { return null; } + @Override public void setDefaultConsumer(Consumer consumer) { } + @Override public void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException { } + @Override public void basicQos(int prefetchCount) throws IOException { } + @Override public void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException { } + @Override public void basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body) throws IOException { } + @Override public DeclareOk exchangeDeclare(String exchange, String type) throws IOException { return null; } + @Override public DeclareOk exchangeDeclare(String exchange, String type, boolean durable) throws IOException { return null; } + @Override public DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, Map arguments) throws IOException { return null; } + @Override public DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map arguments) throws IOException { return null; } + @Override public DeclareOk exchangeDeclarePassive(String name) throws IOException { return null; } + @Override public DeleteOk exchangeDelete(String exchange, boolean ifUnused) throws IOException { return null; } + @Override public DeleteOk exchangeDelete(String exchange) throws IOException { return null; } + @Override public BindOk exchangeBind(String destination, String source, String routingKey) throws IOException { return null; } + @Override public BindOk exchangeBind(String destination, String source, String routingKey, Map arguments) throws IOException { return null; } + @Override public UnbindOk exchangeUnbind(String destination, String source, String routingKey) throws IOException { return null; } + @Override public UnbindOk exchangeUnbind(String destination, String source, String routingKey, Map arguments) throws IOException { return null; } + @Override public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclare() throws IOException { return null; } + @Override public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclare( - String queue, boolean durable, boolean exclusive, + final String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) throws IOException { - return null; + return new com.rabbitmq.client.AMQP.Queue.DeclareOk() { + + @Override + public int protocolClassId() { + return 0; + } + + @Override + public int protocolMethodId() { + return 0; + } + + @Override + public String protocolMethodName() { + return null; + } + + @Override + public int getConsumerCount() { + return 0; + } + + @Override + public int getMessageCount() { + return 0; + } + + @Override + public String getQueue() { + return queue; + }}; } + @Override public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclarePassive( String queue) throws IOException { return null; } + @Override public com.rabbitmq.client.AMQP.Queue.DeleteOk queueDelete(String queue) throws IOException { return null; } + @Override public com.rabbitmq.client.AMQP.Queue.DeleteOk queueDelete( String queue, boolean ifUnused, boolean ifEmpty) throws IOException { return null; } + @Override public com.rabbitmq.client.AMQP.Queue.BindOk queueBind(String queue, String exchange, String routingKey) throws IOException { return null; } + @Override public com.rabbitmq.client.AMQP.Queue.BindOk queueBind(String queue, String exchange, String routingKey, Map arguments) throws IOException { return null; } + @Override public com.rabbitmq.client.AMQP.Queue.UnbindOk queueUnbind( String queue, String exchange, String routingKey) throws IOException { return null; } + @Override public com.rabbitmq.client.AMQP.Queue.UnbindOk queueUnbind( String queue, String exchange, String routingKey, Map arguments) throws IOException { return null; } + @Override public PurgeOk queuePurge(String queue) throws IOException { return null; } + @Override public GetResponse basicGet(String queue, boolean autoAck) throws IOException { return null; } + @Override public void basicAck(long deliveryTag, boolean multiple) throws IOException { } + @Override public void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException { } + @Override public void basicReject(long deliveryTag, boolean requeue) throws IOException { } + @Override public String basicConsume(String queue, Consumer callback) throws IOException { return null; } + @Override public String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException { return null; } + @Override public String basicConsume(String queue, boolean autoAck, String consumerTag, Consumer callback) throws IOException { return null; } + @Override public String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map arguments, Consumer callback) @@ -324,92 +411,116 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { return null; } + @Override public void basicCancel(String consumerTag) throws IOException { } + @Override public RecoverOk basicRecover() throws IOException { return null; } + @Override public RecoverOk basicRecover(boolean requeue) throws IOException { return null; } + @Override @Deprecated public void basicRecoverAsync(boolean requeue) throws IOException { } + @Override public com.rabbitmq.client.AMQP.Tx.SelectOk txSelect() throws IOException { return null; } + @Override public CommitOk txCommit() throws IOException { return null; } + @Override public RollbackOk txRollback() throws IOException { return null; } + @Override public SelectOk confirmSelect() throws IOException { return null; } + @Override public long getNextPublishSeqNo() { return 0; } + @Override public void asyncRpc(Method method) throws IOException { } + @Override public Command rpc(Method method) throws IOException { return null; } + @Override public boolean removeReturnListener(ReturnListener listener) { return false; } + @Override public void clearReturnListeners() { } + @Override public void addFlowListener(FlowListener listener) { } + @Override public boolean removeFlowListener(FlowListener listener) { return false; } + @Override public void clearFlowListeners() { } + @Override public void addConfirmListener(ConfirmListener listener) { } + @Override public boolean removeConfirmListener(ConfirmListener listener) { return false; } + @Override public void clearConfirmListeners() { } + @Override public boolean waitForConfirms() throws InterruptedException { return false; } + @Override public void waitForConfirmsOrDie() throws IOException, InterruptedException { } + @Override public boolean waitForConfirms(long timeout) throws InterruptedException, TimeoutException { return false; } + @Override public void waitForConfirmsOrDie(long timeout) throws IOException, InterruptedException, TimeoutException { } + @Override public void basicPublish(String arg0, String arg1, boolean arg2, BasicProperties arg3, byte[] arg4) throws IOException { } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java index c09276f9c6..05d3523eea 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -28,11 +28,13 @@ import static org.mockito.Mockito.when; import java.util.ArrayList; import java.util.List; +import java.util.Map; import org.apache.commons.logging.Log; import org.junit.Test; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; + import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Message; @@ -45,6 +47,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageDeliveryException; +import com.rabbitmq.client.AMQP.Queue.DeclareOk; import com.rabbitmq.client.Channel; @@ -55,11 +58,16 @@ import com.rabbitmq.client.Channel; */ public class DispatcherHasNoSubscribersTests { + @SuppressWarnings("unchecked") @Test - public void testPtP() { + public void testPtP() throws Exception { final Channel channel = mock(Channel.class); + DeclareOk declareOk = mock(DeclareOk.class); + when(declareOk.getQueue()).thenReturn("noSubscribersChannel"); + when(channel.queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class))).thenReturn(declareOk); Connection connection = mock(Connection.class); doAnswer(new Answer() { + @Override public Channel answer(InvocationOnMock invocation) throws Throwable { return channel; }}).when(connection).createChannel(anyBoolean()); @@ -89,6 +97,7 @@ public class DispatcherHasNoSubscribersTests { final Channel channel = mock(Channel.class); Connection connection = mock(Connection.class); doAnswer(new Answer() { + @Override public Channel answer(InvocationOnMock invocation) throws Throwable { return channel; }}).when(connection).createChannel(anyBoolean()); @@ -121,6 +130,7 @@ public class DispatcherHasNoSubscribersTests { Log logger = mock(Log.class); final ArrayList logList = new ArrayList(); doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) throws Throwable { String message = (String) invocation.getArguments()[0]; diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java index cf4cc122a1..1d2065f54b 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -50,10 +50,6 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.PollableChannel; import org.springframework.integration.amqp.AmqpHeaders; import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.integration.amqp.support.AmqpHeaderMapper; @@ -61,10 +57,14 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.context.NamedComponent; import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ReflectionUtils; @@ -118,6 +118,7 @@ public class AmqpOutboundChannelAdapterParserTests { amqpTemplate = Mockito.spy(amqpTemplate); Mockito.doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) { Object[] args = invocation.getArguments(); org.springframework.amqp.core.Message amqpReplyMessage = (org.springframework.amqp.core.Message) args[2]; @@ -187,6 +188,7 @@ public class AmqpOutboundChannelAdapterParserTests { amqpTemplate = Mockito.spy(amqpTemplate); Mockito.doAnswer(new Answer() { + @Override public Object answer(InvocationOnMock invocation) { Object[] args = invocation.getArguments(); org.springframework.amqp.core.Message amqpReplyMessage = (org.springframework.amqp.core.Message) args[2]; @@ -261,7 +263,7 @@ public class AmqpOutboundChannelAdapterParserTests { MessageChannel requestChannel = context.getBean("toRabbitOnlyWithTemplateChannel", MessageChannel.class); requestChannel.send(MessageBuilder.withPayload("test").build()); Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq("default.test.exchange"), Mockito.eq("default.routing.key"), - Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); + Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); } @Test @@ -277,7 +279,7 @@ public class AmqpOutboundChannelAdapterParserTests { MessageChannel requestChannel = context.getBean("withDefaultAmqpTemplateExchangeAndRoutingKey", MessageChannel.class); requestChannel.send(MessageBuilder.withPayload("test").build()); Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq(""), Mockito.eq(""), - Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); + Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); } @Test @@ -293,7 +295,7 @@ public class AmqpOutboundChannelAdapterParserTests { MessageChannel requestChannel = context.getBean("overrideTemplateAttributesToEmpty", MessageChannel.class); requestChannel.send(MessageBuilder.withPayload("test").build()); Mockito.verify(mockChannel, Mockito.times(1)).basicPublish(Mockito.eq(""), Mockito.eq(""), - Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); + Mockito.anyBoolean(), Mockito.any(BasicProperties.class), Mockito.any(byte[].class)); } @Test