diff --git a/build.gradle b/build.gradle index f85cd37da6..ec97b25390 100644 --- a/build.gradle +++ b/build.gradle @@ -121,7 +121,7 @@ subprojects { subproject -> openJpaVersion = '2.3.0' pahoMqttClientVersion = '1.0.2' postgresVersion = '9.1-901-1.jdbc4' - reactorVersion = '2.0.1.RELEASE' + reactorVersion = '2.0.5.RELEASE' reactorSpringVersion = '2.0.1.RELEASE' romeToolsVersion = '1.5.0' saajApiVersion = '1.3.5' @@ -131,7 +131,7 @@ subprojects { subproject -> tomcatVersion = "8.0.18" smack3Version = '3.2.1' smackVersion = '4.0.6' - springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.5.0.M1' + springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '1.5.0.RC1' // springCloudClusterVersion = '1.0.0.BUILD-SNAPSHOT' springDataMongoVersion = '1.7.2.RELEASE' springDataRedisVersion = '1.5.2.RELEASE' diff --git a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd index b32117690e..06a3b2baf9 100644 --- a/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd +++ b/spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd @@ -109,7 +109,7 @@ - Message Channel to which replies should be sent after being received from an AQMP Queue and converted. + Message Channel to which replies should be sent after being received from an AMQP Queue and converted. @@ -501,6 +501,7 @@ property set to TRUE. + @@ -688,7 +689,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA - Attributes for a SimpleMesssageListenerContainer's properties other than queues, queueNames, messageListener, and + Attributes for a SimpleMessageListenerContainer's properties other than queues, queueNames, messageListener, and autoStartup which may or may not be exposed for configuration depending on what type of component uses this attribute group. This group also does not include any of the properties that are shared with RabbitTemplate, such as channelTransacted, connectionFactory, and messsagePropertiesConverter. @@ -849,7 +850,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA If 'true', and none of the queues are available on the broker, the container will throw a fatal exception during startup and will stop if the queues are deleted when the container is running (after making 3 attempts to passively declare the queues). If false, the container will not throw an exception and go into recovery mode, - attempting to restart according to the 'revcovery-interval'. Default 'true'. + attempting to restart according to the 'recovery-interval'. Default 'true'. 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 355b7e2a58..fb0a47999b 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 @@ -147,6 +147,7 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { } @Override + @SuppressWarnings("deprecation") public boolean flowBlocked() { return false; } @@ -482,15 +483,18 @@ public class StubRabbitConnectionFactory implements ConnectionFactory { } @Override + @SuppressWarnings("deprecation") public void addFlowListener(FlowListener listener) { } @Override + @SuppressWarnings("deprecation") public boolean removeFlowListener(FlowListener listener) { return false; } @Override + @SuppressWarnings("deprecation") public void clearFlowListeners() { } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml index 7652342a50..5fd66b7d5b 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml @@ -46,6 +46,7 @@ @@ -56,20 +57,6 @@ - - - - - - - - - - message = MessageBuilder.withPayload("hello") - .setHeader("amqp_confirmCorrelationData", "foo") - .build(); - requestChannel.send(message); - PollableChannel ackChannel = context.getBean("ackChannel", PollableChannel.class); - publisherCallbackChannel.handleAck(0, false); - Message ack = ackChannel.receive(1000); - assertNotNull(ack); - assertEquals("foo", ack.getPayload()); - assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); - } - @SuppressWarnings("rawtypes") @Test public void amqpOutboundChannelAdapterWithinChain() { @@ -250,42 +222,11 @@ public class AmqpOutboundChannelAdapterParserTests { Mockito.any(CorrelationData.class)); } - @Test - public void withReturns() throws Exception { - ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); - Connection mockConnection = mock(Connection.class); - Channel mockChannel = mock(Channel.class); - - when(connectionFactory.createConnection()).thenReturn(mockConnection); - PublisherCallbackChannelImpl publisherCallbackChannel = spy(new PublisherCallbackChannelImpl(mockChannel)); - doAnswer(new DoesNothing()).when(publisherCallbackChannel).close(); - when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel); - - MessageChannel requestChannel = context.getBean("returnRequestChannel", MessageChannel.class); - Message message = MessageBuilder.withPayload("hello").build(); - requestChannel.send(message); - PollableChannel returnChannel = context.getBean("returnChannel", PollableChannel.class); - RabbitTemplate template = context.getBean("amqpTemplateReturns", RabbitTemplate.class); - Map headers = new HashMap(); - headers.put(PublisherCallbackChannel.RETURN_CORRELATION, template.getUUID()); - BasicProperties properties = mock(BasicProperties.class); - when(properties.getHeaders()).thenReturn(headers); - when(properties.getContentType()).thenReturn("text/plain"); - publisherCallbackChannel.handleReturn(123, "reply text", "anExchange", "bar", properties, "hello".getBytes()); - Message returned = returnChannel.receive(1000); - assertNotNull(returned); - assertEquals(123, returned.getHeaders().get(AmqpHeaders.RETURN_REPLY_CODE)); - assertEquals("reply text", returned.getHeaders().get(AmqpHeaders.RETURN_REPLY_TEXT)); - assertEquals("anExchange", returned.getHeaders().get(AmqpHeaders.RETURN_EXCHANGE)); - assertEquals("bar", returned.getHeaders().get(AmqpHeaders.RETURN_ROUTING_KEY)); - assertEquals("hello", returned.getPayload()); - } - @Test public void testInt2718FailForOutboundAdapterChannelAttribute() { try { new ClassPathXmlApplicationContext("AmqpOutboundChannelAdapterWithinChainParserTests-fail-context.xml", - this.getClass()); + this.getClass()).close(); fail("Expected BeanDefinitionParsingException"); } catch (BeansException e) { @@ -349,7 +290,7 @@ public class AmqpOutboundChannelAdapterParserTests { public void testInt2971HeaderMapperAndMappedHeadersExclusivity() { try { new ClassPathXmlApplicationContext("AmqpOutboundChannelAdapterParserTests-headerMapper-fail-context.xml", - this.getClass()); + this.getClass()).close(); } catch (BeanDefinitionParsingException e) { assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml index 972e5a9b13..338aaa2c0f 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml @@ -15,6 +15,7 @@ exchange-name="si.test.exchange" routing-key="si.test.binding" amqp-template="amqpTemplate" + auto-startup="false" order="5" return-channel="returnChannel"/> @@ -80,34 +81,4 @@ mapped-reply-headers=""/> - - - - - - - - - - - - - - - - diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index bd36379c6d..f22d041eb9 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -23,10 +23,6 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.any; -import static org.mockito.Mockito.doAnswer; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.when; import java.lang.reflect.Field; import java.util.List; @@ -34,17 +30,13 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; import org.mockito.Mockito; -import org.mockito.internal.stubbing.answers.DoesNothing; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; -import org.springframework.amqp.rabbit.connection.Connection; -import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.CorrelationData; -import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; import org.springframework.context.ConfigurableApplicationContext; @@ -56,11 +48,8 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.PollableChannel; import org.springframework.util.ReflectionUtils; -import com.rabbitmq.client.Channel; - /** * @author Oleg Zhurakousky * @author Gary Russell @@ -79,6 +68,7 @@ public class AmqpOutboundGatewayParserTests { ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); Object edc = context.getBean("rabbitGateway"); + assertFalse(TestUtils.getPropertyValue(edc, "autoStartup", Boolean.class)); AmqpOutboundEndpoint gateway = TestUtils.getPropertyValue(edc, "handler", AmqpOutboundEndpoint.class); assertEquals(5, gateway.getOrder()); assertTrue(TestUtils.getPropertyValue(gateway, "requiresReply", Boolean.class)); @@ -347,45 +337,6 @@ public class AmqpOutboundGatewayParserTests { } } - @Test - public void testPublisherConfirms() throws Exception { - ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( - "AmqpOutboundGatewayParserTests-context.xml", this.getClass()); - ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); - Connection mockConnection = mock(Connection.class); - Channel mockChannel = mock(Channel.class); - - when(connectionFactory.createConnection()).thenReturn(mockConnection); - PublisherCallbackChannelImpl publisherCallbackChannel = spy(new PublisherCallbackChannelImpl(mockChannel)); - doAnswer(new DoesNothing()).when(publisherCallbackChannel).close(); - when(mockConnection.createChannel(false)).thenReturn(publisherCallbackChannel); - - MessageChannel requestChannel = context.getBean("pcRequestChannel", MessageChannel.class); - Message message = MessageBuilder.withPayload("hello") - .setHeader("amqp_confirmCorrelationData", "foo") - .build(); - requestChannel.send(message); - PollableChannel ackChannel = context.getBean("ackChannel", PollableChannel.class); - publisherCallbackChannel.handleAck(0, false); - Message ack = ackChannel.receive(1000); - assertNotNull(ack); - assertEquals("foo", ack.getPayload()); - assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); - - // test whole message is correlation - requestChannel = context.getBean("pcMessageCorrelationRequestChannel", MessageChannel.class); - message = MessageBuilder.withPayload("hello") - .build(); - requestChannel.send(message); - publisherCallbackChannel.handleAck(0, false); - ack = ackChannel.receive(1000); - assertNotNull(ack); - assertSame(message.getPayload(), ack.getPayload()); - assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); - - context.close(); - } - public static class FooAdvice extends AbstractRequestHandlerAdvice { @Override diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml new file mode 100644 index 0000000000..6b2eaf2918 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml @@ -0,0 +1,83 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java new file mode 100644 index 0000000000..2ef609ef53 --- /dev/null +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java @@ -0,0 +1,129 @@ +/* + * Copyright 2002-2015 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.amqp.outbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.amqp.rule.BrokerRunning; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.annotation.DirtiesContext.ClassMode; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Oleg Zhurakousky + * @author Gary Russell + * @author Artem Bilan + * @author Gunnar Hillert + * + * @since 2.1 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext(classMode=ClassMode.AFTER_EACH_TEST_METHOD) +public class AmqpOutboundEndpointTests { + + @Rule + public BrokerRunning brokerRunning = BrokerRunning.isRunning(); + + @Autowired + private MessageChannel pcRequestChannel; + + @Autowired + private MessageChannel pcMessageCorrelationRequestChannel; + + @Autowired + private RabbitTemplate amqpTemplateConfirms; + + @Autowired + private Queue queue; + + @Autowired + private PollableChannel ackChannel; + + @Autowired + private MessageChannel pcRequestChannelAdapter; + + @Autowired + private MessageChannel returnRequestChannel; + + @Autowired + private PollableChannel returnChannel; + + @Test + public void testGatewayPublisherConfirms() throws Exception { + + Message message = MessageBuilder.withPayload("hello") + .setHeader("amqp_confirmCorrelationData", "foo") + .build(); + this.pcRequestChannel.send(message); + Message ack = this.ackChannel.receive(10000); + assertNotNull(ack); + assertEquals("foo", ack.getPayload()); + assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); + + // test whole message is correlation + message = MessageBuilder.withPayload("hello") + .build(); + this.pcMessageCorrelationRequestChannel.send(message); + ack = ackChannel.receive(10000); + assertNotNull(ack); + assertSame(message.getPayload(), ack.getPayload()); + assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); + + this.amqpTemplateConfirms.receive(this.queue.getName()); // so queue is deleted + + } + + @Test + public void adapterWithPublisherConfirms() throws Exception { + Message message = MessageBuilder.withPayload("hello") + .setHeader("amqp_confirmCorrelationData", "foo") + .build(); + this.pcRequestChannelAdapter.send(message); + Message ack = this.ackChannel.receive(10000); + assertNotNull(ack); + assertEquals("foo", ack.getPayload()); + assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)); + } + + @Test + public void adapterWithReturns() throws Exception { + Message message = MessageBuilder.withPayload("hello").build(); + this.returnRequestChannel.send(message); + Message returned = returnChannel.receive(10000); + assertNotNull(returned); + assertEquals(message.getPayload(), returned.getPayload()); + } + + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 829d96f21a..67ee3a838d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -64,7 +64,7 @@ import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; import reactor.Environment; -import reactor.fn.Functions; +import reactor.fn.Supplier; import reactor.rx.Promise; import reactor.rx.Promises; @@ -368,7 +368,14 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint throw new IllegalStateException("'reactorEnvironment' is required in case of 'Promise' return type."); } return Promises.task((Environment) this.reactorEnvironment, - Functions.supplier(new AsyncInvocationTask(invocation))); + new Supplier() { + + @Override + public Object get() { + return new AsyncInvocationTask(invocation).call(); + } + + }); } return this.doInvoke(invocation, true); } @@ -637,7 +644,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint } @Override - public Object call() throws Exception { + public Object call() { try { return doInvoke(this.invocation, false); }