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

INT-937, INT-87 Message History (work in progress)

Browse Source
This commit is contained in:
Mark Fisher
2010-02-19 04:24:51 +00:00
parent a41c862558
commit 746b046b24
27 changed files with 183 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java
View File

@@ -20,7 +20,6 @@ import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -48,7 +47,7 @@ import org.springframework.util.CollectionUtils;
public abstract class AbstractMessageAggregator extends
AbstractMessageBarrierHandler<List<Message<?>>> {
private static final String COMPONENT_TYPE_LABEL = "aggregator";
public static final String COMPONENT_TYPE_LABEL = "aggregator";
private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
@@ -89,11 +88,6 @@ public abstract class AbstractMessageAggregator extends
}
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
event.setComponentType(COMPONENT_TYPE_LABEL);
}
protected abstract Message<?> aggregateMessages(List<Message<?>> messages);
}

11
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -22,7 +22,6 @@ import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
@@ -50,6 +49,10 @@ import java.util.concurrent.locks.ReentrantLock;
*/
public class CorrelatingMessageHandler extends AbstractMessageHandler implements Lifecycle {
// TODO: need to support 'resequencer' as well
public static final String COMPONENT_TYPE_LABEL = "aggregator";
private MessageStore store = new SimpleMessageStore(100);
private final CorrelationStrategy correlationStrategy;
private final CompletionStrategy completionStrategy;
@@ -119,12 +122,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
this.sendPartialResultOnTimeout = sendPartialResultOnTimeout;
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
// TODO: need to support 'resequencer' as well
event.setComponentType("aggregator");
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
Object correlationKey = correlationStrategy.getCorrelationKey(message);

8
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java
View File

@@ -23,7 +23,6 @@ import java.util.SortedSet;
import java.util.TreeSet;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.util.CollectionUtils;
/**
@@ -54,7 +53,7 @@ import org.springframework.util.CollectionUtils;
*/
public class Resequencer extends AbstractMessageBarrierHandler<SortedSet<Message<?>>> {
private static final String COMPONENT_TYPE_LABEL = "resequencer";
public static final String COMPONENT_TYPE_LABEL = "resequencer";
private volatile boolean releasePartialSequences = true;
@@ -147,9 +146,4 @@ public class Resequencer extends AbstractMessageBarrierHandler<SortedSet<Message
return true;
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
event.setComponentType(COMPONENT_TYPE_LABEL);
}
}

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

@@ -32,6 +32,10 @@ public class EventDrivenConsumer extends AbstractEndpoint {
private final MessageHandler handler;
private volatile HandlerInvocationChain handlerInvocationChain;
private final Object initializationMonitor = new Object();
public EventDrivenConsumer(SubscribableChannel inputChannel, MessageHandler handler) {
Assert.notNull(inputChannel, "inputChannel must not be null");
@@ -44,12 +48,17 @@ public class EventDrivenConsumer extends AbstractEndpoint {
@Override // guarded by super#lifecycleLock
protected void doStart() {
this.inputChannel.subscribe(this.handler);
synchronized (this.initializationMonitor) {
if (this.handlerInvocationChain == null) {
this.handlerInvocationChain = new HandlerInvocationChain(this.handler, this.getBeanName());
}
}
this.inputChannel.subscribe(this.handlerInvocationChain);
}
@Override // guarded by super#lifecycleLock
protected void doStop() {
this.inputChannel.unsubscribe(this.handler);
this.inputChannel.unsubscribe(this.handlerInvocationChain);
}
}

80
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerInvocationChain.java Normal file
View File

@@ -0,0 +1,80 @@
/*
* Copyright 2002-2010 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.endpoint;
import java.lang.reflect.Field;
import org.springframework.core.Ordered;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.StringUtils;
/**
* A {@link MessageHandler} implementation that delegates to a target
* handler but also adds history information.
*
* @author Mark Fisher
* @since 2.0
*/
// TODO: add support for interceptors.
class HandlerInvocationChain implements MessageHandler, Ordered {
private final MessageHandler handler;
private final String componentName;
private final String componentType;
public HandlerInvocationChain(MessageHandler handler, String componentName) {
this.handler = handler;
this.componentName = componentName;
this.componentType = determineComponentTypeFromHandlerIfPossible(handler);
}
public int getOrder() {
return (this.handler instanceof Ordered) ?
((Ordered) this.handler).getOrder() : Ordered.LOWEST_PRECEDENCE;
}
public void handleMessage(Message<?> message) {
MessageHistoryEvent event = message.getHeaders().getHistory().addEvent(this.componentName);
if (this.componentType != null) {
event.setComponentType(this.componentType);
}
this.handler.handleMessage(message);
}
private static String determineComponentTypeFromHandlerIfPossible(MessageHandler handler) {
String type = null;
try {
Field componentTypeField = handler.getClass().getField("COMPONENT_TYPE_LABEL");
Object componentType = componentTypeField.get(null);
if (componentType instanceof String && StringUtils.hasText((String) componentType)) {
type = (String) componentType;
}
}
catch (Exception e) {
// no COMPONENT_TYPE_LABEL avaiable
}
return type;
}
}

21
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 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.
@@ -33,8 +33,14 @@ public class PollingConsumer extends AbstractPollingEndpoint {
private final MessageHandler handler;
private volatile MessageHandler handlerInvocationChain;
private volatile long receiveTimeout = 1000;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) {
Assert.notNull(inputChannel, "inputChannel must not be null");
@@ -48,6 +54,17 @@ public class PollingConsumer extends AbstractPollingEndpoint {
this.receiveTimeout = receiveTimeout;
}
@Override
protected void onInit() {
synchronized (this.initializationMonitor) {
if (!this.initialized) {
this.handlerInvocationChain = new HandlerInvocationChain(this.handler, this.getBeanName());
}
this.initialized = true;
}
super.onInit();
}
@Override
protected boolean doPoll() {
Message<?> message = (this.receiveTimeout >= 0)
@@ -56,7 +73,7 @@ public class PollingConsumer extends AbstractPollingEndpoint {
if (message == null) {
return false;
}
this.handler.handleMessage(message);
this.handlerInvocationChain.handleMessage(message);
return true;
}

8
org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java
View File

@@ -18,7 +18,6 @@ package org.springframework.integration.filter;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageRejectedException;
@@ -38,7 +37,7 @@ import org.springframework.util.Assert;
*/
public class MessageFilter extends AbstractReplyProducingMessageHandler {
private static final String COMPONENT_TYPE_LABEL = "filter";
public static final String COMPONENT_TYPE_LABEL = "filter";
private final MessageSelector selector;
@@ -103,9 +102,4 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler {
return null;
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent e) {
e.setComponentType(COMPONENT_TYPE_LABEL);
}
}

8
org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayInvokingMessageHandler.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.gateway;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
@@ -30,7 +29,7 @@ import org.springframework.util.Assert;
*/
public class GatewayInvokingMessageHandler extends AbstractReplyProducingMessageHandler {
private static final String COMPONENT_TYPE_LABEL = "gateway";
public static final String COMPONENT_TYPE_LABEL = "gateway";
private GenericSendAndRecieveGateway gateway;
@@ -50,9 +49,4 @@ public class GatewayInvokingMessageHandler extends AbstractReplyProducingMessage
return gateway.sendAndRecieve(requestMessage);
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
event.setComponentType(COMPONENT_TYPE_LABEL);
}
}

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

@@ -24,7 +24,6 @@ import org.springframework.integration.channel.ChannelResolutionException;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
@@ -59,8 +58,6 @@ public abstract class AbstractMessageHandler implements MessageHandler, Ordered
if (this.logger.isDebugEnabled()) {
this.logger.debug(this + " received message: " + message);
}
MessageHistoryEvent event = message.getHeaders().getHistory().addEvent(this.toString());
this.postProcessHistoryEvent(event);
try {
this.handleMessageInternal(message);
}
@@ -73,16 +70,6 @@ public abstract class AbstractMessageHandler implements MessageHandler, Ordered
}
}
/**
* Post process the history event. For example, this method is commonly overridden
* to set the 'componentType' label for the specific handler implementation. As a
* result, the "logical" name is available in MessageHistory events. Such a name
* should typically match the corresponding configuration element's name (in XML or
* Annotations), such as "router" or "splitter". By default this method is a no-op.
*/
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
}
protected abstract void handleMessageInternal(Message<?> message) throws Exception;
protected final MessageChannel resolveReplyChannel(Message<?> requestMessage,

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

@@ -17,7 +17,6 @@
package org.springframework.integration.handler;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.util.Assert;
/**
@@ -36,7 +35,7 @@ import org.springframework.util.Assert;
*/
public class BridgeHandler extends AbstractReplyProducingMessageHandler {
private static final String COMPONENT_TYPE_LABEL = "bridge";
public static final String COMPONENT_TYPE_LABEL = "bridge";
@Override
@@ -47,11 +46,6 @@ public class BridgeHandler extends AbstractReplyProducingMessageHandler {
return requestMessage;
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent e) {
e.setComponentType(COMPONENT_TYPE_LABEL);
}
private void verifyOutputChannel() {
Assert.state(super.getOutputChannel() != null, "Bridge handler requires an output channel");
}

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

@@ -20,7 +20,6 @@ import java.io.PrintWriter;
import java.io.StringWriter;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.util.StringUtils;
/**
@@ -36,7 +35,7 @@ public class LoggingHandler extends AbstractMessageHandler {
private static enum Level { FATAL, ERROR, WARN, INFO, DEBUG, TRACE }
private static final String COMPONENT_TYPE_LABEL = "logging-channel-adapter";
public static final String COMPONENT_TYPE_LABEL = "logging-channel-adapter";
private boolean shouldLogFullMessage;
@@ -109,9 +108,4 @@ public class LoggingHandler extends AbstractMessageHandler {
}
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent e) {
e.setComponentType(COMPONENT_TYPE_LABEL);
}
}

12
org.springframework.integration/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java
View File

@@ -49,6 +49,7 @@ import org.springframework.integration.annotation.Header;
import org.springframework.integration.annotation.Headers;
import org.springframework.integration.annotation.Payload;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.util.ClassUtils;
@@ -166,12 +167,21 @@ public class MethodInvokingMessageProcessor implements MessageProcessor {
List<HandlerMethod> candidates = this.findHandlerMethodsForMessage(message);
for (HandlerMethod candidate : candidates) {
try {
Object result = candidate.getExpression().getValue(this.evaluationContext, message);
Expression expression = candidate.getExpression();
Object result = expression.getValue(this.evaluationContext, message);
if (this.requiresReply) {
// TODO: remove this if SpEL is modified to throw an EvaluationException instead
// e.g. we can invoke getValue(this.evaluationContext, message, candidate.getReturnType);
Assert.notNull(result, "Expression evaluation result was null, but this processor requires a reply.");
}
MessageHistoryEvent event = message.getHeaders().getHistory().getCurrentEvent();
if (event != null) {
String typeName = org.springframework.util.ClassUtils.getShortNameAsProperty(this.targetObject.getClass());
event.setProperty("targetType", typeName);
if (candidate.method != null) {
event.setProperty("targetMethod", candidate.method.getName());
}
}
return result;
}
catch (EvaluationException e) {

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

@@ -20,7 +20,6 @@ import java.lang.reflect.Method;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.message.MessageHandlingException;
/**
@@ -28,7 +27,7 @@ import org.springframework.integration.message.MessageHandlingException;
*/
public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandler {
private static final String COMPONENT_TYPE_LABEL = "service-activator";
public static final String COMPONENT_TYPE_LABEL = "service-activator";
private final MethodInvokingMessageProcessor processor;
@@ -60,11 +59,6 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl
}
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
event.setComponentType(COMPONENT_TYPE_LABEL);
}
public String toString() {
return "ServiceActivator for [" + this.processor + "]";
}

8
org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
View File

@@ -21,7 +21,6 @@ import java.util.Collection;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
@@ -32,7 +31,7 @@ import org.springframework.integration.message.MessageDeliveryException;
*/
public abstract class AbstractMessageRouter extends AbstractMessageHandler {
private static final String COMPONENT_TYPE_LABEL = "router";
public static final String COMPONENT_TYPE_LABEL = "router";
private volatile MessageChannel defaultOutputChannel;
@@ -94,11 +93,6 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
}
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
event.setComponentType(COMPONENT_TYPE_LABEL);
}
/**
* Subclasses must implement this method to return the target channels for
* a given Message.

18
org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java
View File

@@ -16,20 +16,25 @@
package org.springframework.integration.router;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
import java.util.*;
/**
* <pre>
* &lt;recipient-list-router id="simpleRouter" input-channel="routingChannelA"&gt;
@@ -60,7 +65,7 @@ import java.util.*;
*/
public class RecipientListRouter extends AbstractMessageHandler implements InitializingBean {
private static final String COMPONENT_TYPE_LABEL = "recipient-list-router";
public static final String COMPONENT_TYPE_LABEL = "recipient-list-router";
private volatile boolean ignoreSendFailures;
@@ -159,9 +164,4 @@ public class RecipientListRouter extends AbstractMessageHandler implements Initi
}
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
event.setComponentType(COMPONENT_TYPE_LABEL);
}
}

8
org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
View File

@@ -24,7 +24,6 @@ import java.util.UUID;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.MessageBuilder;
@@ -35,7 +34,7 @@ import org.springframework.integration.message.MessageBuilder;
*/
public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
private static final String COMPONENT_TYPE_LABEL = "splitter";
public static final String COMPONENT_TYPE_LABEL = "splitter";
@Override
@@ -70,11 +69,6 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
return messageBuilders;
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
event.setComponentType(COMPONENT_TYPE_LABEL);
}
@SuppressWarnings("unchecked")
private MessageBuilder createBuilder(Object item, Object correlationId, int sequenceNumber, int sequenceSize) {
MessageBuilder builder = (item instanceof Message) ?

8
org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.transformer;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHistoryEvent;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.util.Assert;
@@ -30,7 +29,7 @@ import org.springframework.util.Assert;
*/
public class MessageTransformingHandler extends AbstractReplyProducingMessageHandler {
private static final String COMPONENT_TYPE_LABEL = "transformer";
public static final String COMPONENT_TYPE_LABEL = "transformer";
private final Transformer transformer;
@@ -59,9 +58,4 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan
}
}
@Override
protected void postProcessHistoryEvent(MessageHistoryEvent event) {
event.setComponentType(COMPONENT_TYPE_LABEL);
}
}