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

INT-3802: Spring AMQP 1.5.0.RC1, Reactor 2.0.5

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3802

For publisher confirms and returns, the `RabbitTemplate` now checks for a
`CachingConnectionFactory` which can't be mocked/stubbed.

Convert the mock tests for these features to real broker tests.

Also see AMQP-524.

Also, the lifecycle attributes were missing from the AMQP outbound endpoints.

Fix a couple typos in the AMQP xsd.
Fix deprecation since Reactor-2.0.5
This commit is contained in:
Gary Russell
2015-08-14 11:58:45 -04:00
committed by Artem Bilan
parent 1043ffabcb
commit 043e8514f6
10 changed files with 241 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.gradle
View File

@@ -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'

7
spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp-4.2.xsd
View File

@@ -109,7 +109,7 @@
<xsd:attribute name="reply-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -501,6 +501,7 @@ property set to TRUE.
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:complexType>
<xsd:simpleType name="deliveryModeEnumeration">
@@ -688,7 +689,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
<xsd:attributeGroup name="containerOnlyAttributes">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>

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

@@ -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() {
}

15
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests-context.xml
View File

@@ -46,6 +46,7 @@
<amqp:outbound-channel-adapter id="withPublisherConfirms" channel="pcRequestChannel"
exchange-name="outboundchanneladapter.test.1"
mapped-request-headers="foo*"
auto-startup="false"
amqp-template="amqpTemplateConfirms"
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
confirm-ack-channel="ackChannel"/>
@@ -56,20 +57,6 @@
<int:queue/>
</int:channel>
<rabbit:template id="amqpTemplateReturns" connection-factory="connectionFactory" mandatory="true" />
<amqp:outbound-channel-adapter id="withReturns" channel="returnRequestChannel"
exchange-name="outboundchanneladapter.test.1"
mapped-request-headers="foo*"
amqp-template="amqpTemplateReturns"
return-channel="returnChannel"/>
<int:channel id="returnRequestChannel"/>
<int:channel id="returnChannel">
<int:queue/>
</int:channel>
<amqp:outbound-channel-adapter id="withDefaultAmqpTemplateExchangeAndRoutingKey"/>
<rabbit:template id="amqpTemplateWithSuppliedExchangeAndRoutingKey" connection-factory="connectionFactory"

71
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* 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.
@@ -34,9 +34,7 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
@@ -48,16 +46,12 @@ import org.mockito.internal.stubbing.answers.DoesNothing;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
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.PublisherCallbackChannel;
import org.springframework.amqp.rabbit.support.PublisherCallbackChannelImpl;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.BeansException;
@@ -80,13 +74,15 @@ 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.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ReflectionUtils;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
@@ -187,30 +183,6 @@ public class AmqpOutboundChannelAdapterParserTests {
assertSame(nullChannel, TestUtils.getPropertyValue(endpoint, "confirmNackChannel"));
}
@Test
public void withPublisherConfirms() 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("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));
}
@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<String, Object> headers = new HashMap<String, Object>();
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 " +

31
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests-context.xml
View File

@@ -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=""/>
</int:chain>
<rabbit:template id="amqpTemplateConfirms" connection-factory="connectionFactory" reply-timeout="10"/>
<amqp:outbound-gateway request-channel="pcRequestChannel"
reply-channel="fromRabbit"
exchange-name="si.test.exchange"
mapped-request-headers="foo*"
requires-reply="false"
amqp-template="amqpTemplateConfirms"
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
confirm-ack-channel="ackChannel"/>
<int:channel id="pcRequestChannel"/>
<rabbit:template id="amqpTemplateConfirmsMC" connection-factory="connectionFactory" reply-timeout="10"/>
<amqp:outbound-gateway request-channel="pcMessageCorrelationRequestChannel"
reply-channel="fromRabbit"
exchange-name="si.test.exchange"
mapped-request-headers="foo*"
requires-reply="false"
amqp-template="amqpTemplateConfirmsMC"
confirm-correlation-expression="#this"
confirm-ack-channel="ackChannel"/>
<int:channel id="pcMessageCorrelationRequestChannel"/>
<int:channel id="ackChannel">
<int:queue/>
</int:channel>
</beans>

51
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
View File

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

83
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml Normal file
View File

@@ -0,0 +1,83 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:amqp="http://www.springframework.org/schema/integration/amqp"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp http://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:channel id="fromRabbit">
<int:queue />
</int:channel>
<rabbit:template id="amqpTemplateConfirms" connection-factory="connectionFactory" reply-timeout="10" />
<amqp:outbound-gateway id="pcGateway"
request-channel="pcRequestChannel"
reply-channel="fromRabbit"
exchange-name=""
routing-key="#{queue.name}"
mapped-request-headers="foo*"
requires-reply="false"
amqp-template="amqpTemplateConfirms"
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
confirm-ack-channel="ackChannel" />
<int:channel id="pcRequestChannel"/>
<rabbit:template id="amqpTemplateConfirmsMC" connection-factory="connectionFactory" reply-timeout="10" />
<amqp:outbound-gateway request-channel="pcMessageCorrelationRequestChannel"
reply-channel="fromRabbit"
exchange-name=""
routing-key="#{queue.name}"
mapped-request-headers="foo*"
requires-reply="false"
amqp-template="amqpTemplateConfirmsMC"
confirm-correlation-expression="#this"
confirm-ack-channel="ackChannel" />
<int:channel id="pcMessageCorrelationRequestChannel" />
<int:channel id="ackChannel">
<int:queue />
</int:channel>
<rabbit:template id="amqpTemplateConfirmsAdapter" connection-factory="connectionFactory"/>
<amqp:outbound-channel-adapter id="withPublisherConfirms" channel="pcRequestChannelAdapter"
exchange-name=""
routing-key="#{queue.name}"
mapped-request-headers="foo*"
amqp-template="amqpTemplateConfirmsAdapter"
confirm-correlation-expression="headers['amqp_confirmCorrelationData']"
confirm-ack-channel="ackChannel" />
<int:channel id="pcRequestChannelAdapter"/>
<rabbit:template id="amqpTemplateReturns" connection-factory="connectionFactory" mandatory="true" />
<amqp:outbound-channel-adapter id="withReturns" channel="returnRequestChannel"
exchange-name=""
routing-key="#{queue.name + queue.name}"
mapped-request-headers="foo*"
amqp-template="amqpTemplateReturns"
return-channel="returnChannel" />
<int:channel id="returnRequestChannel"/>
<int:channel id="returnChannel">
<int:queue />
</int:channel>
<rabbit:connection-factory id="connectionFactory"
host="localhost" publisher-returns="true" publisher-confirms="true" />
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:queue id="queue" />
</beans>

129
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java Normal file
View File

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

13
spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java
View File

@@ -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.<Object>task((Environment) this.reactorEnvironment,
Functions.supplier(new AsyncInvocationTask(invocation)));
new Supplier<Object>() {
@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);
}