diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index efe82457db..67728973b6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -23,13 +23,13 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.core.OrderComparator; import org.springframework.core.convert.ConversionService; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessagingException; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.util.Assert; @@ -42,6 +42,7 @@ import org.springframework.util.StringUtils; * of any {@link ChannelInterceptor ChannelInterceptors}. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public abstract class AbstractMessageChannel extends IntegrationObjectSupport implements MessageChannel { @@ -54,10 +55,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im private volatile Class[] datatypes = new Class[] { Object.class }; private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); - - - @Override - public String getComponentType() { + + public String getComponentType(){ return "channel"; } @@ -159,10 +158,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im * time or the sending thread is interrupted. */ public final boolean send(Message message, long timeout) { + MessageHistory.writeMessageHistory(message, this, this.getBeanFactory()); 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().addEvent(this); message = this.interceptors.preSend(message, this); if (message == null) { return false; @@ -287,5 +286,4 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im return message; } } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 134488ca37..391447e55a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -26,6 +26,7 @@ import org.springframework.util.ErrorHandler; * A channel that sends Messages to each of its subscribers. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public class PublishSubscribeChannel extends AbstractSubscribableChannel { @@ -39,7 +40,9 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { private volatile boolean applySequence; - + public String getComponentType(){ + return "pub-sub-channel"; + } /** * Create a PublishSubscribeChannel that will use an {@link Executor} * to invoke the handlers. If this is null, each invocation will occur in diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java index 8edcd3c7f0..98bea89804 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java @@ -34,12 +34,16 @@ import org.springframework.util.Assert; * {@link RendezvousChannel}. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public class QueueChannel extends AbstractPollableChannel { private final BlockingQueue> queue; + public String getComponentType(){ + return "queue-channel"; + } /** * Create a channel with the specified queue. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 48db714d3d..a87b15883f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -26,6 +26,7 @@ import org.springframework.context.SmartLifecycle; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.SubscribableChannel; import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; @@ -36,6 +37,7 @@ import org.springframework.util.Assert; /** * @author Mark Fisher + * @author Oleg Zhurakousky */ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, InitializingBean, SmartLifecycle { @@ -94,6 +96,9 @@ public class ConsumerEndpointFactoryBean public void afterPropertiesSet() throws Exception { this.initializeEndpoint(); + if (this.handler instanceof IntegrationObjectSupport){ + ((IntegrationObjectSupport)this.handler).setComponentName(this.beanName); + } } public boolean isSingleton() { @@ -149,7 +154,7 @@ public class ConsumerEndpointFactoryBean throw new IllegalArgumentException( "unsupported channel type: [" + channel.getClass() + "]"); } - this.endpoint.setBeanName(this.beanName); + //this.endpoint.setBeanName(this.beanName); this.endpoint.setBeanFactory(this.beanFactory); this.endpoint.setAutoStartup(this.autoStartup); this.endpoint.afterPropertiesSet(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index 526756ada7..3cc5ddbce9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java @@ -28,6 +28,7 @@ import org.springframework.integration.channel.BeanFactoryChannelResolver; import org.springframework.integration.channel.ChannelResolver; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * A base class that provides convenient access to the bean factory as @@ -47,6 +48,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo protected final Log logger = LogFactory.getLog(getClass()); private volatile String beanName; + + private volatile String componentName; private volatile BeanFactory beanFactory; @@ -61,11 +64,21 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo this.beanName = beanName; } - - public final String getComponentName() { - return this.beanName; + /** + * Will return the name of this component identified by {@link this#componentName} field. + * If {@link this#componentName} was not set this method will default to the 'beanName' of this component; + */ + public final String getComponentName() { + return StringUtils.hasText(this.componentName) ? this.componentName : this.beanName; + } + /** + * Sets the name of this component. + * + * @param componentName + */ + public void setComponentName(String componentName) { + this.componentName = componentName; } - /** * Subclasses may implement this method to provide component type information. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java index 9b04d0e045..d6093c6fd2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java @@ -25,6 +25,7 @@ import org.springframework.util.Assert; * to a {@link SubscribableChannel}. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public class EventDrivenConsumer extends AbstractEndpoint { @@ -32,11 +33,6 @@ 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"); Assert.notNull(handler, "handler must not be null"); @@ -45,20 +41,13 @@ public class EventDrivenConsumer extends AbstractEndpoint { this.setPhase(Integer.MIN_VALUE); } - - @Override // guarded by super#lifecycleLock + @Override // protected void doStart() { - synchronized (this.initializationMonitor) { - if (this.handlerInvocationChain == null) { - this.handlerInvocationChain = new HandlerInvocationChain(this.handler, this.getComponentName()); - } - } - this.inputChannel.subscribe(this.handlerInvocationChain); + this.inputChannel.subscribe(this.handler); } - @Override // guarded by super#lifecycleLock + @Override protected void doStop() { - this.inputChannel.unsubscribe(this.handlerInvocationChain); + this.inputChannel.unsubscribe(this.handler); } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/HandlerInvocationChain.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/HandlerInvocationChain.java deleted file mode 100644 index 41435d39fe..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/HandlerInvocationChain.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * 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 org.springframework.core.Ordered; -import org.springframework.integration.context.IntegrationObjectSupport; -import org.springframework.integration.context.NamedComponent; -import org.springframework.integration.core.Message; -import org.springframework.integration.message.MessageHandler; - -/** - * 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 EndpointNamedComponent namedComponent; - - - public HandlerInvocationChain(MessageHandler handler, String endpointName) { - this.handler = handler; - String handlerType = (handler instanceof IntegrationObjectSupport) ? - ((IntegrationObjectSupport) this.handler).getComponentType() : null; - this.namedComponent = (endpointName != null) ? - new EndpointNamedComponent(endpointName, handlerType) : null; - } - - - public int getOrder() { - return (this.handler instanceof Ordered) ? - ((Ordered) this.handler).getOrder() : Ordered.LOWEST_PRECEDENCE; - } - - public void handleMessage(Message message) { - if (message != null && this.namedComponent != null) { - message.getHeaders().getHistory().addEvent(this.namedComponent); - } - this.handler.handleMessage(message); - } - - - private static class EndpointNamedComponent implements NamedComponent { - - private final String componentName; - - private final String componentType; - - private EndpointNamedComponent(String componentName, String componentType) { - this.componentName = componentName; - this.componentType = componentType; - } - - public String getComponentName() { - return this.componentName; - } - - public String getComponentType() { - return this.componentType; - } - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index d1faf73b87..59c371f045 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -26,6 +26,7 @@ import org.springframework.util.Assert; * to a {@link PollableChannel}. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public class PollingConsumer extends AbstractPollingEndpoint { @@ -33,15 +34,8 @@ 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"); Assert.notNull(handler, "handler must not be null"); @@ -53,18 +47,6 @@ public class PollingConsumer extends AbstractPollingEndpoint { public void setReceiveTimeout(long receiveTimeout) { this.receiveTimeout = receiveTimeout; } - - @Override - protected void onInit() { - synchronized (this.initializationMonitor) { - if (!this.initialized) { - this.handlerInvocationChain = new HandlerInvocationChain(this.handler, this.getComponentName()); - } - this.initialized = true; - } - super.onInit(); - } - @Override protected boolean doPoll() { Message message = (this.receiveTimeout >= 0) @@ -73,8 +55,7 @@ public class PollingConsumer extends AbstractPollingEndpoint { if (message == null) { return false; } - this.handlerInvocationChain.handleMessage(message); + this.handler.handleMessage(message); return true; } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java index 0a101d5df8..28f8d15c68 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java @@ -62,6 +62,10 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { private final Object replyMessageCorrelatorMonitor = new Object(); + @Override + public String getComponentType(){ + return "gateway"; + } /** * Set the request channel. @@ -272,5 +276,4 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { this.replyMessageCorrelator.stop(); } } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java index cbc734d01c..66ca3882ae 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java @@ -18,6 +18,7 @@ package org.springframework.integration.gateway; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessagingException; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.InboundMessageMapper; import org.springframework.integration.message.OutboundMessageMapper; import org.springframework.util.Assert; @@ -80,9 +81,7 @@ public class SimpleMessagingGateway extends AbstractMessagingGateway { Message message = null; try { message = this.inboundMapper.toMessage(object); - if (message != null) { - message.getHeaders().getHistory().addEvent(this); - } + MessageHistory.writeMessageHistory(message, this, this.getBeanFactory()); } catch (Exception e) { if (e instanceof RuntimeException) { @@ -92,5 +91,4 @@ public class SimpleMessagingGateway extends AbstractMessagingGateway { } return message; } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 570c76325e..2488de36a6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -18,7 +18,6 @@ package org.springframework.integration.handler; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.core.Ordered; import org.springframework.integration.channel.ChannelResolutionException; import org.springframework.integration.channel.ChannelResolver; @@ -26,6 +25,7 @@ import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessagingException; +import org.springframework.integration.history.MessageHistory; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.MessageHandlingException; import org.springframework.util.Assert; @@ -37,6 +37,7 @@ import org.springframework.util.Assert; * checked exceptions into runtime {@link MessagingException}s. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler, Ordered { @@ -52,10 +53,16 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im public int getOrder() { return this.order; } + + @Override + public String getComponentType() { + return "message-handler"; + } public final void handleMessage(Message message) { Assert.notNull(message, "Message must not be null"); Assert.notNull(message.getPayload(), "Message payload must not be null"); + MessageHistory.writeMessageHistory(message, this, this.getBeanFactory()); if (this.logger.isDebugEnabled()) { this.logger.debug(this + " received message: " + message); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index 20982caa7d..4c779bcea8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -21,10 +21,6 @@ import java.util.UUID; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.Ordered; @@ -33,6 +29,7 @@ import org.springframework.integration.channel.ChannelResolutionException; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.MessageChannelTemplate; import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; import org.springframework.integration.core.MessageHeaders; @@ -73,7 +70,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @since 1.0.3 */ -public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, InitializingBean, DisposableBean { +public class DelayHandler extends IntegrationObjectSupport implements MessageHandler, Ordered, DisposableBean { private final Log logger = LogFactory.getLog(this.getClass()); @@ -81,7 +78,7 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, private volatile String delayHeaderName; - private final TaskScheduler taskScheduler; + private boolean waitForTasksToCompleteOnShutdown = false;; private volatile MessageChannel outputChannel; @@ -94,6 +91,10 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, private volatile int order = Ordered.LOWEST_PRECEDENCE; + @Override + public String getComponentType() { + return "delayer"; + } /** * Create a DelayHandler with the given default delay. The sending of Messages after * the delay will be handled by a scheduled thread pool with a size of 1. @@ -108,8 +109,7 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, */ public DelayHandler(long defaultDelay, TaskScheduler taskScheduler) { this.defaultDelay = defaultDelay; - this.taskScheduler = (taskScheduler != null) - ? taskScheduler : new ThreadPoolTaskScheduler(); + this.setTaskScheduler(taskScheduler != null ? taskScheduler : new ThreadPoolTaskScheduler()); } @@ -165,13 +165,7 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, * @see ExecutorConfigurationSupport#setWaitForTasksToCompleteOnShutdown(boolean) */ public void setWaitForTasksToCompleteOnShutdown(boolean waitForJobsToCompleteOnShutdown) { - if (this.taskScheduler instanceof ExecutorConfigurationSupport) { - ((ExecutorConfigurationSupport) this.taskScheduler).setWaitForTasksToCompleteOnShutdown(waitForJobsToCompleteOnShutdown); - } - else if (logger.isWarnEnabled()) { - logger.warn("The 'waitForJobsToCompleteOnShutdown' property is not supported for TaskScheduler of type [" + - this.taskScheduler.getClass() + "]"); - } + this.waitForTasksToCompleteOnShutdown = waitForJobsToCompleteOnShutdown; } public void setOrder(int order) { @@ -182,17 +176,22 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, return this.order; } - public void setBeanFactory(BeanFactory beanFactory) throws BeansException { - this.channelResolver = new BeanFactoryChannelResolver(beanFactory); - } - - - public void afterPropertiesSet() throws Exception { + protected void onInit() throws Exception{ + if (this.getTaskScheduler() instanceof ExecutorConfigurationSupport) { + ((ExecutorConfigurationSupport) this.getTaskScheduler()).setWaitForTasksToCompleteOnShutdown(this.waitForTasksToCompleteOnShutdown); + } + else if (logger.isWarnEnabled()) { + logger.warn("The 'waitForJobsToCompleteOnShutdown' property is not supported for TaskScheduler of type [" + + this.getTaskScheduler().getClass() + "]"); + } if (this.messageStore == null) { this.messageStore = new SimpleMessageStore(); } - if (this.taskScheduler instanceof InitializingBean) { - ((InitializingBean) this.taskScheduler).afterPropertiesSet(); + if (this.getTaskScheduler() instanceof InitializingBean) { + ((InitializingBean) this.getTaskScheduler()).afterPropertiesSet(); + } + if (this.getBeanFactory() != null){ + this.channelResolver = new BeanFactoryChannelResolver(this.getBeanFactory()); } } @@ -232,7 +231,7 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, private void releaseMessageAfterDelay(final Message message, long delay) { Assert.state(this.messageStore != null, "MessageStore must not be null"); final Message storedMessage = this.messageStore.addMessage(message); - this.taskScheduler.schedule(new Runnable() { + this.getTaskScheduler().schedule(new Runnable() { public void run() { try { releaseMessage(storedMessage.getHeaders().getId()); @@ -316,9 +315,8 @@ public class DelayHandler implements MessageHandler, Ordered, BeanFactoryAware, } public void destroy() throws Exception { - if (this.taskScheduler instanceof DisposableBean) { - ((DisposableBean) this.taskScheduler).destroy(); + if (this.getTaskScheduler() instanceof DisposableBean) { + ((DisposableBean) this.getTaskScheduler()).destroy(); } } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistory.java b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistory.java index 28e7919719..190d287382 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistory.java @@ -22,7 +22,9 @@ import java.util.Iterator; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.context.NamedComponent; +import org.springframework.integration.core.Message; import org.springframework.util.StringUtils; /** @@ -76,4 +78,14 @@ public class MessageHistory implements Iterable, Serializab public String toString() { return new ArrayList(events).toString(); } + + public static void writeMessageHistory(Message message, NamedComponent component, BeanFactory beanFactory){ + if (beanFactory != null){ + if (beanFactory.containsBean(MessageHistoryWriter.HISTORY_WRITER_BEAN_NAME)){ + MessageHistoryWriter writer = + beanFactory.getBean(MessageHistoryWriter.HISTORY_WRITER_BEAN_NAME, MessageHistoryWriter.class); + writer.writeHistory(component, message); + } + } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryEvent.java b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryEvent.java index 81cfbf6e60..a3aef9ae0a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryEvent.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryEvent.java @@ -23,6 +23,7 @@ import java.io.Serializable; * with a timestamp that is generated when this event is created. * * @author Mark Fisher + * @author Oleg Zhurakousky * @since 2.0 */ public class MessageHistoryEvent implements Serializable { @@ -68,7 +69,7 @@ public class MessageHistoryEvent implements Serializable { } if (this.name != null) { if (this.type != null) { - sb.append(':'); + sb.append('@'); } sb.append(name); //sb.append("[" + timestamp + "]"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryWriter.java b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryWriter.java new file mode 100644 index 0000000000..4f5f6cf5eb --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/history/MessageHistoryWriter.java @@ -0,0 +1,31 @@ +/* + * 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.history; + +import org.springframework.integration.context.NamedComponent; +import org.springframework.integration.core.Message; + +/** + * @author Oleg Zhurakousky + * @since 2.0 + */ +public class MessageHistoryWriter { + public final static String HISTORY_WRITER_BEAN_NAME = "historyWriter"; + + public void writeHistory(NamedComponent component, Message message){ + message.getHeaders().getHistory().addEvent(component); + } +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java index 8865b54bd2..1d1fbe040a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayProxyFactoryBeanTests.java @@ -32,6 +32,7 @@ import org.mockito.Mockito; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.context.support.GenericApplicationContext; import org.springframework.core.convert.converter.Converter; import org.springframework.core.convert.support.ConversionServiceFactory; import org.springframework.core.convert.support.GenericConversionService; @@ -44,6 +45,7 @@ import org.springframework.integration.core.MessageChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.history.MessageHistoryEvent; +import org.springframework.integration.history.MessageHistoryWriter; import org.springframework.integration.message.MessageHandler; import org.springframework.integration.message.StringMessage; import org.springframework.util.ReflectionUtils; @@ -296,15 +298,21 @@ public class GatewayProxyFactoryBeanTests { @Test public void testHistory() throws Exception { + GenericApplicationContext context = new GenericApplicationContext(); + context.getBeanFactory().registerSingleton("historyWriter", new MessageHistoryWriter()); GatewayProxyFactoryBean proxyFactory = new GatewayProxyFactoryBean(); + proxyFactory.setBeanFactory(context); proxyFactory.setBeanName("testGateway"); DirectChannel channel = new DirectChannel(); channel.setBeanName("testChannel"); + channel.setBeanFactory(context); channel.afterPropertiesSet(); BridgeHandler bridgeHandler = new BridgeHandler(); + bridgeHandler.setBeanFactory(context); bridgeHandler.afterPropertiesSet(); + bridgeHandler.setBeanName("testBridge"); EventDrivenConsumer consumer = new EventDrivenConsumer(channel, bridgeHandler); - consumer.setBeanName("testBridge"); + consumer.setBeanFactory(context); consumer.afterPropertiesSet(); consumer.start(); proxyFactory.setDefaultRequestChannel(channel); @@ -316,7 +324,9 @@ public class GatewayProxyFactoryBeanTests { MessageHistoryEvent event1 = historyIterator.next(); MessageHistoryEvent event2 = historyIterator.next(); MessageHistoryEvent event3 = historyIterator.next(); + //assertEquals("echo", event1.getAttribute("method", String.class)); + assertEquals("gateway", event1.getType()); assertEquals("testGateway", event1.getName()); assertEquals("channel", event2.getType()); assertEquals("testChannel", event2.getName()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java index 8fda2f44e6..81cf6cd7cf 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java @@ -73,7 +73,6 @@ public class SimpleMessagingGatewayTests { @Test public void sendMessage() { - expect(messageMock.getHeaders()).andReturn(new MessageHeaders(null)); expect(requestChannel.send(messageMock)).andReturn(true); replay(allmocks); this.simpleMessagingGateway.send(messageMock);