diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 540617a2e7..0e7db12776 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -25,6 +25,8 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.BeanNameAware; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.core.MessageHistory.ComponentType; +import org.springframework.util.Assert; /** * Base class for {@link MessageChannel} implementations providing common @@ -109,6 +111,8 @@ public abstract class AbstractMessageChannel implements MessageChannel, BeanName * time or the sending thread is interrupted. */ public final boolean send(Message message, long timeout) { + Assert.notNull(message, "message must not be null"); + message.getHeaders().getHistory().add(ComponentType.channel, this.getName()); message = this.interceptors.preSend(message, this); if (message == null) { return false; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHeaders.java b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHeaders.java index e28eb7cfaf..331a145dd9 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHeaders.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHeaders.java @@ -54,6 +54,8 @@ public final class MessageHeaders implements Map, Serializable { public static final String ERROR_CHANNEL = PREFIX + "errorChannel"; + public static final String HISTORY = "history"; + public static final String EXPIRATION_DATE = PREFIX + "expirationDate"; public static final String PRIORITY = PREFIX + "priority"; @@ -76,6 +78,9 @@ public final class MessageHeaders implements Map, Serializable { if (this.headers.get(TIMESTAMP) == null) { this.headers.put(TIMESTAMP, new Long(System.currentTimeMillis())); } + if (this.headers.get(HISTORY) == null) { + this.headers.put(HISTORY, new MessageHistory()); + } } @@ -87,6 +92,10 @@ public final class MessageHeaders implements Map, Serializable { return this.get(TIMESTAMP, Long.class); } + public MessageHistory getHistory() { + return this.get(HISTORY, MessageHistory.class); + } + public Long getExpirationDate() { return this.get(EXPIRATION_DATE, Long.class); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHistory.java b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHistory.java new file mode 100644 index 0000000000..792d986320 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHistory.java @@ -0,0 +1,85 @@ +/* + * Copyright 2002-2009 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.io.Serializable; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class MessageHistory implements Iterable, Serializable { + + private final List events = new CopyOnWriteArrayList(); + + + public void add(ComponentType componentType, String componentName) { + this.events.add(new Event(componentType, componentName)); + } + + public Iterator iterator() { + return Collections.unmodifiableList(this.events).iterator(); + } + + public String toString() { + return this.events.toString(); + } + + + public static enum ComponentType { + channel, endpoint; + } + + + 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() { + return "name=" + this.componentName + ";type=" + this.componentType + ";timestamp=" + this.timestamp; + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 6110efc1b8..569111328f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -23,6 +23,7 @@ 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; @@ -58,6 +59,7 @@ 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()); try { this.handleMessageInternal(message); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java index 4d83b89381..00a3c46057 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MixedDispatcherConfigurationScenarioTests.java @@ -47,6 +47,7 @@ import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrateg import org.springframework.integration.dispatcher.UnicastingDispatcher; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageRejectedException; +import org.springframework.integration.message.StringMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; /** @@ -77,8 +78,8 @@ public class MixedDispatcherConfigurationScenarioTests { @Mock private MessageHandler handlerC; - @Mock - private Message message; + private Message message = new StringMessage("test"); + @Before public void initialize() throws Exception { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/ArgumentArrayMessageMapperFromMessageTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/ArgumentArrayMessageMapperFromMessageTests.java index 02a8737609..bb2d37ce40 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/ArgumentArrayMessageMapperFromMessageTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/ArgumentArrayMessageMapperFromMessageTests.java @@ -139,8 +139,8 @@ public class ArgumentArrayMessageMapperFromMessageTests { .setHeader("prop1", "foo").setHeader("prop2", "bar").build(); Object[] args = mapper.fromMessage(message); Map result = (Map) args[0]; - //Map also contains id and timestamp - assertEquals(4, result.size()); + //Map also contains id, timestamp, and history + assertEquals(5, result.size()); assertEquals("foo", result.get("prop1")); assertEquals("bar", result.get("prop2")); assertEquals("test", args[1]);