From 654a62641926498de4b95914daa166e51800431f Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 28 Jul 2008 23:42:50 +0000 Subject: [PATCH] Added MessageExchangeTemplate, AsyncMessageExchangeTemplate, and AsyncMessage. Removed Poller, SimplePoller, DefaultEndpointPoller, SourceInvoker, and TargetInvoker. --- .../config/AbstractHandlerEndpointParser.java | 2 +- .../config/IntegrationNamespaceUtils.java | 13 +- .../dispatcher/AbstractDispatcher.java | 10 +- .../endpoint/AbstractEndpoint.java | 35 +-- .../endpoint/DefaultEndpointPoller.java | 119 ------- .../integration/message/AsyncMessage.java | 101 ++++++ .../message/AsyncMessageExchangeTemplate.java | 107 +++++++ .../message/MessageExchangeTemplate.java | 295 ++++++++++++++++++ .../integration/message/Poller.java | 29 -- .../integration/message/SimplePoller.java | 64 ---- .../integration/message/SourceInvoker.java | 42 --- .../integration/message/TargetInvoker.java | 42 --- 12 files changed, 530 insertions(+), 329 deletions(-) delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointPoller.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/message/Poller.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/message/SimplePoller.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/message/SourceInvoker.java delete mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/message/TargetInvoker.java diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java index 617a77ad18..4a2261fc71 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java @@ -112,7 +112,7 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe schedule = this.parseSchedule(childElement); } else if (POLLER_ELEMENT.equals(localName)) { - builder.addPropertyReference("poller", + builder.addPropertyReference("messageExchangeTemplate", IntegrationNamespaceUtils.parsePoller(childElement, parserContext)); } else if (INTERCEPTORS_ELEMENT.equals(localName)) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java index 11595d0e3b..9f66b8ba71 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java @@ -24,7 +24,8 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate; import org.springframework.beans.factory.xml.ParserContext; -import org.springframework.integration.endpoint.DefaultEndpointPoller; +import org.springframework.integration.message.AsyncMessageExchangeTemplate; +import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.transaction.support.DefaultTransactionDefinition; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; @@ -106,7 +107,12 @@ public abstract class IntegrationNamespaceUtils { } return ref; } - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(DefaultEndpointPoller.class); + Class beanClass = (StringUtils.hasText(taskExecutorRef)) ? + AsyncMessageExchangeTemplate.class : MessageExchangeTemplate.class; + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(beanClass); + if (StringUtils.hasText(taskExecutorRef)) { + builder.addConstructorArgReference(taskExecutorRef); + } if (txElement != null) { builder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager")); builder.addPropertyValue("propagationBehaviorName", DefaultTransactionDefinition.PREFIX_PROPAGATION + txElement.getAttribute("propagation")); @@ -114,9 +120,6 @@ public abstract class IntegrationNamespaceUtils { builder.addPropertyValue("transactionTimeout", txElement.getAttribute("timeout")); builder.addPropertyValue("transactionReadOnly", txElement.getAttribute("read-only")); } - if (StringUtils.hasText(taskExecutorRef)) { - builder.addPropertyReference("taskExecutor", taskExecutorRef); - } String receiveTimeout = element.getAttribute("receive-timeout"); if (StringUtils.hasText(receiveTimeout)) { builder.addPropertyValue("receiveTimeout", Long.parseLong(receiveTimeout)); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index f5dc15ca7a..5cbe34b224 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -24,8 +24,8 @@ import org.apache.commons.logging.LogFactory; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.TargetInvoker; /** * Base class for {@link MessageDispatcher} implementations. @@ -38,15 +38,13 @@ public abstract class AbstractDispatcher implements MessageDispatcher { protected final List targets = new CopyOnWriteArrayList(); - private volatile long timeout = 0; - private volatile TaskExecutor taskExecutor; - private final TargetInvoker targetInvoker = new TargetInvoker(); + private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate(); public void setTimeout(long timeout) { - this.timeout = timeout; + this.messageExchangeTemplate.setSendTimeout(timeout); } public boolean addTarget(MessageTarget target) { @@ -71,7 +69,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher { } protected final boolean sendMessageToTarget(Message message, MessageTarget target) { - return this.targetInvoker.invoke(target, message, this.timeout); + return this.messageExchangeTemplate.send(message, target); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index 46f7b1f88f..6f6454c423 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -28,13 +28,11 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageRejectedException; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.MessagingException; -import org.springframework.integration.message.Poller; -import org.springframework.integration.message.Subscribable; -import org.springframework.integration.message.TargetInvoker; import org.springframework.integration.message.selector.MessageSelector; import org.springframework.integration.scheduling.Schedule; @@ -57,13 +55,9 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist private volatile MessageTarget target; - private volatile Poller poller; - private volatile Schedule schedule; - private final TargetInvoker targetInvoker = new TargetInvoker(); - - private volatile long sendTimeout; + private volatile MessageExchangeTemplate messageExchangeTemplate; private volatile MessageSelector selector; @@ -92,8 +86,8 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist return this.schedule; } - public void setPoller(Poller poller) { - this.poller = poller; + public void setMessageExchangeTemplate(MessageExchangeTemplate messageExchangeTemplate) { + this.messageExchangeTemplate = messageExchangeTemplate; } public void setInputChannelName(String inputChannelName) { @@ -147,7 +141,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist } public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; + this.messageExchangeTemplate.setSendTimeout(sendTimeout); } public MessageChannel getOutputChannel() { @@ -200,9 +194,9 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist } public void afterPropertiesSet() { - if (this.poller == null && this.source != null - && !(this.source instanceof Subscribable)) { - this.poller = new DefaultEndpointPoller(); + if (this.messageExchangeTemplate == null) { + this.messageExchangeTemplate = new MessageExchangeTemplate(); + this.messageExchangeTemplate.afterPropertiesSet(); } if (this.target == null) { this.target = this.getOutputChannel(); @@ -254,32 +248,31 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist } private boolean doSend(Message message) { + if (this.messageExchangeTemplate == null) { + this.afterPropertiesSet(); + } if (!this.supports(message)) { throw new MessageRejectedException(message, "unsupported message"); } Message result = this.handleMessage(message); if (result != null) { - return this.targetInvoker.invoke(this.target, result, this.sendTimeout); + return this.messageExchangeTemplate.send(message, this.target); } return true; } public final boolean poll() { - if (this.poller == null) { + if (this.messageExchangeTemplate == null) { this.afterPropertiesSet(); - if (this.poller == null) { - throw new MessagingException("endpoint '" + this + "' has no poller"); - } } if (this.source == null) { throw new MessagingException("endpoint '" + this + "' has no source"); } - int result = this.poller.poll(this.source, new MessageTarget() { + return this.messageExchangeTemplate.receiveAndForward(this.source, new MessageTarget() { public boolean send(Message message) { return AbstractEndpoint.this.send(message, 0); } }); - return (result > 0); } protected boolean supports(Message message) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointPoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointPoller.java deleted file mode 100644 index 9ce0d5b905..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpointPoller.java +++ /dev/null @@ -1,119 +0,0 @@ -/* - * Copyright 2002-2008 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.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.InitializingBean; -import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.message.MessageSource; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.SimplePoller; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.TransactionStatus; -import org.springframework.transaction.support.TransactionCallback; -import org.springframework.transaction.support.TransactionTemplate; - -/** - * An extension of the {@link SimplePoller} that adds concurrency and - * transactional capabilities. - * - * @author Mark Fisher - */ -public class DefaultEndpointPoller extends SimplePoller implements InitializingBean { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private volatile TaskExecutor taskExecutor; - - private volatile PlatformTransactionManager transactionManager; - - private volatile TransactionTemplate transactionTemplate; - - private volatile String propagationBehaviorName = "PROPAGATION_REQUIRED"; - - private volatile String isolationLevelName = "ISOLATION_DEFAULT"; - - private volatile int transactionTimeout = -1; - - private volatile boolean readOnly = false; - - - public void setTaskExecutor(TaskExecutor taskExecutor) { - this.taskExecutor = taskExecutor; - } - - public void setTransactionManager(PlatformTransactionManager transactionManager) { - this.transactionManager = transactionManager; - } - - public void setPropagationBehaviorName(String propagationBehaviorName) { - this.propagationBehaviorName = propagationBehaviorName; - } - - public void setIsolationLevelName(String isolationLevelName) { - this.isolationLevelName = isolationLevelName; - } - - public void setTransactionTimeout(int transactionTimeout) { - this.transactionTimeout = transactionTimeout; - } - - public void setTransactionReadOnly(boolean readOnly) { - this.readOnly = readOnly; - } - - public void afterPropertiesSet() { - if (this.transactionManager != null) { - TransactionTemplate template = new TransactionTemplate(this.transactionManager); - template.setPropagationBehaviorName(this.propagationBehaviorName); - template.setIsolationLevelName(this.isolationLevelName); - template.setTimeout(this.transactionTimeout); - template.setReadOnly(this.readOnly); - this.transactionTemplate = template; - } - } - - public int poll(final MessageSource source, final MessageTarget target) { - if (this.taskExecutor != null) { - this.taskExecutor.execute(new Runnable() { - public void run() { - doPoll(source, target); - } - }); - return 1; - } - return doPoll(source, target); - } - - private int doPoll(final MessageSource source, final MessageTarget target) { - if (this.transactionTemplate != null) { - int result = (Integer) this.transactionTemplate.execute(new TransactionCallback() { - public Object doInTransaction(TransactionStatus status) { - if (logger.isDebugEnabled()) { - logger.debug("Polling source '" + source + "' within transaction [" + status + "]"); - } - return DefaultEndpointPoller.super.poll(source, target); - } - }); - return result; - } - return super.poll(source, target); - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java new file mode 100644 index 0000000000..ae3c2a791e --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java @@ -0,0 +1,101 @@ +/* + * Copyright 2002-2008 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.message; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHeaders; +import org.springframework.integration.message.MessagingException; + +/** + * A Message wrapper for asynchronous operations. Implements both + * Message and Future so that simple blocking invocations on the + * Message (e.g. {@link #getPayload()}) are still allowed, while + * timeout-aware methods such as {@link #get(long, TimeUnit)} + * are also supported. + * + * @author Mark Fisher + */ +public class AsyncMessage implements Future>, Message { + + private final Future> future; + + + public AsyncMessage(Future> future) { + this.future = future; + } + + + public boolean cancel(boolean mayInterruptIfRunning) { + return this.future.cancel(mayInterruptIfRunning); + } + + public Message get() throws InterruptedException, ExecutionException { + return this.future.get(); + } + + public Message get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + return this.future.get(timeout, unit); + } + + public boolean isCancelled() { + return this.future.isCancelled(); + } + + public boolean isDone() { + return this.future.isDone(); + } + + public MessageHeaders getHeaders() { + try { + return this.future.get().getHeaders(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } catch (ExecutionException e) { + throw new MessagingException("failure occurred in AsyncMessage", e); + } + } + + public Object getId() { + try { + return this.future.get().getId(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } catch (ExecutionException e) { + throw new MessagingException("failure occurred in AsyncMessage", e); + } + } + + public T getPayload() { + try { + return this.future.get().getPayload(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } catch (ExecutionException e) { + throw new MessagingException("failure occurred in AsyncMessage", e); + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java new file mode 100644 index 0000000000..88abedc39e --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java @@ -0,0 +1,107 @@ +/* + * Copyright 2002-2008 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.message; + +import java.util.concurrent.Callable; +import java.util.concurrent.FutureTask; + +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageExchangeTemplate; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.MessageTarget; +import org.springframework.util.Assert; + +/** + * An asynchronous version of the {@link MessageExchangeTemplate}. + * + * @author Mark Fisher + */ +public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate { + + private final TaskExecutor taskExecutor; + + + public AsyncMessageExchangeTemplate(TaskExecutor taskExecutor) { + Assert.notNull(taskExecutor, "TaskExecutor must not be null"); + this.taskExecutor = taskExecutor; + } + + + /** + * Send the provided message to the given target. Note that the actual + * sending occurs asynchronously, so this method will always return + * true unless an exception is thrown by the executor. + */ + @Override + public boolean send(final Message message, final MessageTarget target) { + this.taskExecutor.execute(new Runnable() { + public void run() { + AsyncMessageExchangeTemplate.super.send(message, target); + } + }); + return true; + } + + /** + * Send the provided message to the given target and receive the + * result as an {@link AsyncMessage}. + */ + @Override + @SuppressWarnings("unchecked") + public Message sendAndReceive(final Message request, final MessageTarget target) { + FutureTask> task = new FutureTask>(new Callable>() { + public Message call() throws Exception { + return AsyncMessageExchangeTemplate.super.sendAndReceive(request, target); + } + }); + this.taskExecutor.execute(task); + return new AsyncMessage(task); + } + + /** + * Receive an {@link AsyncMessage} from the provided source. + */ + @Override + @SuppressWarnings("unchecked") + public Message receive(final MessageSource source) { + FutureTask> task = new FutureTask>(new Callable>() { + public Message call() throws Exception { + return AsyncMessageExchangeTemplate.super.receive(source); + } + }); + this.taskExecutor.execute(task); + return new AsyncMessage(task); + } + + /** + * Receive a Message from the provided source and if not null, + * send it to the given target. Note that the receive and send operations + * occur asynchronously, so this method will always return true + * unless an exception is thrown by the executor. + */ + @Override + public boolean receiveAndForward(final MessageSource source, final MessageTarget target) { + this.taskExecutor.execute(new Runnable() { + public void run() { + AsyncMessageExchangeTemplate.super.receiveAndForward(source, target); + } + }); + return true; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java new file mode 100644 index 0000000000..7abc0c6385 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java @@ -0,0 +1,295 @@ +/* + * Copyright 2002-2008 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.message; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.message.BlockingSource; +import org.springframework.integration.message.BlockingTarget; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.MessageTarget; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; + +/** + * This is the central class for invoking message exchange operations + * across {@link MessageSource}s and {@link MessageTarget}s. It supports + * one-way send and receive calls as well as request/reply. Additionally, + * the {@link #receiveAndForward(MessageSource, MessageTarget)} method + * plays the role of a polling-consumer while actually sending any + * received message to an event-driven consumer. + * + *

To enable transactions, configure the 'transactionManager' property + * with a reference to an instance of Spring's {@link PlatformTransactionManager} + * strategy and optionally provide the other transactional attributes + * (e.g. 'propagationBehaviorName'). + * + * @author Mark Fisher + */ +public class MessageExchangeTemplate implements InitializingBean { + + protected final Log logger = LogFactory.getLog(this.getClass()); + + private volatile long sendTimeout = -1; + + private volatile long receiveTimeout = -1; + + private volatile PlatformTransactionManager transactionManager; + + private volatile TransactionTemplate transactionTemplate; + + private volatile String propagationBehaviorName = "PROPAGATION_REQUIRED"; + + private volatile String isolationLevelName = "ISOLATION_DEFAULT"; + + private volatile int transactionTimeout = -1; + + private volatile boolean readOnly = false; + + private volatile boolean initialized; + + private final Object initializationMonitor = new Object(); + + + /** + * Specify the timeout value to use for send operations. + * Note that this value will only apply to {@link BlockingTarget}s. + * + * @param sendTimeout the send timeout in milliseconds + */ + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + + /** + * Specify the timeout value to use for receive operations. + * Note that this value will only apply to {@link BlockingSource}s. + * + * @param receiveTimeout the receive timeout in milliseconds + */ + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + /** + * Specify a transaction manager to use for all exchange operations. + * If none is provided, then the operations will occur without any + * transactional behavior (i.e. there is no default transaction manager). + */ + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + public void setPropagationBehaviorName(String propagationBehaviorName) { + this.propagationBehaviorName = propagationBehaviorName; + } + + public void setIsolationLevelName(String isolationLevelName) { + this.isolationLevelName = isolationLevelName; + } + + public void setTransactionTimeout(int transactionTimeout) { + this.transactionTimeout = transactionTimeout; + } + + public void setTransactionReadOnly(boolean readOnly) { + this.readOnly = readOnly; + } + + private TransactionTemplate getTransactionTemplate() { + if (!this.initialized) { + this.afterPropertiesSet(); + } + return this.transactionTemplate; + } + + public void afterPropertiesSet() { + synchronized (this.initializationMonitor) { + if (this.initialized) { + return; + } + if (this.transactionManager != null) { + TransactionTemplate template = new TransactionTemplate(this.transactionManager); + template.setPropagationBehaviorName(this.propagationBehaviorName); + template.setIsolationLevelName(this.isolationLevelName); + template.setTimeout(this.transactionTimeout); + template.setReadOnly(this.readOnly); + this.transactionTemplate = template; + } + this.initialized = true; + } + } + + public boolean send(final Message message, final MessageTarget target) { + TransactionTemplate txTemplate = this.getTransactionTemplate(); + if (txTemplate != null) { + return (Boolean) txTemplate.execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + return doSend(message, target); + } + }); + } + return this.doSend(message, target); + } + + public Message sendAndReceive(final Message request, final MessageTarget target) { + TransactionTemplate txTemplate = this.getTransactionTemplate(); + if (txTemplate != null) { + return (Message) txTemplate.execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + return doSendAndReceive(request, target); + } + }); + } + return this.doSendAndReceive(request, target); + } + + public Message receive(final MessageSource source) { + TransactionTemplate txTemplate = this.getTransactionTemplate(); + if (txTemplate != null) { + return (Message) txTemplate.execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + return doReceive(source); + } + }); + } + return this.doReceive(source); + } + + public boolean receiveAndForward(final MessageSource source, final MessageTarget target) { + TransactionTemplate txTemplate = this.getTransactionTemplate(); + if (txTemplate != null) { + return (Boolean) txTemplate.execute(new TransactionCallback() { + public Object doInTransaction(TransactionStatus status) { + return doReceiveAndForward(source, target); + } + }); + } + return this.doReceiveAndForward(source, target); + } + + + private boolean doSend(Message message, MessageTarget target) { + long timeout = this.sendTimeout; + boolean sent = (timeout >= 0 && target instanceof BlockingTarget) + ? ((BlockingTarget) target).send(message, timeout) + : target.send(message); + if (!sent && this.logger.isTraceEnabled()) { + this.logger.trace("failed to send message to target '" + target + "' within timeout: " + timeout); + } + return sent; + } + + private Message doReceive(MessageSource source) { + long timeout = this.receiveTimeout; + Message message = (timeout >= 0 && source instanceof BlockingSource) + ? ((BlockingSource) source).receive(timeout) + : source.receive(); + if (message == null && this.logger.isTraceEnabled()) { + this.logger.trace("failed to receive message from source '" + source + "' within timeout: " + timeout); + } + return message; + } + + private Message doSendAndReceive(Message request, MessageTarget target) { + TemporaryReturnAddress returnAddress = new TemporaryReturnAddress(this.receiveTimeout); + request = MessageBuilder.fromMessage(request).setReturnAddress(returnAddress).build(); + if (!this.doSend(request, target)) { + return null; + } + return this.doReceive(returnAddress); + } + + private boolean doReceiveAndForward(MessageSource source, MessageTarget target) { + Message message = this.doReceive(source); + if (message == null) { + return false; + } + try { + boolean sent = this.doSend(message, target); + if (source instanceof MessageDeliveryAware) { + if (sent) { + ((MessageDeliveryAware) source).onSend(message); + } + else { + ((MessageDeliveryAware) source).onFailure(new MessageDeliveryException(message, "failed to send message")); + } + } + return sent; + } + catch (Exception e) { + MessagingException exception = new MessagingException( + "exception occurred in receive-and-forward exchange", e); + if (source instanceof MessageDeliveryAware) { + ((MessageDeliveryAware) source).onFailure(exception); + } + throw exception; + } + } + + + @SuppressWarnings("unchecked") + private static class TemporaryReturnAddress implements BlockingSource, MessageTarget { + + private volatile Message message; + + private final long receiveTimeout; + + private final CountDownLatch latch = new CountDownLatch(1); + + + public TemporaryReturnAddress(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + + public Message receive() { + return this.receive(-1); + } + + public Message receive(long timeout) { + try { + if (this.receiveTimeout < 0) { + this.latch.await(); + } + else { + this.latch.await(this.receiveTimeout, TimeUnit.MILLISECONDS); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return this.message; + } + + public boolean send(Message message) { + this.message = message; + this.latch.countDown(); + return true; + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/Poller.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/Poller.java deleted file mode 100644 index b402d2ed85..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/Poller.java +++ /dev/null @@ -1,29 +0,0 @@ -/* - * Copyright 2002-2008 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.message; - -/** - * Strategy interface for polling a {@link MessageSource} and passing - * any Messages received from that source to a {@link MessageTarget}. - * - * @author Mark Fisher - */ -public interface Poller { - - int poll(MessageSource source, MessageTarget target); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/SimplePoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/SimplePoller.java deleted file mode 100644 index dd3c3f5678..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/SimplePoller.java +++ /dev/null @@ -1,64 +0,0 @@ -/* - * Copyright 2002-2008 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.message; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -/** - * @author Mark Fisher - */ -public class SimplePoller implements Poller { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private long receiveTimeout = 5000; - - private long sendTimeout = -1; - - - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; - } - - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - - public int poll(MessageSource source, MessageTarget target) { - Message message = (source instanceof BlockingSource && this.receiveTimeout >= 0) ? - ((BlockingSource) source).receive(this.receiveTimeout) : source.receive(); - if (message == null) { - if (logger.isDebugEnabled()) { - logger.debug("received no Message from source '" + source + "'"); - } - return 0; - } - boolean sent = (target instanceof BlockingTarget && this.sendTimeout >= 0) ? - ((BlockingTarget) target).send(message, this.sendTimeout) : target.send(message); - if (source instanceof MessageDeliveryAware) { - if (sent) { - ((MessageDeliveryAware) source).onSend(message); - } - else { - ((MessageDeliveryAware) source).onFailure(new MessageDeliveryException(message, "failed to send message")); - } - } - return (sent ? 1 : 0); - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/SourceInvoker.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/SourceInvoker.java deleted file mode 100644 index e0d433c2eb..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/SourceInvoker.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2002-2008 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.message; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -/** - * A helper class for receiving {@link Message}s from a {@link MessageSource}. - * - * @author Mark Fisher - */ -public class SourceInvoker { - - private final Log logger = LogFactory.getLog(this.getClass()); - - - public Message invoke(MessageSource source, long timeout) { - Message message = (source instanceof BlockingSource && timeout >= 0) - ? ((BlockingSource) source).receive(timeout) - : source.receive(); - if (message == null && this.logger.isTraceEnabled()) { - this.logger.trace("failed to receive message from source '" + source + "' within timeout: " + timeout); - } - return message; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/TargetInvoker.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/TargetInvoker.java deleted file mode 100644 index 4f45f20363..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/TargetInvoker.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2002-2008 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.message; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -/** - * A helper class for sending {@link Message}s to a {@link MessageTarget}. - * - * @author Mark Fisher - */ -public class TargetInvoker { - - private final Log logger = LogFactory.getLog(this.getClass()); - - - public boolean invoke(MessageTarget target, Message message, long timeout) { - boolean sent = (target instanceof BlockingTarget && timeout >= 0) - ? ((BlockingTarget) target).send(message, timeout) - : target.send(message); - if (!sent && this.logger.isTraceEnabled()) { - this.logger.trace("failed to send message to target '" + target + "' within timeout: " + timeout); - } - return sent; - } - -}