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

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 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.
@@ -20,6 +20,7 @@ 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;
@@ -47,6 +48,9 @@ import org.springframework.util.CollectionUtils;
 public abstract class AbstractMessageAggregator extends
 		AbstractMessageBarrierHandler<List<Message<?>>> {
 
+	private static final String COMPONENT_TYPE_LABEL = "aggregator";
+
+
 	private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
 
 	
@@ -85,6 +89,11 @@ public abstract class AbstractMessageAggregator extends
 		}
 	}
 
+	@Override
+	protected void postProcessHistoryEvent(MessageHistoryEvent event) {
+		event.setComponentType(COMPONENT_TYPE_LABEL);
+	}
+
     protected abstract Message<?> aggregateMessages(List<Message<?>> messages);
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java

@@ -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.
@@ -22,6 +22,7 @@ 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;
@@ -118,6 +119,12 @@ 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);

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 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.
@@ -23,6 +23,7 @@ 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;
 
 /**
@@ -53,10 +54,13 @@ import org.springframework.util.CollectionUtils;
  */
 public class Resequencer extends AbstractMessageBarrierHandler<SortedSet<Message<?>>> {
 
+	private static final String COMPONENT_TYPE_LABEL = "resequencer";
+
+
 	private volatile boolean releasePartialSequences = true;
 	
 	private static final String LAST_RELEASED_SEQUENCE_NUMBER = "last.released.sequence.number";
-	
+
 
 	public void setReleasePartialSequences(boolean releasePartialSequences) {
 		this.releasePartialSequences = releasePartialSequences;
@@ -143,4 +147,9 @@ public class Resequencer extends AbstractMessageBarrierHandler<SortedSet<Message
 		return true;
 	}
 
+	@Override
+	protected void postProcessHistoryEvent(MessageHistoryEvent event) {
+		event.setComponentType(COMPONENT_TYPE_LABEL);
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java

@@ -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.
@@ -29,7 +29,6 @@ import org.springframework.core.convert.ConversionService;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessageChannel;
 import org.springframework.integration.core.MessagingException;
-import org.springframework.integration.core.MessageHistory.ComponentType;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.util.Assert;
@@ -171,7 +170,7 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanFact
 		Assert.notNull(message, "message must not be null");
 		Assert.notNull(message.getPayload(), "message payload must not be null");
 		message = this.convertPayloadIfNecessary(message);
-		message.getHeaders().getHistory().add(ComponentType.channel, this.getName());
+		message.getHeaders().getHistory().addEvent(this.getName()).setComponentType("channel");
 		message = this.interceptors.preSend(message, this);
 		if (message == null) {
 			return false;

org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHistory.java

@@ -17,79 +17,59 @@
 package org.springframework.integration.core;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Date;
 import java.util.Iterator;
 import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 /**
  * @author Mark Fisher
  * @since 2.0
  */
-public class MessageHistory implements Iterable<MessageHistory.Event>, Serializable {
+public class MessageHistory implements Iterable<MessageHistoryEvent>, Serializable {
 
-	private final List<Event> events = new CopyOnWriteArrayList<Event>();
+	private final List<MessageHistoryEvent> events = new ArrayList<MessageHistoryEvent>();
+
+	private final ReadWriteLock lock = new ReentrantReadWriteLock();
 
 
-	public void add(ComponentType componentType, String componentName) {
-		this.events.add(new Event(componentType, componentName));
+	public MessageHistoryEvent addEvent(String componentName) {
+		try {
+			this.lock.writeLock().lock();
+			MessageHistoryEvent event = new MessageHistoryEvent(componentName);
+			this.events.add(event);
+			return event;
+		}
+		finally {
+			this.lock.writeLock().unlock();
+		}
 	}
 
-	public Iterator<Event> iterator() {
-		return Collections.unmodifiableList(this.events).iterator();
+	public MessageHistoryEvent getCurrentEvent() {
+		try {
+			this.lock.readLock().lock();
+			int size = this.events.size();
+			return (size > 0) ? this.events.get(size - 1) : null;
+		}
+		finally {
+			this.lock.readLock().unlock();
+		}
+	}
+
+	public Iterator<MessageHistoryEvent> iterator() {
+		try {
+			this.lock.readLock().lock();
+			return Collections.unmodifiableList(this.events).iterator();
+		}
+		finally {
+			this.lock.readLock().unlock();
+		}
 	}
 
 	public String toString() {
 		return this.events.toString();
 	}
 
-
-	public static enum ComponentType {
-		channel, endpoint, gateway;
-	}
-
-
-	public static class Event implements Serializable {
-
-		private final ComponentType componentType;
-
-		private final String componentName;
-
-		private final long timestamp;
-
-
-		public Event(ComponentType componentType, String componentName) {
-			this.componentType = componentType;
-			this.componentName = componentName;
-			this.timestamp = System.currentTimeMillis();
-		}
-
-
-		public ComponentType getComponentType() {
-			return this.componentType;
-		}
-
-		public String getComponentName() {
-			return this.componentName;
-		}
-
-		public long getTimestamp() {
-			return this.timestamp;
-		}
-
-		public String toString() {
-			StringBuilder sb = new StringBuilder("[");
-			if (this.componentName != null) {
-				sb.append("{name=" + this.componentName + "}");
-			}
-			if (this.componentType != null) {
-				sb.append("{type=" + this.componentType + "}");
-			}
-			sb.append("{timestamp=" + new Date(this.timestamp) + "}");
-			sb.append("]");
-			return sb.toString();
-		}
-	}
-
 }

org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHistoryEvent.java

@@ -0,0 +1,93 @@
+/*
+ * 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.core;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class MessageHistoryEvent {
+
+	public static final String COMPONENT_NAME = "componentName";
+
+	public static final String COMPONENT_TYPE = "componentType";
+
+	public static final String TIMESTAMP = "timestamp";
+
+
+	private final Map<String, Object> properties = new HashMap<String, Object>();
+
+
+	public MessageHistoryEvent(String componentName) {
+		this.setProperty(COMPONENT_NAME, componentName);
+		this.setProperty(TIMESTAMP, System.currentTimeMillis());
+	}
+
+
+	public MessageHistoryEvent setComponentType(String componentType) {
+		this.setProperty(COMPONENT_TYPE, componentType);
+		return this;
+	}
+
+	public String getComponentType() {
+		return this.getProperty(COMPONENT_TYPE, String.class);
+	}
+
+	public String getComponentName() {
+		return this.getProperty(COMPONENT_NAME, String.class);
+	}
+
+	public long getTimestamp() {
+		return this.getProperty(TIMESTAMP, long.class);
+	}
+
+	public MessageHistoryEvent setProperty(String key, String value) {
+		this.properties.put(key, value);
+		return this;
+	}
+
+	public MessageHistoryEvent setProperty(String key, Number value) {
+		this.properties.put(key, value);
+		return this;
+	}
+
+	public MessageHistoryEvent setProperty(String key, Boolean value) {
+		this.properties.put(key, value);
+		return this;
+	}
+
+	public Object getProperty(String key) {
+		return this.properties.get(key);
+	}
+
+	@SuppressWarnings("unchecked")
+	public <T> T getProperty(String key, Class<T> type) {
+		Object value = this.properties.get(key);
+		if (value != null && type.isAssignableFrom(value.getClass())) {
+			return (T) value;
+		}
+		return null;
+	}
+
+	public String toString() {
+		return this.properties.toString();
+	}
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 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.
@@ -19,6 +19,7 @@ package org.springframework.integration.endpoint;
 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.util.Assert;
 
 /**
@@ -48,7 +49,16 @@ public abstract class MessageProducerSupport extends AbstractEndpoint {
 	}
 
 	protected boolean sendMessage(Message<?> message) {
+		String componentName = this.getBeanName();
+		if (componentName != null) {
+			MessageHistoryEvent event = message.getHeaders().getHistory().addEvent(componentName);
+			this.postProcessHistoryEvent(event);
+		}
 		return this.channelTemplate.send(message, this.outputChannel);
 	}
 
+	protected void postProcessHistoryEvent(MessageHistoryEvent event) {
+		event.setComponentType("producer");
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java

@@ -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.
@@ -18,6 +18,7 @@ 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;
@@ -37,6 +38,9 @@ import org.springframework.util.Assert;
  */
 public class MessageFilter extends AbstractReplyProducingMessageHandler {
 
+	private static final String COMPONENT_TYPE_LABEL = "filter";
+
+
 	private final MessageSelector selector;
 
 	private volatile boolean throwExceptionOnRejection;
@@ -99,4 +103,9 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler {
 		return null;
 	}
 
+	@Override
+	protected void postProcessHistoryEvent(MessageHistoryEvent e) {
+		e.setComponentType(COMPONENT_TYPE_LABEL);
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayInvokingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 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.
@@ -13,9 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 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;
@@ -26,14 +28,17 @@ import org.springframework.util.Assert;
  * @author Oleg Zhurakousky
  * @since 2.0
  */
-public class GatewayInvokingMessageHandler extends
-		AbstractReplyProducingMessageHandler {
+public class GatewayInvokingMessageHandler extends AbstractReplyProducingMessageHandler {
+
+	private static final String COMPONENT_TYPE_LABEL = "gateway";
+
+
 	private GenericSendAndRecieveGateway gateway;
+
 	/**
-	 * 
 	 * @param gateway
 	 */
-	public GatewayInvokingMessageHandler(GenericSendAndRecieveGateway gateway){
+	public GatewayInvokingMessageHandler(GenericSendAndRecieveGateway gateway) {
 		Assert.notNull(gateway, "gateway must not be null");
 		this.gateway = gateway;
 	}
@@ -44,4 +49,10 @@ public class GatewayInvokingMessageHandler extends
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		return gateway.sendAndRecieve(requestMessage);
 	}
+
+	@Override
+	protected void postProcessHistoryEvent(MessageHistoryEvent event) {
+		event.setComponentType(COMPONENT_TYPE_LABEL);
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java

@@ -234,7 +234,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Factory
 
 	private MessagingGateway createGatewayForMethod(Method method) throws Exception {
 		SimpleMessagingGateway gateway = new SimpleMessagingGateway(
-				new ArgumentArrayMessageMapper(method), new SimpleMessageMapper());
+				new ArgumentArrayMessageMapper(method, this.getBeanName()), new SimpleMessageMapper());
 		if (this.getTaskScheduler() != null) {
 			gateway.setTaskScheduler(this.getTaskScheduler());
 		}

org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java

@@ -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.
@@ -20,14 +20,14 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.core.Ordered;
-import org.springframework.integration.core.Message;
-import org.springframework.integration.core.MessagingException;
-import org.springframework.integration.core.MessageChannel;
-import org.springframework.integration.core.MessageHistory.ComponentType;
-import org.springframework.integration.message.MessageHandler;
-import org.springframework.integration.message.MessageHandlingException;
 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;
 import org.springframework.util.Assert;
 
 /**
@@ -59,7 +59,8 @@ public abstract class AbstractMessageHandler implements MessageHandler, Ordered
         if (this.logger.isDebugEnabled()) {
             this.logger.debug(this + " received message: " + message);
         }
-        message.getHeaders().getHistory().add(ComponentType.endpoint, this.toString());
+        MessageHistoryEvent event = message.getHeaders().getHistory().addEvent(this.toString());
+        this.postProcessHistoryEvent(event);
         try {
             this.handleMessageInternal(message);
         }
@@ -72,6 +73,16 @@ 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,

org.springframework.integration/src/main/java/org/springframework/integration/handler/ArgumentArrayMessageMapper.java

@@ -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.
@@ -20,13 +20,11 @@ import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
 import java.util.Date;
 import java.util.HashMap;
-import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 
-import org.springframework.context.expression.MapAccessor;
 import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ParameterNameDiscoverer;
@@ -34,21 +32,13 @@ import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.convert.converter.ConverterRegistry;
 import org.springframework.core.convert.support.ConversionServiceFactory;
-import org.springframework.expression.Expression;
-import org.springframework.expression.ExpressionParser;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
-import org.springframework.expression.spel.support.StandardEvaluationContext;
-import org.springframework.expression.spel.support.StandardTypeConverter;
 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.MessagingException;
-import org.springframework.integration.core.MessageHistory.ComponentType;
 import org.springframework.integration.message.InboundMessageMapper;
 import org.springframework.integration.message.MessageBuilder;
-import org.springframework.integration.message.MessageHandlingException;
-import org.springframework.integration.message.OutboundMessageMapper;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -109,7 +99,7 @@ import org.springframework.util.StringUtils;
  * @author Oleg Zhurakousky
  * @since 2.0
  */
-public class ArgumentArrayMessageMapper implements InboundMessageMapper<Object[]>, OutboundMessageMapper<Object[]> {
+public class ArgumentArrayMessageMapper implements InboundMessageMapper<Object[]> {
 
 	private static ConversionService conversionService;
 
@@ -128,26 +118,21 @@ public class ArgumentArrayMessageMapper implements InboundMessageMapper<Object[]
 	}
 
 
-	private final ExpressionParser expressionParser = new SpelExpressionParser();
-
 	private final Method method;
 
+	private final String gatewayName;
+
 	private final List<MethodParameter> parameterList;
 
 
-	public ArgumentArrayMessageMapper(Method method) {
+	public ArgumentArrayMessageMapper(Method method, String gatewayName) {
 		Assert.notNull(method, "method must not be null");
+		Assert.notNull(gatewayName, "gatewayName must not be null");
 		this.method = method;
+		this.gatewayName = gatewayName;
 		this.parameterList = this.getMethodParameterList(method);
 	}
 
-
-	public Object[] fromMessage(Message<?> message) {
-		Assert.notNull(message, "cannot map a null Message");
-		this.validateMessageMapppings(message);
-		return this.mapMessageToArguments(message);
-	}
-
 	public Message<?> toMessage(Object[] arguments) {
 		Assert.notNull(arguments, "cannot map null arguments to Message");
 		if (arguments.length != this.parameterList.size()) {
@@ -157,60 +142,14 @@ public class ArgumentArrayMessageMapper implements InboundMessageMapper<Object[]
 		}
 		Message<?> message = this.mapArgumentsToMessage(arguments);
 		if (message != null) {
-			message.getHeaders().getHistory().add(ComponentType.gateway, this.method.getName());
+			message.getHeaders().getHistory().addEvent(this.gatewayName)
+					.setComponentType("gateway")
+					// TODO: add METHOD_NAME key for props?
+					.setProperty("method", this.method.getName());
 		}
 		return message;
 	}
 
-	private Object[] mapMessageToArguments(Message<?> message) {
-		final Map<String, Object> messageArgumentsMap = new LinkedHashMap<String, Object>();
-		for (MethodParameter methodParameter : this.parameterList) {
-			String parameterName = methodParameter.getParameterName();
-			Annotation mappingAnnotation = null;
-			Object value = null;
-			mappingAnnotation = this.findMappingAnnotation(methodParameter.getParameterAnnotations());
-			if (mappingAnnotation == null) {
-				String[] expressions = null;
-				if ("headers".equals(parameterName) || "payload".equals(parameterName)) {
-					expressions = new String[] { parameterName };
-				}
-				else if ("message".equals(parameterName)) {
-					// just in case 'parameterName' is 'message' but type is not Message
-					expressions = new String[] { "#this", "payload" };
-				}
-				else {
-					expressions = new String[] { "payload." + parameterName, "headers." + parameterName, "payload", "headers", "#this" };
-				}
-				value = this.getValueFromMessageBasedOnEL(message, methodParameter.getParameterType(), false, expressions);
-			}
-			else {
-				if (mappingAnnotation.annotationType().equals(Header.class)) {
-					value = this.retrieveHeaderFromMessage((Header) mappingAnnotation, message, methodParameter)[1];
-				}
-				else if (mappingAnnotation.annotationType().equals(Headers.class)) {
-					String[] expressions = new String[] {"headers"};
-					value = this.getValueFromMessageBasedOnEL(message, methodParameter.getParameterType(), false, expressions);
-				}
-				else if (mappingAnnotation.annotationType().isAssignableFrom(Payload.class)) {
-					String[] expressions = null;
-					String payloadExpression = ((Payload) mappingAnnotation).value();
-					if (payloadExpression.length() == 0) {
-						expressions = new String[] { "payload" };
-					}
-					else {
-						expressions = new String[] { "payload." + payloadExpression };
-					}
-					value = this.getValueFromMessageBasedOnEL(message, methodParameter.getParameterType(), true, expressions);
-				}
-				else {
-					throw new IllegalArgumentException("unsupported mapping annotation: " + mappingAnnotation);
-				}
-			}
-			messageArgumentsMap.put(methodParameter.getParameterIndex() + ":" + parameterName, value);		
-		}
-		return messageArgumentsMap.values().toArray();
-	}
-
 	@SuppressWarnings("unchecked")
 	private Message<?> mapArgumentsToMessage(Object[] arguments) {
 		Object messageOrPayload = null;
@@ -312,18 +251,6 @@ public class ArgumentArrayMessageMapper implements InboundMessageMapper<Object[]
 				"found more than one on method [" + methodParameter.getMethod() + "]");
 	}
 
-	private Object[] retrieveHeaderFromMessage(Header headerAnnotation, Message<?> message, MethodParameter methodParameter) {
-		Object headerValue = null;
-		String headerName = this.determineHeaderName(headerAnnotation, methodParameter);
-		if (message != null) {
-			headerValue = this.getValueFromMessageBasedOnEL(message, methodParameter.getParameterType(), false, "headers." + headerName);
-		}
-		if (headerAnnotation.required() && headerValue == null) {
-			throw new MessageHandlingException(message, "Message is missing required header: '" + headerName + "'");
-		}
-		return new Object[] {headerName, headerValue};
-	}
-
 	private String determineHeaderName(Header headerAnnotation, MethodParameter methodParameter) {
 		String valueAttribute = headerAnnotation.value();
 		String headerName = StringUtils.hasText(valueAttribute) ? valueAttribute : methodParameter.getParameterName();
@@ -332,34 +259,6 @@ public class ArgumentArrayMessageMapper implements InboundMessageMapper<Object[]
 		return headerName;
 	}
 
-	@SuppressWarnings("unchecked") 
-	private Object getValueFromMessageBasedOnEL(Message message, Class targetType, boolean rethrowException, String... expressions) {
-		Object value = null;
-		for (String expression : expressions) {
-			StandardEvaluationContext context = new StandardEvaluationContext(message);
-			context.setTypeConverter(new StandardTypeConverter(conversionService));
-			try { 
-				Expression exp = expressionParser.parseExpression(expression); 
-				context.addPropertyAccessor(new MapAccessor());	
-				value = exp.getValue(context);
-				if (value != null && conversionService.canConvert(value.getClass(), targetType)) {
-					// to accommodate Map->Properties conversion
-					value = expression.equals("headers") ? conversionService.convert(value, targetType) : value;
-					break;
-				}
-				else {
-					value = null;
-				}
-			}
-			catch (Throwable e) {
-				if (rethrowException) {
-					throw new MessageHandlingException(message, e);
-				}
-			}
-		}
-		return value;
-	}
-
 	private List<MethodParameter> getMethodParameterList(Method method) {
 		List<MethodParameter> parameterList = new LinkedList<MethodParameter>();
 		ParameterNameDiscoverer parameterNameDiscoverer = new LocalVariableTableParameterNameDiscoverer(); 

org.springframework.integration/src/main/java/org/springframework/integration/handler/BridgeHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 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.
@@ -17,6 +17,7 @@
 package org.springframework.integration.handler;
 
 import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessageHistoryEvent;
 import org.springframework.util.Assert;
 
 /**
@@ -35,6 +36,9 @@ import org.springframework.util.Assert;
  */
 public class BridgeHandler extends AbstractReplyProducingMessageHandler {
 
+	private static final String COMPONENT_TYPE_LABEL = "bridge";
+
+
 	@Override
 	protected Object handleRequestMessage(Message<?> requestMessage) {
 		if (requestMessage.getHeaders().getReplyChannel() == null) {
@@ -43,6 +47,11 @@ 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");
 	}

org.springframework.integration/src/main/java/org/springframework/integration/handler/LoggingHandler.java

@@ -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.
@@ -20,6 +20,7 @@ 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;
 
 /**
@@ -33,7 +34,9 @@ import org.springframework.util.StringUtils;
  */
 public class LoggingHandler extends AbstractMessageHandler {
 
-	private static enum Level { FATAL, ERROR, WARN, INFO, DEBUG, TRACE };
+	private static enum Level { FATAL, ERROR, WARN, INFO, DEBUG, TRACE }
+
+	private static final String COMPONENT_TYPE_LABEL = "logging-channel-adapter";
 
 
 	private boolean shouldLogFullMessage;
@@ -106,4 +109,9 @@ public class LoggingHandler extends AbstractMessageHandler {
 		}
 	}
 
+	@Override
+	protected void postProcessHistoryEvent(MessageHistoryEvent e) {
+		e.setComponentType(COMPONENT_TYPE_LABEL);
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java

@@ -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.
@@ -20,6 +20,7 @@ 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;
 
 /**
@@ -27,6 +28,9 @@ import org.springframework.integration.message.MessageHandlingException;
  */
 public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandler {
 
+	private static final String COMPONENT_TYPE_LABEL = "service-activator";
+
+
 	private final MethodInvokingMessageProcessor processor;
 
 
@@ -56,6 +60,11 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl
 		}
 	}
 
+	@Override
+	protected void postProcessHistoryEvent(MessageHistoryEvent event) {
+		event.setComponentType(COMPONENT_TYPE_LABEL);
+	}
+
 	public String toString() {
 		return "ServiceActivator for [" + this.processor + "]";
 	}

org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 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.
@@ -21,6 +21,7 @@ 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;
 
@@ -31,6 +32,9 @@ import org.springframework.integration.message.MessageDeliveryException;
  */
 public abstract class AbstractMessageRouter extends AbstractMessageHandler {
 
+	private static final String COMPONENT_TYPE_LABEL = "router";
+
+
 	private volatile MessageChannel defaultOutputChannel;
 
 	private volatile boolean resolutionRequired;
@@ -90,6 +94,11 @@ 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.

org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java

@@ -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.
@@ -21,6 +21,7 @@ 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;
@@ -59,7 +60,10 @@ import java.util.*;
  */
 public class RecipientListRouter extends AbstractMessageHandler implements InitializingBean {
 
-    private volatile boolean ignoreSendFailures;
+    private static final String COMPONENT_TYPE_LABEL = "recipient-list-router";
+
+
+	private volatile boolean ignoreSendFailures;
 
     private volatile boolean applySequence;
 
@@ -155,4 +159,9 @@ public class RecipientListRouter extends AbstractMessageHandler implements Initi
         }
     }
 
+	@Override
+	protected void postProcessHistoryEvent(MessageHistoryEvent event) {
+		event.setComponentType(COMPONENT_TYPE_LABEL);
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java

@@ -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.
@@ -24,6 +24,7 @@ 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;
 
@@ -34,6 +35,9 @@ import org.springframework.integration.message.MessageBuilder;
  */
 public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
 
+	private static final String COMPONENT_TYPE_LABEL = "splitter";
+
+
 	@Override
 	@SuppressWarnings("unchecked")
 	protected final Object handleRequestMessage(Message<?> message) {
@@ -66,6 +70,11 @@ 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) ?

org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 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.
@@ -17,6 +17,7 @@
 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;
 
@@ -29,6 +30,9 @@ import org.springframework.util.Assert;
  */
 public class MessageTransformingHandler extends AbstractReplyProducingMessageHandler {
 
+	private static final String COMPONENT_TYPE_LABEL = "transformer";
+
+
 	private final Transformer transformer;
 
 
@@ -55,4 +59,9 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan
 		}
 	}
 
+	@Override
+	protected void postProcessHistoryEvent(MessageHistoryEvent event) {
+		event.setComponentType(COMPONENT_TYPE_LABEL);
+	}
+
 }