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

Merge branch 'master' of git.springsource.org:spring-integration/spring-integration

Browse Source 
Conflicts:
	spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
This commit is contained in:
Iwein Fuld
2010-09-13 22:12:35 +02:00
parent 95e1230238 eea658deae
commit aae01dc3d5
19 changed files with 342 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-core/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java
View File

@@ -31,6 +31,7 @@ import org.springframework.util.StringUtils;
public class SplitterFactoryBean extends AbstractMessageHandlerFactoryBean {
private volatile Long sendTimeout;
private volatile boolean requiresReply;
public void setSendTimeout(Long sendTimeout) {
this.sendTimeout = sendTimeout;
@@ -64,7 +65,14 @@ public class SplitterFactoryBean extends AbstractMessageHandlerFactoryBean {
if (this.sendTimeout != null) {
splitter.setSendTimeout(sendTimeout);
}
splitter.setRequiresReply(requiresReply);
return splitter;
}
public boolean isRequiresReply() {
return requiresReply;
}
public void setRequiresReply(boolean requiresReply) {
this.requiresReply = requiresReply;
}
}

12
spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingOperations.java
View File

@@ -27,6 +27,18 @@ import org.springframework.integration.MessageChannel;
*/
public interface AsyncMessagingOperations {
Future<?> asyncSend(Message<?> message);
Future<?> asyncSend(MessageChannel channel, Message<?> message);
Future<?> asyncSend(String channelName, Message<?> message);
Future<?> asyncConvertAndSend(Object message);
Future<?> asyncConvertAndSend(MessageChannel channel, Object message);
Future<?> asyncConvertAndSend(String channelName, Object message);
Future<Message<?>> asyncReceive();
Future<Message<?>> asyncReceive(PollableChannel channel);

48
spring-integration-core/src/main/java/org/springframework/integration/core/AsyncMessagingTemplate.java
View File

@@ -42,6 +42,54 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe
(AsyncTaskExecutor) executor : new TaskExecutorAdapter(executor);
}
public Future<?> asyncSend(final Message<?> message) {
return this.executor.submit(new Runnable() {
public void run() {
send(message);
}
});
}
public Future<?> asyncSend(final MessageChannel channel, final Message<?> message) {
return this.executor.submit(new Runnable() {
public void run() {
send(channel, message);
}
});
}
public Future<?> asyncSend(final String channelName, final Message<?> message) {
return this.executor.submit(new Runnable() {
public void run() {
send(channelName, message);
}
});
}
public Future<?> asyncConvertAndSend(final Object object) {
return this.executor.submit(new Runnable() {
public void run() {
convertAndSend(object);
}
});
}
public Future<?> asyncConvertAndSend(final MessageChannel channel, final Object object) {
return this.executor.submit(new Runnable() {
public void run() {
convertAndSend(channel, object);
}
});
}
public Future<?> asyncConvertAndSend(final String channelName, final Object object) {
return this.executor.submit(new Runnable() {
public void run() {
convertAndSend(channelName, object);
}
});
}
public Future<Message<?>> asyncReceive() {
return this.executor.submit(new Callable<Message<?>>() {
public Message<?> call() throws Exception {

9
spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
View File

@@ -20,6 +20,8 @@ import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import java.util.*;
@@ -38,7 +40,10 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
@SuppressWarnings("unchecked")
protected final Object handleRequestMessage(Message<?> message) {
Object result = this.splitMessage(message);
if (result == null) {
// return null if 'null', empty Collection or empty Array
if ( result == null ||
(result instanceof Collection && CollectionUtils.isEmpty((Collection<?>)result)) ||
(result.getClass().isArray() && ObjectUtils.isEmpty((Object[]) result)) ) {
return null;
}
MessageHeaders headers = message.getHeaders();
@@ -53,7 +58,7 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
} else {
messageBuilders = Collections.singletonList(this.createBuilder(result, incomingSequenceDetails, correlationId, 1, 1));
}
return messageBuilders.isEmpty() ? null : messageBuilders;
return messageBuilders;
}
private List<MessageBuilder> messageBuildersForArray(Object result, List<Object[]> incomingSequenceDetails, Object correlationId) {

9
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -1948,6 +1948,15 @@ Name of the header whose value to use.
<xsd:complexContent>
<xsd:extension base="expressionOrInnerEndpointDefinitionAware">
<xsd:attributeGroup ref="inputOutputChannelGroup" />
<xsd:attribute name="requires-reply" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
Specify whether the splitter 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>

92
spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.core;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -47,6 +48,97 @@ import org.springframework.util.Assert;
*/
public class AsyncMessagingTemplateTests {
@Test
public void asyncSendWithDefaultChannel() throws Exception {
QueueChannel channel = new QueueChannel();
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setDefaultChannel(channel);
Message<?> message = MessageBuilder.withPayload("test").build();
Future<?> future = template.asyncSend(message);
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals(message, result);
}
@Test
public void asyncSendWithExplicitChannel() throws Exception {
QueueChannel channel = new QueueChannel();
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
Message<?> message = MessageBuilder.withPayload("test").build();
Future<?> future = template.asyncSend(channel, message);
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals(message, result);
}
@Test
public void asyncSendWithResolvedChannel() throws Exception {
StaticApplicationContext context = new StaticApplicationContext();
context.registerSingleton("testChannel", QueueChannel.class);
context.refresh();
QueueChannel channel = context.getBean("testChannel", QueueChannel.class);
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setBeanFactory(context);
Message<?> message = MessageBuilder.withPayload("test").build();
Future<?> future = template.asyncSend("testChannel", message);
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals(message, result);
}
@Test(expected = TimeoutException.class)
public void asyncSendWithTimeoutException() throws Exception {
QueueChannel channel = new QueueChannel(1);
channel.send(MessageBuilder.withPayload("blocker").build());
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
Future<?> result = template.asyncSend(channel, MessageBuilder.withPayload("test").build());
result.get(100, TimeUnit.MILLISECONDS);
}
@Test
public void asyncConvertAndSendWithDefaultChannel() throws Exception {
QueueChannel channel = new QueueChannel();
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setDefaultChannel(channel);
Future<?> future = template.asyncConvertAndSend("test");
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals("test", result.getPayload());
}
@Test
public void asyncConvertAndSendWithExplicitChannel() throws Exception {
QueueChannel channel = new QueueChannel();
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
Future<?> future = template.asyncConvertAndSend(channel, "test");
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals("test", result.getPayload());
}
@Test
public void asyncConvertAndSendWithResolvedChannel() throws Exception {
StaticApplicationContext context = new StaticApplicationContext();
context.registerSingleton("testChannel", QueueChannel.class);
context.refresh();
QueueChannel channel = context.getBean("testChannel", QueueChannel.class);
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
template.setBeanFactory(context);
Future<?> future = template.asyncConvertAndSend("testChannel", "test");
assertNull(future.get(1000, TimeUnit.MILLISECONDS));
Message<?> result = channel.receive(0);
assertEquals("test", result.getPayload());
}
@Test(expected = TimeoutException.class)
public void asyncConvertAndSendWithTimeoutException() throws Exception {
QueueChannel channel = new QueueChannel(1);
channel.send(MessageBuilder.withPayload("blocker").build());
AsyncMessagingTemplate template = new AsyncMessagingTemplate();
Future<?> result = template.asyncConvertAndSend(channel, "test");
result.get(100, TimeUnit.MILLISECONDS);
}
@Test
public void asyncReceiveWithDefaultChannel() throws Exception {
QueueChannel channel = new QueueChannel();

15
spring-integration-core/src/test/java/org/springframework/integration/router/config/SplitterParserTests.java
View File

@@ -19,13 +19,19 @@ package org.springframework.integration.router.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import java.util.Collections;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
/**
* @author Mark Fisher
@@ -88,5 +94,14 @@ public class SplitterParserTests {
assertEquals("test", result4.getPayload());
assertNull(output.receive(0));
}
@Test(expected=MessageHandlingException.class)
public void splitterParserTestWithRequiresReply() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"splitterParserTests.xml", this.getClass());
context.start();
DirectChannel inputChannel = context.getBean("requiresReplyInput", DirectChannel.class);
inputChannel.send(MessageBuilder.withPayload(Collections.emptyList()).build());
}
}

5
spring-integration-core/src/test/java/org/springframework/integration/router/config/splitterParserTests.xml
View File

@@ -26,6 +26,11 @@
ref="splitterImpl"
input-channel="splitterImplementationInput"
output-channel="output"/>
<splitter id="splitterImplementationRequiresReply"
input-channel="requiresReplyInput"
output-channel="output"
requires-reply="true"/>
<beans:bean id="splitterBean" class="org.springframework.integration.router.config.TestSplitterBean"/>

21
spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java
View File

@@ -16,6 +16,17 @@
package org.springframework.integration.splitter;
import static junit.framework.Assert.assertEquals;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.channel.DirectChannel;
@@ -23,15 +34,6 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.support.MessageBuilder;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import static junit.framework.Assert.assertEquals;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.*;
/**
* @author Mark Fisher
* @author Iwein Fuld
@@ -104,5 +106,4 @@ public class DefaultSplitterTests {
Message<?> output = replyChannel.receive(15);
assertThat(output, is(nullValue()));
}
}