INT-1391 added 'requires-reply' attribute to <service-activator> element

org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java

@@ -34,12 +34,12 @@ public class ServiceActivatorParser extends AbstractConsumerEndpointParser {
 	@Override
 	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
 		BeanDefinition innerHandlerDefinition = this.parseInnerHandlerDefinition(element, parserContext);
-		
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
 				IntegrationNamespaceUtils.BASE_PACKAGE + ".handler.ServiceActivatingHandler");
 		if (innerHandlerDefinition != null){
 			builder.addConstructorArgValue(innerHandlerDefinition);
-		} else {
+		}
+		else {
 			String ref = element.getAttribute(REF_ATTRIBUTE);
 			if (!StringUtils.hasText(ref)) {
 				parserContext.getReaderContext().error("The '" + REF_ATTRIBUTE + "' attribute is required for element "
@@ -47,11 +47,11 @@ public class ServiceActivatorParser extends AbstractConsumerEndpointParser {
 			}
 			builder.addConstructorArgReference(ref);
 		}
-		
 		if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
 			String method = element.getAttribute(METHOD_ATTRIBUTE);
 			builder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String");
 		}
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
 		return builder;
 	}

org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-1.0.xsd

@@ -443,7 +443,7 @@
 			default="true" />
 	</xsd:complexType>
 
-	<xsd:element name="service-activator" type="innerEndpointDefinitionAware">
+	<xsd:element name="service-activator">
 		<xsd:annotation>
 			<xsd:documentation>
 				Defines an endpoint for exposing any bean
@@ -459,6 +459,21 @@
 				should be provided along with 'ref'.
 			</xsd:documentation>
 		</xsd:annotation>
+		<xsd:complexType>
+			<xsd:complexContent>
+				<xsd:extension base="innerEndpointDefinitionAware">
+					<xsd:attribute name="requires-reply" type="xsd:string" use="optional">
+						<xsd:annotation>
+							<xsd:documentation>
+	Specify whether the service method must return a non-null value. This value will be
+	FALSE by default, but if set to TRUE, a MessageHandlingException will be thrown when
+	the underlying service method (or expression) returns a NULL value.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
+				</xsd:extension>
+			</xsd:complexContent>
+		</xsd:complexType>
 	</xsd:element>
 
 	<xsd:complexType name="handlerEndpointType">

org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayRequiresReplyTests-context.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/integration"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd
+			http://www.springframework.org/schema/integration
+			http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<gateway id="gateway"
+			 default-request-channel="requestChannel"
+			 default-reply-timeout="3000"
+			 service-interface="org.springframework.integration.gateway.GatewayRequiresReplyTests$TestService" />
+	
+
+	<service-activator input-channel="requestChannel"
+					   ref="testBean"
+					   requires-reply="true"/>
+
+	<beans:bean id="testBean" class="org.springframework.integration.gateway.GatewayRequiresReplyTests$TestServiceBean"/>
+
+	<gateway id="timeoutGateway"
+			 default-request-channel="timeoutChannel"
+			 default-reply-timeout="1000"
+			 service-interface="org.springframework.integration.gateway.GatewayRequiresReplyTests$TestService" />
+			 
+	<channel id="timeoutChannel">
+		<dispatcher task-executor="executor"/>
+	</channel>
+			 
+	<service-activator input-channel="timeoutChannel">
+		<beans:bean class="org.springframework.integration.gateway.GatewayRequiresReplyTests$LongRunningService"/>
+	</service-activator>
+
+	<thread-pool-task-executor id="executor" max-size="5"/>
+
+</beans:beans>

org.springframework.integration/src/test/java/org/springframework/integration/gateway/GatewayRequiresReplyTests.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2002-2011 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.gateway;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+ 
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Mark Fisher
+ * @since 1.0.5
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class GatewayRequiresReplyTests {
+
+	@Autowired
+	private ApplicationContext applicationContext;
+
+
+	@Test
+	public void replyReceived() {
+		TestService gateway = (TestService) applicationContext.getBean("gateway");
+		String result = gateway.test("foo");
+		assertEquals("bar", result);
+	}
+
+	@Test(expected = MessageHandlingException.class)
+	public void noReplyReceived() {
+		TestService gateway = (TestService) applicationContext.getBean("gateway");
+		gateway.test("bad");
+	}
+	
+	@Test
+	public void timedOutGateway() {
+		TestService gateway = (TestService) applicationContext.getBean("timeoutGateway");
+		String result = gateway.test("hello");
+		assertNull(result);
+	}
+
+
+	public static interface TestService {
+		public String test(String s);
+	}
+
+
+	public static class TestServiceBean {
+		public String process(String s) {
+			return "foo".equals(s) ? "bar" : null;
+		}
+	}
+
+
+	public static class LongRunningService {
+		public String echo(String value) throws Exception{
+			Thread.sleep(5000);
+			return value;
+		}
+	}
+
+}