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

SplitterMessageHandler replaces SplitterMessageHandlerAdapter.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-12 00:22:02 +00:00
parent c63f65f256
commit c2e6be171d
14 changed files with 198 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.config.annotation;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
/**
* Strategy interface for post-processing annotated methods.
@@ -27,7 +27,7 @@ public interface AnnotationMethodPostProcessor {
Object postProcess(Object bean, String beanName, Class<?> originalBeanClass);
AbstractEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
MessageEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
org.springframework.integration.annotation.MessageEndpoint endpointAnnotation);
}

8
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java
View File

@@ -37,9 +37,9 @@ import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.SimpleEndpoint;
import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
@@ -129,9 +129,9 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost
return handlerChain;
}
public AbstractEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
public MessageEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) {
HandlerEndpoint endpoint = new HandlerEndpoint((MessageHandler) bean);
SimpleEndpoint<MessageHandler> endpoint = new SimpleEndpoint<MessageHandler>((MessageHandler) bean);
String outputChannelName = endpointAnnotation.output();
if (StringUtils.hasText(outputChannelName)) {
endpoint.setOutputChannelName(outputChannelName);

5
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -33,7 +33,6 @@ import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
@@ -87,10 +86,10 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init
AnnotationMethodPostProcessor postProcessor = entry.getValue();
bean = postProcessor.postProcess(bean, beanName, beanClass);
if (endpointAnnotation != null && entry.getKey().isAssignableFrom(bean.getClass())) {
AbstractEndpoint endpoint =
org.springframework.integration.endpoint.MessageEndpoint endpoint =
postProcessor.createEndpoint(bean, beanName, beanClass, endpointAnnotation);
if (endpoint != null) {
endpoint.setName(beanName + "." + entry.getKey().getSimpleName() + ".endpoint");
endpoint.setBeanName(beanName + "." + entry.getKey().getSimpleName() + ".endpoint");
Poller pollerAnnotation = AnnotationUtils.findAnnotation(beanClass, Poller.class);
if (pollerAnnotation != null) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());

7
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.endpoint;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.Schedule;
@@ -25,10 +26,12 @@ import org.springframework.integration.scheduling.Schedule;
*
* @author Mark Fisher
*/
public interface MessageEndpoint extends MessageTarget {
public interface MessageEndpoint extends MessageTarget, BeanNameAware {
String getName();
void setSchedule(Schedule schedule);
Schedule getSchedule();
void setSource(MessageSource<?> source);
@@ -39,6 +42,8 @@ public interface MessageEndpoint extends MessageTarget {
MessageTarget getTarget();
void setInputChannelName(String inputChannelName);
String getInputChannelName();
String getOutputChannelName();

13
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SimpleEndpoint.java
View File

@@ -332,9 +332,10 @@ public class SimpleEndpoint<T extends MessageHandler> implements MessageEndpoint
/* TODO: remove the following methods after they are removed from the MessageEndpoint interface. */
private String inputChannelName;
private String outputChannelName;
private MessageSource<?> source;
private volatile String inputChannelName;
private volatile String outputChannelName;
private volatile MessageSource<?> source;
private volatile Schedule schedule;
public String getInputChannelName() {
return this.inputChannelName;
@@ -356,7 +357,11 @@ public class SimpleEndpoint<T extends MessageHandler> implements MessageEndpoint
}
public Schedule getSchedule() {
return null;
return this.schedule;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
public MessageSource<?> getSource() {

13
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -181,8 +181,8 @@ public abstract class AbstractMessageHandler implements MessageHandler, Initiali
}
}
public Message<?> handle(Message<?> message) {
if (message == null || message.getPayload() == null) {
public Message<?> handle(Message<?> requestMessage) {
if (requestMessage == null || requestMessage.getPayload() == null) {
if (logger.isDebugEnabled()) {
logger.debug("message handler received a null message");
}
@@ -191,21 +191,22 @@ public abstract class AbstractMessageHandler implements MessageHandler, Initiali
if (!this.initialized) {
this.afterPropertiesSet();
}
Object result = (this.invoker != null) ? this.invokeHandlerMethod(message) : message.getPayload();
Object result = (this.invoker != null) ?
this.invokeHandlerMethod(requestMessage) : requestMessage.getPayload();
if (result == null) {
return null;
}
return this.createReplyMessage(result, message.getHeaders());
return this.createReplyMessage(result, requestMessage);
}
/**
* Subclasses must implement this method to generate the reply Message.
*
* @param result the return value from an adapter method, or the Message payload if not acting as an adapter
* @param requestHeaders the MessageHeaders of the original request Message
* @param requestMessage the original request Message
* @return the Message to be sent to the reply MessageTarget
*/
protected abstract Message<?> createReplyMessage(Object result, MessageHeaders requestHeaders);
protected abstract Message<?> createReplyMessage(Object result, Message<?> requestMessage);
private Object invokeHandlerMethod(Message<?> message) {

8
org.springframework.integration/src/main/java/org/springframework/integration/handler/DefaultMessageHandler.java
View File

@@ -18,7 +18,6 @@ package org.springframework.integration.handler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHeaders;
/**
* The default MessageHandler implementation. Creates a Message for the reply payload.
@@ -30,8 +29,11 @@ import org.springframework.integration.message.MessageHeaders;
public class DefaultMessageHandler extends AbstractMessageHandler {
@Override
protected Message<?> createReplyMessage(Object result, MessageHeaders requestHeaders) {
return MessageBuilder.fromPayload(result).copyHeaders(requestHeaders).setCorrelationId(requestHeaders.getId()).build();
protected Message<?> createReplyMessage(Object result, Message<?> requestMessage) {
return MessageBuilder.fromPayload(result)
.copyHeaders(requestMessage.getHeaders())
.setCorrelationId(requestMessage.getHeaders().getId())
.build();
}
}

133
org.springframework.integration/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java
View File

@@ -1,133 +0,0 @@
/*
* Copyright 2002-2008 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.router;
import java.lang.reflect.Method;
import java.util.Collection;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
/**
* MessageHandler adapter for methods annotated with {@link Splitter @Splitter}.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter implements ChannelRegistryAware {
private volatile String outputChannelName;
private volatile long sendTimeout = -1;
public SplitterMessageHandlerAdapter(Object object, Method method) {
this.setObject(object);
this.setMethod(method);
if (method.getParameterTypes().length < 1) {
throw new ConfigurationException("The splitter method must accept at least one argument.");
}
if (method.getParameterTypes()[0].equals(Message.class)) {
this.setMethodExpectsMessage(true);
}
}
public SplitterMessageHandlerAdapter(Object object, String methodName) {
this.setObject(object);
this.setMethodName(methodName);
}
public SplitterMessageHandlerAdapter() {
}
public void setOutputChannelName(String outputChannelName) {
this.outputChannelName = outputChannelName;
}
public void setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
}
@Override
protected final Message<?> handleReturnValue(Object returnValue, Message<?> originalMessage) {
if (returnValue == null) {
if (logger.isWarnEnabled()) {
logger.warn("Splitter method returned null.");
}
return null;
}
if (returnValue instanceof Collection) {
Collection<?> items = (Collection<?>) returnValue;
int sequenceNumber = 0;
int sequenceSize = items.size();
for (Object item : items) {
Message<?> splitMessage = (item instanceof Message<?>) ?
(Message<?>) item : this.createReplyMessage(item, originalMessage);
splitMessage = MessageBuilder.fromMessage(splitMessage)
.setCorrelationId(originalMessage.getHeaders().getId())
.setSequenceNumber(++sequenceNumber)
.setSequenceSize(sequenceSize).build();
this.sendMessage(splitMessage, this.outputChannelName);
}
}
else if (returnValue.getClass().isArray()) {
Object[] array = (Object[]) returnValue;
int sequenceNumber = 0;
int sequenceSize = array.length;
for (Object item : array) {
Message<?> splitMessage = (item instanceof Message<?>) ?
(Message<?>) item : this.createReplyMessage(item, originalMessage);
splitMessage = MessageBuilder.fromMessage(splitMessage)
.setCorrelationId(originalMessage.getHeaders().getId())
.setSequenceNumber(++sequenceNumber)
.setSequenceSize(sequenceSize).build();
this.sendMessage(splitMessage, this.outputChannelName);
}
}
else {
throw new ConfigurationException(
"splitter method must return either a Collection or array");
}
return null;
}
private boolean sendMessage(Message<?> message, String channelName) {
ChannelRegistry channelRegistry = this.getChannelRegistry();
if (channelRegistry == null) {
throw new IllegalStateException(this.getClass().getSimpleName() + " requires a ChannelRegistry reference.");
}
MessageChannel channel = channelRegistry.lookupChannel(channelName);
if (channel == null) {
if (logger.isWarnEnabled()) {
logger.warn("unable to resolve channel for name '" + channelName + "'");
}
return false;
}
if (logger.isDebugEnabled()) {
logger.debug("sending message to channel '" + channelName + "'");
}
return (this.sendTimeout < 0) ? channel.send(message) : channel.send(message, this.sendTimeout);
}
}

6
org.springframework.integration/src/main/java/org/springframework/integration/router/config/SplitterMessageHandlerCreator.java
View File

@@ -25,7 +25,7 @@ import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.config.AbstractMessageHandlerCreator;
import org.springframework.integration.router.SplitterMessageHandlerAdapter;
import org.springframework.integration.splitter.SplitterMessageHandler;
/**
* Creates a {@link MessageHandler} adapter for splitter methods.
@@ -42,9 +42,7 @@ public class SplitterMessageHandlerCreator extends AbstractMessageHandlerCreator
outputChannelName = endpointAnnotation.output();
}
}
SplitterMessageHandlerAdapter adapter = new SplitterMessageHandlerAdapter(object, method);
adapter.setOutputChannelName(outputChannelName);
return adapter;
return new SplitterMessageHandler(object, method);
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/splitter/SplitterMessageHandler.java
View File

@@ -71,8 +71,9 @@ public class SplitterMessageHandler extends AbstractMessageHandler {
this.delimiters = delimiters;
}
protected CompositeMessage createReplyMessage(Object result, MessageHeaders requestHeaders) {
List<Message<?>> results = new ArrayList<Message<?>>();
protected CompositeMessage createReplyMessage(Object result, Message<?> requestMessage) {
MessageHeaders requestHeaders = requestMessage.getHeaders();
List<Message<?>> results = new ArrayList<Message<?>>();
if (result instanceof Collection) {
Collection<?> items = (Collection<?>) result;
int sequenceNumber = 0;