INT-87 initial commit of MessageHistory support (work in progress)
This commit is contained in:
@@ -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;
|
||||
|
||||
@@ -54,6 +54,8 @@ public final class MessageHeaders implements Map<String, Object>, 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<String, Object>, 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<String, Object>, 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);
|
||||
}
|
||||
|
||||
@@ -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<MessageHistory.Event>, Serializable {
|
||||
|
||||
private final List<Event> events = new CopyOnWriteArrayList<Event>();
|
||||
|
||||
|
||||
public void add(ComponentType componentType, String componentName) {
|
||||
this.events.add(new Event(componentType, componentName));
|
||||
}
|
||||
|
||||
public Iterator<Event> 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;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user