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

INT-4100: Improve RequestReplyExchanger

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

* Add `throws MessagingException` to the `RequestReplyExchanger` to make it fully messaging contract based
* Add test to the `ServiceActivatorDefaultFrameworkMethodTests` to demonstrate how many info is in the stack trace after `RequestReplyExchanger`
* Document how to be with the `throws` clause in custom gateways
* Fix `boolean` attributes in the `spring-integration-5.0.xsd`

* Fix `AbstractChannelAdapterParser` do not populate endpoint attributes if component is nested
* Fix typo in the `AmqpOutboundChannelAdapterParserTests`
* Increase timeouts in the  `MixedDispatcherConfigurationScenarioTests`

Reinstate `auto-startup` to none by default.
 Not all components expects `true` by default in their initial state

Also https://jira.spring.io/browse/INT-3432

Document the behavior of the `throws` clause on gateway method

Remove unrelated `bridge` id from the `<chain>`

Add What's new about  `RequestReplyExchanger` changes

Doc Polishing.
This commit is contained in:
Artem Bilan
2017-03-02 15:31:45 -05:00
committed by Gary Russell
parent 10ce68d3e3
commit 8da89c7cce
9 changed files with 126 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -184,9 +184,9 @@ public class AmqpOutboundChannelAdapterParserTests {
@SuppressWarnings("rawtypes")
@Test
public void amqpOutboundChannelAdapterWithinChain() {
Object eventDrivernConsumer = context.getBean("chainWithRabbitOutbound");
Object eventDrivenConsumer = context.getBean("chainWithRabbitOutbound");
List chainHandlers = TestUtils.getPropertyValue(eventDrivernConsumer, "handler.handlers", List.class);
List chainHandlers = TestUtils.getPropertyValue(eventDrivenConsumer, "handler.handlers", List.class);
AmqpOutboundEndpoint endpoint = (AmqpOutboundEndpoint) chainHandlers.get(0);
assertNull(TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode"));

30
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2017 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.
@@ -65,20 +65,22 @@ public abstract class AbstractChannelAdapterParser extends AbstractBeanDefinitio
}
AbstractBeanDefinition beanDefinition = doParse(element, parserContext, channelName);
MutablePropertyValues propertyValues = beanDefinition.getPropertyValues();
String autoStartup = element.getAttribute(IntegrationNamespaceUtils.AUTO_STARTUP);
if (StringUtils.hasText(autoStartup)) {
propertyValues.add("autoStartup", new TypedStringValue(autoStartup));
}
String phase = element.getAttribute(IntegrationNamespaceUtils.PHASE);
if (StringUtils.hasText(phase)) {
propertyValues.add("phase", new TypedStringValue(phase));
}
String role = element.getAttribute(IntegrationNamespaceUtils.ROLE);
if (StringUtils.hasText(role)) {
if (!StringUtils.hasText(element.getAttribute(ID_ATTRIBUTE))) {
parserContext.getReaderContext().error("When using 'role', 'id' is required", element);
if (!parserContext.isNested()) {
String autoStartup = element.getAttribute(IntegrationNamespaceUtils.AUTO_STARTUP);
if (StringUtils.hasText(autoStartup)) {
propertyValues.add("autoStartup", new TypedStringValue(autoStartup));
}
String phase = element.getAttribute(IntegrationNamespaceUtils.PHASE);
if (StringUtils.hasText(phase)) {
propertyValues.add("phase", new TypedStringValue(phase));
}
String role = element.getAttribute(IntegrationNamespaceUtils.ROLE);
if (StringUtils.hasText(role)) {
if (!StringUtils.hasText(element.getAttribute(ID_ATTRIBUTE))) {
parserContext.getReaderContext().error("When using 'role', 'id' is required", element);
}
IntegrationNamespaceUtils.putLifecycleInRole(role, element.getAttribute(ID_ATTRIBUTE), parserContext);
}
IntegrationNamespaceUtils.putLifecycleInRole(role, element.getAttribute(ID_ATTRIBUTE), parserContext);
}
return beanDefinition;
}

7
spring-integration-core/src/main/java/org/springframework/integration/gateway/RequestReplyExchanger.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2017 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.
@@ -17,6 +17,7 @@
package org.springframework.integration.gateway;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
/**
* Interface for a request/reply Message exchange. This will be used as a default
@@ -24,11 +25,13 @@ import org.springframework.messaging.Message;
*
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Artem Bilan
*
* @since 2.0
*/
@FunctionalInterface
public interface RequestReplyExchanger {
Message<?> exchange(Message<?> request);
Message<?> exchange(Message<?> request) throws MessagingException;
}

33
spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd
View File

@@ -182,7 +182,7 @@
</xsd:annotation>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="fixed-subscriber" default="false" type="xsd:boolean">
<xsd:attribute name="fixed-subscriber" default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
When true, only one subscriber is allowed; the subscriber must be available at context initialization time,
@@ -192,6 +192,9 @@
'message-converter' is not allowed. Default: false.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
@@ -1859,7 +1862,11 @@
<xsd:complexContent>
<xsd:extension base="basePollerType">
<xsd:attribute name="id" type="xsd:string" />
<xsd:attribute name="default" type="xsd:boolean" default="false" />
<xsd:attribute name="default" default="false">
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -2904,12 +2911,15 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="throw-exception-on-rejection" type="xsd:string">
<xsd:attribute name="throw-exception-on-rejection" default="false">
<xsd:annotation>
<xsd:documentation>
Throw an exception if the filter rejects the message (default false).
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
@@ -3632,7 +3642,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="expire-groups-upon-completion" type="xsd:string">
<xsd:attribute name="expire-groups-upon-completion" default="false">
<xsd:annotation>
<xsd:documentation>
Boolean flag specifying if MessageGroup should be removed once completed. Useful for
@@ -3640,6 +3650,9 @@
is the same as the completed MessageGroup will be discarded. Default is 'false'
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
@@ -3769,7 +3782,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="expire-groups-upon-timeout" type="xsd:string">
<xsd:attribute name="expire-groups-upon-timeout">
<xsd:annotation>
<xsd:documentation>
Boolean flag specifying, if a group is completed due to timeout (reaper or
@@ -3779,6 +3792,9 @@
resequencer.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="lock-registry" type="xsd:string">
<xsd:annotation>
@@ -4977,12 +4993,15 @@ endpoint itself is a Polling Consumer for a channel with a queue.
</xsd:attributeGroup>
<xsd:attributeGroup name="smartLifeCycleAttributeGroup">
<xsd:attribute name="auto-startup" type="xsd:string">
<xsd:attribute name="auto-startup">
<xsd:annotation>
<xsd:documentation><![CDATA[
Boolean value indicating whether this endpoint should start automatically.
Boolean value indicating whether this endpoint should start automatically.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="phase" type="xsd:string">
<xsd:annotation>

4
spring-integration-core/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java
View File

@@ -326,10 +326,10 @@ public class MixedDispatcherConfigurationScenarioTests {
executor.execute(messageSenderTask);
}
start.countDown();
assertTrue(allDone.await(5, TimeUnit.SECONDS));
assertTrue(allDone.await(10, TimeUnit.SECONDS));
executor.shutdown();
executor.awaitTermination(5, TimeUnit.SECONDS);
executor.awaitTermination(10, TimeUnit.SECONDS);
assertTrue("not all messages were accepted", failed.get());
verify(handlerA, times(14)).handleMessage(message);

6
spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests-context.xml
View File

@@ -42,9 +42,9 @@
<gateway id="gateway" default-request-channel="requestChannel" default-reply-channel="replyChannel"/>
<channel id="requestChannel"/>
<bridge id="bridge" input-channel="requestChannel" output-channel="replyChannel"/>
<chain input-channel="requestChannel" output-channel="replyChannel">
<transformer expression="payload == 'foo' ? T(org.springframework.integration.handler.ServiceActivatorDefaultFrameworkMethodTests).throwIllegalStateException('Wrong payload') : payload"/>
</chain>
<channel id="replyChannel">
<queue/>

24
spring-integration-core/src/test/java/org/springframework/integration/handler/ServiceActivatorDefaultFrameworkMethodTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.handler;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
@@ -41,10 +42,12 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transformer.MessageTransformationException;
import org.springframework.integration.util.StackTraceUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
@@ -59,6 +62,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
*
* @since 2.0.1
*/
@ContextConfiguration
@@ -107,8 +111,22 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(replyChannel).build();
this.gatewayTestInputChannel.send(message);
Message<?> reply = replyChannel.receive(0);
assertEquals("gatewayTestInputChannel,gatewayTestService,gateway,requestChannel,bridge,replyChannel",
assertEquals("gatewayTestInputChannel,gatewayTestService,gateway,requestChannel,replyChannel",
reply.getHeaders().get("history").toString());
message = MessageBuilder.withPayload("foo").setReplyChannel(replyChannel).build();
try {
this.gatewayTestInputChannel.send(message);
fail("Exception expected");
}
catch (Exception e) {
assertThat(e, instanceOf(MessageHandlingException.class));
assertThat(e.getCause(), instanceOf(MessageTransformationException.class));
assertThat(e.getCause().getCause(), instanceOf(MessageHandlingException.class));
assertThat(e.getCause().getCause().getCause(), instanceOf(java.lang.IllegalStateException.class));
assertThat(e.getMessage(), containsString("Expression evaluation failed"));
assertThat(e.getMessage(), containsString("Wrong payload"));
}
}
@Test
@@ -258,6 +276,10 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
assertEquals("test", ((MessagingException) error.getPayload()).getFailedMessage().getPayload());
}
public static void throwIllegalStateException(String message) {
throw new IllegalStateException(message);
}
@SuppressWarnings("unused")
private static class TestReplyingMessageHandler extends AbstractReplyProducingMessageHandler {

46
src/reference/asciidoc/gateway.adoc
View File

@@ -322,10 +322,52 @@ If a method has no argument and no return value, but does contain a payload expr
==== Error Handling
Of course, the Gateway invocation might result in errors.
By default any error that has occurred downstream will be re-thrown as a `MessagingException` (`RuntimeException`) upon the Gateway's method invocation.
By default, any error that occurs downstream will be re-thrown as is upon the Gateway's method invocation.
For example, consider the following simple flow:
[source]
----
gateway -> service-activator
----
If the service invoked by the service activator throws a `FooException`, the framework wraps it in a `MessagingException`, attaching the message passed to the service activator in the `failedMessage` property.
Any logging performed by the framework will therefore have full context of the failure.
When the exception is caught by the gateway, by default, the `FooException` will be unwrapped and thrown to the caller.
You can configure a `throws` clause on the gateway method declaration for matching the particular exception type in the cause chain.
For example if you would like to catch a whole `MessagingException` with all the messaging information of the reason of downstream error, you should have a gateway method like this:
[source, java]
----
public interface MyGateway {
void performProcess() throws MessagingException;
}
----
Since we encourage POJO programming, you may not want to expose the caller to messaging infrastructure.
If your gateway method does not have a `throws` clause, the gateway will traverse the cause tree looking for a `RuntimeException` (that is not a `MessagingException`).
If none is found, the framework will simply throw the `MessagingException`.
If the `FooException` in the discussion above has a cause `BarException` and your method `throws BarException` then the gateway will further unwrap that and throw it to the caller.
When a gateway is declared with no `service-interface`, an internal framework interface `RequestReplyExchanger` is used.
[source, java]
----
public interface RequestReplyExchanger {
Message<?> exchange(Message<?> request) throws MessagingException;
}
----
Before _version 5.0_ this `exchange` method did not have a `throws` clause and therefore the exception was unwrapped.
If you are using this interface, and wish to restore the previous unwrap behavior, use a custom `service-interface` instead, or simply access the `cause` of the `MessagingException` yourself.
However there are times when you may want to simply log the error rather than propagating it, or you may want to treat an Exception as a valid reply, by mapping it to a Message that will conform to some "error message" contract that the caller understands.
To accomplish this, the Gateway provides support for a Message Channel dedicated to the errors via the _error-channel_ attribute.
In the example below, you can see that a 'transformer' is used to create a reply Message from the Exception.
In the example below, you can see that a 'transformer' is used to create a reply `Message` from the `Exception`.
[source,xml]
----

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

@@ -67,6 +67,9 @@ If your application is relying on the previous behavior, rename your `connection
The gateway now correctly sets the `errorChannel` header when the gateway method has a `void` return type and an error channel is provided.
Previously, the header was not populated.
This had the effect that synchronous downstream flows (running on the calling thread) would send the exception to the configured channel but an exception on an async downstream flow would be sent to the default `errorChannel` instead.
The `RequestReplyExchanger` interface now has a `throws MessagingException` clause to meet all the proposed messages exchange contract.
See <<gateway-error-handling>> for more information.
==== Mail Changes
@@ -107,10 +110,10 @@ The FTP and SFTP outbound gateways now support the `REPLACE_IF_MODIFIED` `FileEx
The (S)FTP streaming inbound channel adapters now add remote file information in a message header.
See <<ftp>> and <<sftp>> for more information.
The FTP and SFTP outbound channel adapters, as well as `PUT` command of the outbound gateways, now support `InputStream` as `payload`, too.
See <<ftp>> and <<sftp>> for more information.
==== Integration Properties
Since _version 4.3.2_ a new `spring.integration.readOnly.headers` global property has been added to customize the list of headers which should not be copied to a newly created `Message` by the `MessageBuilder`.