Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

MessageChannels no longer implement MessageTarget, and MessageEndpoints that send a reply have a setOutputChannel() method instead of setTarget().

Browse Source
This commit is contained in:
Mark Fisher
2008-09-05 16:36:13 +00:00
parent 63720ded3f
commit b47d81ff16
42 changed files with 304 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java
View File

@@ -30,13 +30,12 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.BlockingChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.AbstractInOutEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
@@ -199,19 +198,19 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractInOutEndpoi
private void afterRelease(Object correlationId, List<Message<?>> releasedMessages) {
Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
for (Message<?> result : processedMessages) {
MessageTarget replyTarget = this.getTarget();
if (replyTarget == null) {
replyTarget = this.resolveReplyTargetFromMessage(result);
if (replyTarget == null) {
replyTarget = this.resolveReplyTargetFromMessage(releasedMessages.get(0));
MessageChannel replyChannel = this.getOutputChannel();
if (replyChannel == null) {
replyChannel = this.resolveReplyChannelFromMessage(result);
if (replyChannel == null) {
replyChannel = this.resolveReplyChannelFromMessage(releasedMessages.get(0));
}
}
if (replyTarget != null) {
if (replyTarget instanceof BlockingTarget && this.sendTimeout >= 0) {
((BlockingTarget) replyTarget).send(result, this.sendTimeout);
if (replyChannel != null) {
if (replyChannel instanceof BlockingChannel && this.sendTimeout >= 0) {
((BlockingChannel) replyChannel).send(result, this.sendTimeout);
}
else {
replyTarget.send(result);
replyChannel.send(result);
}
}
else if (logger.isWarnEnabled()) {
@@ -222,7 +221,10 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractInOutEndpoi
private void sendToDiscardChannelIfAvailable(Message<?> message) {
if (this.discardChannel != null) {
if (!this.discardChannel.send(message, this.sendTimeout)) {
boolean sent = (this.discardChannel instanceof BlockingChannel && this.sendTimeout >= 0)
? ((BlockingChannel) this.discardChannel).send(message, this.sendTimeout)
: this.discardChannel.send(message);
if (!sent) {
if (logger.isWarnEnabled()) {
logger.warn("unable to send to 'discardChannel', message: " + message);
}
@@ -230,14 +232,14 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractInOutEndpoi
}
}
protected MessageTarget resolveReplyTargetFromMessage(Message<?> message) {
protected MessageChannel resolveReplyChannelFromMessage(Message<?> message) {
Object returnAddress = message.getHeaders().getReturnAddress();
if (returnAddress != null) {
if (returnAddress instanceof MessageTarget) {
return (MessageTarget) returnAddress;
if (returnAddress instanceof MessageChannel) {
return (MessageChannel) returnAddress;
}
if (logger.isWarnEnabled()) {
logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageTarget.");
logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageChannel.");
}
}
return null;

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java
View File

@@ -34,7 +34,7 @@ import org.springframework.integration.message.Message;
*
* @author Mark Fisher
*/
public abstract class AbstractMessageChannel implements MessageChannel, BeanNameAware {
public abstract class AbstractMessageChannel implements BlockingChannel, BeanNameAware {
private final Log logger = LogFactory.getLog(this.getClass());

52
org.springframework.integration/src/main/java/org/springframework/integration/channel/BlockingChannel.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* 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.channel;
import org.springframework.integration.message.Message;
/**
* Extends the base MessageChannel interface for channels that may block when a
* Message is sent. Adds the timeout-aware {@link #send(Message, long)} method.
*
* @author Mark Fisher
*/
public interface BlockingChannel extends MessageChannel {
/**
* Send a message, blocking indefinitely if necessary.
*
* @param message the {@link Message} to send
*
* @return <code>true</code> if the message is sent successfully,
* <code>false</false> if interrupted
*/
boolean send(Message<?> message);
/**
* Send a message, blocking until either the message is accepted or the
* specified timeout period elapses.
*
* @param message the {@link Message} to send
* @param timeout the timeout in milliseconds
*
* @return <code>true</code> if the message is sent successfully,
* <code>false</false> if the specified timeout period elapses or
* the send is interrupted
*/
boolean send(Message<?> message, long timeout);
}

14
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannel.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.channel;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
/**
@@ -24,11 +24,21 @@ import org.springframework.integration.message.MessageSource;
*
* @author Mark Fisher
*/
public interface MessageChannel extends MessageSource, BlockingTarget {
public interface MessageChannel extends MessageSource {
/**
* Return the name of this channel.
*/
String getName();
/**
* Send a {@link Message} to this channel. May throw a RuntimeException for non-recoverable
* errors. Otherwise, if the Message cannot be sent for a non-fatal reason this method will
* return 'false', and if the Message is sent successfully, it will return 'true'.
*
* @param message the Message to send
* @return whether the Message has been sent successfully
*/
boolean send(Message<?> message);
}

7
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java
View File

@@ -62,7 +62,12 @@ public class MessagePublishingErrorHandler implements ErrorHandler {
}
if (this.errorChannel != null) {
try {
this.errorChannel.send(new ErrorMessage(t), this.sendTimeout);
if (this.errorChannel instanceof BlockingChannel && this.sendTimeout >= 0) {
((BlockingChannel) this.errorChannel).send(new ErrorMessage(t), this.sendTimeout);
}
else {
this.errorChannel.send(new ErrorMessage(t));
}
}
catch (Throwable ignore) { // message will be logged only
}

5
org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java
View File

@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.BlockingChannel;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
@@ -70,7 +71,9 @@ public class MessageStoringInterceptor extends ChannelInterceptorAdapter {
}
List<Message<?>> storedMessages = this.messageStore.list();
for (Message<?> message : storedMessages) {
if (!channel.send(message, 0)) {
boolean sent = (channel instanceof BlockingChannel)
? ((BlockingChannel) channel).send(message, 0) : channel.send(message);
if (!sent) {
throw new MessagingException("failed to initialize channel from MessageStore");
}
}

30
org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java
View File

@@ -20,11 +20,11 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.BlockingChannel;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.util.Assert;
@@ -38,7 +38,7 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle {
private static final Log logger = LogFactory.getLog(WireTap.class);
private final MessageTarget target;
private final MessageChannel channel;
private volatile long timeout = 0;
@@ -50,22 +50,22 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle {
/**
* Create a new wire tap with <em>no</em> {@link MessageSelector}.
*
* @param target the MessageTarget to which intercepted messages will be sent
* @param channel the MessageChannel to which intercepted messages will be sent
*/
public WireTap(MessageTarget target) {
this(target, null);
public WireTap(MessageChannel channel) {
this(channel, null);
}
/**
* Create a new wire tap with the provided {@link MessageSelector}.
*
* @param target the target to which intercepted messages will be sent
* @param selector the selector that must accept a message for it to
* be sent to the intercepting target
* @param channel the channel to which intercepted messages will be sent
* @param selector the selector that must accept a message for it to be
* sent to the intercepting channel
*/
public WireTap(MessageTarget target, MessageSelector selector) {
Assert.notNull(target, "target must not be null");
this.target = target;
public WireTap(MessageChannel channel, MessageSelector selector) {
Assert.notNull(channel, "channel must not be null");
this.channel = channel;
this.selector = selector;
}
@@ -110,11 +110,11 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
if (this.running && (this.selector == null || this.selector.accept(message))) {
boolean sent = (this.target instanceof BlockingTarget)
? ((BlockingTarget) this.target).send(message, this.timeout)
: this.target.send(message);
boolean sent = (this.channel instanceof BlockingChannel && this.timeout >= 0)
? ((BlockingChannel) this.channel).send(message, this.timeout)
: this.channel.send(message);
if (!sent && logger.isWarnEnabled()) {
logger.warn("failed to send message to WireTap target '" + this.target + "'");
logger.warn("failed to send message to WireTap channel '" + this.channel + "'");
}
}
return message;

3
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java
View File

@@ -95,8 +95,7 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
else {
builder.addPropertyReference("source", inputChannel);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(
builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, SELECTOR_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, ERROR_HANDLER_ATTRIBUTE);
Element interceptorsElement = DomUtils.getChildElementByTagName(element, INTERCEPTORS_ELEMENT);

7
org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
View File

@@ -81,11 +81,10 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser {
adapterBuilder.addPropertyReference("source", source);
}
if (StringUtils.hasText(channelName)) {
adapterBuilder.addPropertyReference("target", channelName);
adapterBuilder.addPropertyReference("outputChannel", channelName);
}
else {
adapterBuilder.addPropertyReference("target",
this.createDirectChannel(element, parserContext));
adapterBuilder.addPropertyReference("outputChannel", this.createDirectChannel(element, parserContext));
}
}
else if (StringUtils.hasText(target)) {
@@ -96,7 +95,7 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser {
target = BeanDefinitionReaderUtils.registerWithGeneratedName(invokerBuilder.getBeanDefinition(), parserContext.getRegistry());
}
adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(OutboundChannelAdapter.class);
adapterBuilder.addPropertyReference("target", target);
adapterBuilder.addConstructorArgReference(target);
if (pollerElement != null) {
if (!StringUtils.hasText(channelName)) {
throw new ConfigurationException("outbound channel-adapter with a 'poller' requires a 'channel' to poll");

2
org.springframework.integration/src/main/java/org/springframework/integration/config/ResequencerParser.java
View File

@@ -52,7 +52,7 @@ public class ResequencerParser extends AbstractSimpleBeanDefinitionParser {
@Override
protected void postProcess(BeanDefinitionBuilder builder, Element element) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, INPUT_CHANNEL_ATTRIBUTE, "source");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "target");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
}

2
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -108,7 +108,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
if (outputChannel == null) {
throw new ConfigurationException("unable to resolve outputChannel '" + outputChannelName + "'");
}
endpoint.setTarget(outputChannel);
endpoint.setOutputChannel(outputChannel);
}
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java
View File

@@ -91,14 +91,13 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
PollingDispatcher poller = this.createPoller(source, pollerAnnotation);
InboundChannelAdapter adapter = new InboundChannelAdapter();
adapter.setSource(poller);
adapter.setTarget(channel);
adapter.setOutputChannel(channel);
adapter.setBeanName(this.generateUniqueName(channel.getName() + ".inboundAdapter"));
return adapter;
}
private OutboundChannelAdapter createOutboundChannelAdapter(MethodInvokingTarget target, MessageChannel channel, Poller pollerAnnotation) {
OutboundChannelAdapter adapter = new OutboundChannelAdapter();
adapter.setTarget(target);
OutboundChannelAdapter adapter = new OutboundChannelAdapter(target);
if (channel instanceof PollableChannel) {
PollingDispatcher poller = (pollerAnnotation != null)
? this.createPoller((PollableChannel) channel, pollerAnnotation)

11
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -24,7 +24,6 @@ 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;
/**
@@ -40,11 +39,11 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
private volatile TaskExecutor taskExecutor;
private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate();
public void setSendTimeout(long sendTimeout) {
this.messageExchangeTemplate.setSendTimeout(sendTimeout);
// TODO: dispatcher should not implement channel, need to move TX support into the poller
// so that the messageExchangeTemplate is not required for sending to a dispatcher
public String getName() {
return "dispatcher";
}
public boolean subscribe(MessageTarget target) {
@@ -72,7 +71,7 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
* A convenience method for subclasses to send a Message to a single target.
*/
protected final boolean sendMessageToTarget(Message<?> message, MessageTarget target) {
return this.messageExchangeTemplate.send(message, target);
return target.send(message);
}
public String toString() {

9
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.SubscribableSource;
@@ -25,16 +26,10 @@ import org.springframework.integration.message.SubscribableSource;
*
* @author Mark Fisher
*/
public interface MessageDispatcher extends MessageTarget, SubscribableSource {
public interface MessageDispatcher extends MessageChannel, SubscribableSource {
boolean send(Message<?> message);
/**
* Specify the timeout for sending to a target (in milliseconds).
* Note that this value will only be applicable for blocking targets.
*/
void setSendTimeout(long sendTimeout);
boolean subscribe(MessageTarget target);
boolean unsubscribe(MessageTarget target);

10
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java
View File

@@ -20,7 +20,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.BlockingTarget;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollableSource;
@@ -87,15 +86,6 @@ public class PollingDispatcher implements SchedulableTask, SubscribableSource {
this.messageExchangeTemplate.setReceiveTimeout(receiveTimeout);
}
/**
* Specify the timeout to use when sending to a target (in milliseconds).
* Note that this value will only be applicable if the target is an instance
* of {@link BlockingTarget}.
*/
public void setSendTimeout(long sendTimeout) {
this.dispatcher.setSendTimeout(sendTimeout);
}
/**
* Set the maximum number of messages to receive for each poll.
* A non-positive value indicates that polling should repeat as long

12
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -22,11 +22,11 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
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.MessageHandlingException;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.util.ErrorHandler;
@@ -43,7 +43,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
private MessageSource<?> source;
private MessageTarget target;
private MessageChannel outputChannel;
private volatile ErrorHandler errorHandler;
@@ -71,12 +71,12 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
this.source = source;
}
public MessageTarget getTarget() {
return this.target;
public MessageChannel getOutputChannel() {
return this.outputChannel;
}
public void setTarget(MessageTarget target) {
this.target = target;
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
protected ChannelRegistry getChannelRegistry() {

34
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractInOutEndpoint.java
View File

@@ -21,13 +21,13 @@ import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.CompositeMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.selector.MessageSelector;
@@ -82,17 +82,17 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
return true;
}
Message<?> reply = buildReplyMessage(result, message.getHeaders());
MessageTarget replyTarget = this.resolveReplyTarget(message);
MessageChannel replyChannel = this.resolveReplyChannel(message);
if (reply instanceof CompositeMessage && this.shouldSplitComposite()) {
boolean sentAtLeastOne = false;
for (Message<?> nextReply : (CompositeMessage) reply) {
boolean sent = this.sendReplyMessage(nextReply, replyTarget);
boolean sent = this.sendReplyMessage(nextReply, replyChannel);
sentAtLeastOne = (sentAtLeastOne || sent);
}
return sentAtLeastOne;
}
else {
return this.sendReplyMessage(reply, replyTarget);
return this.sendReplyMessage(reply, replyChannel);
}
}
@@ -112,7 +112,7 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
return false;
}
private boolean sendReplyMessage(Message<?> replyMessage, MessageTarget replyTarget) {
private boolean sendReplyMessage(Message<?> replyMessage, MessageChannel replyChannel) {
for (int i = this.interceptors.size() - 1; i >= 0; i--) {
EndpointInterceptor interceptor = this.interceptors.get(i);
if (interceptor != null) {
@@ -122,7 +122,7 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
}
}
}
return this.getMessageExchangeTemplate().send(replyMessage, replyTarget);
return this.getMessageExchangeTemplate().send(replyMessage, replyChannel);
}
private Message<?> buildReplyMessage(Object result, MessageHeaders requestHeaders) {
@@ -149,26 +149,30 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
.build();
}
private MessageTarget resolveReplyTarget(Message<?> requestMessage) {
MessageTarget replyTarget = this.getTarget();
if (replyTarget == null) {
private MessageChannel resolveReplyChannel(Message<?> requestMessage) {
MessageChannel replyChannel = this.getOutputChannel();
if (replyChannel == null) {
Object returnAddress = requestMessage.getHeaders().getReturnAddress();
if (returnAddress != null) {
if (returnAddress instanceof MessageTarget) {
replyTarget = (MessageTarget) returnAddress;
if (returnAddress instanceof MessageChannel) {
replyChannel = (MessageChannel) returnAddress;
}
else if (returnAddress instanceof String) {
ChannelRegistry channelRegistry = this.getChannelRegistry();
if (channelRegistry != null) {
replyTarget = channelRegistry.lookupChannel((String) returnAddress);
replyChannel = channelRegistry.lookupChannel((String) returnAddress);
}
}
else {
throw new MessagingException("expected a MessageChannel or String for 'returnAddress', but type is ["
+ returnAddress.getClass() + "]");
}
}
}
if (replyTarget == null) {
throw new MessagingException("unable to resolve reply target");
if (replyChannel == null) {
throw new MessagingException("unable to resolve reply channel");
}
return replyTarget;
return replyChannel;
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java
View File

@@ -34,7 +34,7 @@ public class InboundChannelAdapter extends AbstractEndpoint {
@Override
protected boolean sendInternal(Message<?> message) {
try {
boolean sent = this.getMessageExchangeTemplate().send(message, this.getTarget());
boolean sent = this.getMessageExchangeTemplate().send(message, this.getOutputChannel());
if (sent && this.getSource() instanceof MessageDeliveryAware) {
((MessageDeliveryAware) this.getSource()).onSend(message);
}

13
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessagingBridge.java
View File

@@ -17,12 +17,23 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class MessagingBridge extends AbstractRequestReplyEndpoint {
private final MessageTarget target;
public MessagingBridge(MessageTarget target) {
Assert.notNull(target, "target must not be null");
this.target = target;
}
@Override
protected Message<?> handleRequestMessage(Message<?> requestMessage) {
return requestMessage;
@@ -35,7 +46,7 @@ public class MessagingBridge extends AbstractRequestReplyEndpoint {
@Override
protected void sendReplyMessage(Message<?> replyMessage, Message<?> requestMessage) {
this.getMessageExchangeTemplate().send(replyMessage, this.getTarget());
this.target.send(replyMessage);
}
}

13
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java
View File

@@ -18,6 +18,8 @@ package org.springframework.integration.endpoint;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
/**
* A Channel Adapter implementation for connecting a {@link MessageChannel}
@@ -27,9 +29,18 @@ import org.springframework.integration.message.Message;
*/
public class OutboundChannelAdapter extends AbstractEndpoint {
private final MessageTarget target;
public OutboundChannelAdapter(MessageTarget target) {
Assert.notNull(target, "target must not be null");
this.target = target;
}
@Override
protected boolean sendInternal(Message<?> message) {
return this.getMessageExchangeTemplate().send(message, this.getTarget());
return this.target.send(message);
}
}

22
org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessageExchangeTemplate.java
View File

@@ -20,9 +20,7 @@ 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.MessageTarget;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.util.Assert;
/**
@@ -42,30 +40,30 @@ public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate {
/**
* Send the provided message to the given target. Note that the actual
* Send the provided message to the given channel. Note that the actual
* sending occurs asynchronously, so this method will always return
* <code>true</code> unless an exception is thrown by the executor.
*/
@Override
public boolean send(final Message<?> message, final MessageTarget target) {
public boolean send(final Message<?> message, final MessageChannel channel) {
this.taskExecutor.execute(new Runnable() {
public void run() {
AsyncMessageExchangeTemplate.super.send(message, target);
AsyncMessageExchangeTemplate.super.send(message, channel);
}
});
return true;
}
/**
* Send the provided message to the given target and receive the
* Send the provided message to the given channel and receive the
* result as an {@link AsyncMessage}.
*/
@Override
@SuppressWarnings("unchecked")
public Message<?> sendAndReceive(final Message<?> request, final MessageTarget target) {
public Message<?> sendAndReceive(final Message<?> request, final MessageChannel channel) {
FutureTask<Message<?>> task = new FutureTask<Message<?>>(new Callable<Message<?>>() {
public Message<?> call() throws Exception {
return AsyncMessageExchangeTemplate.super.sendAndReceive(request, target);
return AsyncMessageExchangeTemplate.super.sendAndReceive(request, channel);
}
});
this.taskExecutor.execute(task);
@@ -89,15 +87,15 @@ public class AsyncMessageExchangeTemplate extends MessageExchangeTemplate {
/**
* Receive a Message from the provided source and if not <code>null</code>,
* send it to the given target. Note that the receive and send operations
* send it to the given channel. Note that the receive and send operations
* occur asynchronously, so this method will always return <code>true</code>
* unless an exception is thrown by the executor.
*/
@Override
public boolean receiveAndForward(final PollableSource<?> source, final MessageTarget target) {
public boolean receiveAndForward(final PollableSource<?> source, final MessageChannel channel) {
this.taskExecutor.execute(new Runnable() {
public void run() {
AsyncMessageExchangeTemplate.super.receiveAndForward(source, target);
AsyncMessageExchangeTemplate.super.receiveAndForward(source, channel);
}
});
return true;

21
org.springframework.integration/src/main/java/org/springframework/integration/message/BlockingTarget.java
View File

@@ -23,27 +23,6 @@ package org.springframework.integration.message;
*/
public interface BlockingTarget extends MessageTarget {
/**
* Send a message, blocking indefinitely if necessary.
*
* @param message the {@link Message} to send
*
* @return <code>true</code> if the message is sent successfully,
* <code>false</false> if interrupted
*/
boolean send(Message<?> message);
/**
* Send a message, blocking until either the message is accepted or the
* specified timeout period elapses.
*
* @param message the {@link Message} to send
* @param timeout the timeout in milliseconds
*
* @return <code>true</code> if the message is sent successfully,
* <code>false</false> if the specified timeout period elapses or
* the send is interrupted
*/
boolean send(Message<?> message, long timeout);
}

3
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java
View File

@@ -21,6 +21,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -156,7 +157,7 @@ public final class MessageBuilder<T> {
return this.setHeader(MessageHeaders.CORRELATION_ID, correlationId);
}
public MessageBuilder<T> setReturnAddress(MessageTarget returnAddress) {
public MessageBuilder<T> setReturnAddress(MessageChannel returnAddress) {
return this.setHeader(MessageHeaders.RETURN_ADDRESS, returnAddress);
}

55
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java
View File

@@ -23,11 +23,8 @@ 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.MessageTarget;
import org.springframework.integration.channel.BlockingChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
@@ -36,9 +33,9 @@ import org.springframework.util.Assert;
/**
* This is the central class for invoking message exchange operations
* across {@link PollableSource}s and {@link MessageTarget}s. It supports
* across {@link PollableSource}s and {@link MessageChannel}s. It supports
* one-way send and receive calls as well as request/reply. Additionally,
* the {@link #receiveAndForward(PollableSource, MessageTarget)} method
* the {@link #receiveAndForward(PollableSource, MessageChannel)} method
* plays the role of a polling-consumer while actually sending any
* received message to an event-driven consumer.
*
@@ -143,28 +140,28 @@ public class MessageExchangeTemplate implements InitializingBean {
}
}
public boolean send(final Message<?> message, final MessageTarget target) {
public boolean send(final Message<?> message, final MessageChannel channel) {
TransactionTemplate txTemplate = this.getTransactionTemplate();
if (txTemplate != null) {
return (Boolean) txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return doSend(message, target);
return doSend(message, channel);
}
});
}
return this.doSend(message, target);
return this.doSend(message, channel);
}
public Message<?> sendAndReceive(final Message<?> request, final MessageTarget target) {
public Message<?> sendAndReceive(final Message<?> request, final MessageChannel channel) {
TransactionTemplate txTemplate = this.getTransactionTemplate();
if (txTemplate != null) {
return (Message<?>) txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return doSendAndReceive(request, target);
return doSendAndReceive(request, channel);
}
});
}
return this.doSendAndReceive(request, target);
return this.doSendAndReceive(request, channel);
}
public Message<?> receive(final PollableSource<?> source) {
@@ -179,27 +176,27 @@ public class MessageExchangeTemplate implements InitializingBean {
return this.doReceive(source);
}
public boolean receiveAndForward(final PollableSource<?> source, final MessageTarget target) {
public boolean receiveAndForward(final PollableSource<?> source, final MessageChannel channel) {
TransactionTemplate txTemplate = this.getTransactionTemplate();
if (txTemplate != null) {
return (Boolean) txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return doReceiveAndForward(source, target);
return doReceiveAndForward(source, channel);
}
});
}
return this.doReceiveAndForward(source, target);
return this.doReceiveAndForward(source, channel);
}
private boolean doSend(Message<?> message, MessageTarget target) {
Assert.notNull(target, "target must not be null");
private boolean doSend(Message<?> message, MessageChannel channel) {
Assert.notNull(channel, "channel must not be null");
long timeout = this.sendTimeout;
boolean sent = (timeout >= 0 && target instanceof BlockingTarget)
? ((BlockingTarget) target).send(message, timeout)
: target.send(message);
boolean sent = (timeout >= 0 && channel instanceof BlockingChannel)
? ((BlockingChannel) channel).send(message, timeout)
: channel.send(message);
if (!sent && this.logger.isTraceEnabled()) {
this.logger.trace("failed to send message to target '" + target + "' within timeout: " + timeout);
this.logger.trace("failed to send message to channel '" + channel + "' within timeout: " + timeout);
}
return sent;
}
@@ -216,23 +213,23 @@ public class MessageExchangeTemplate implements InitializingBean {
return message;
}
private Message<?> doSendAndReceive(Message<?> request, MessageTarget target) {
private Message<?> doSendAndReceive(Message<?> request, MessageChannel channel) {
TemporaryReturnAddress returnAddress = new TemporaryReturnAddress(this.receiveTimeout);
request = MessageBuilder.fromMessage(request).setReturnAddress(returnAddress).build();
if (!this.doSend(request, target)) {
if (!this.doSend(request, channel)) {
return null;
}
return this.doReceive(returnAddress);
}
private boolean doReceiveAndForward(PollableSource<?> source, MessageTarget target) {
private boolean doReceiveAndForward(PollableSource<?> source, MessageChannel channel) {
Message<?> message = null;
try {
message = this.doReceive(source);
if (message == null) {
return false;
}
boolean sent = this.doSend(message, target);
boolean sent = this.doSend(message, channel);
if (source instanceof MessageDeliveryAware) {
if (sent) {
((MessageDeliveryAware) source).onSend(message);
@@ -260,7 +257,7 @@ public class MessageExchangeTemplate implements InitializingBean {
@SuppressWarnings("unchecked")
private static class TemporaryReturnAddress implements BlockingSource, MessageTarget {
private static class TemporaryReturnAddress implements BlockingSource, MessageChannel {
private volatile Message<?> message;
@@ -274,6 +271,10 @@ public class MessageExchangeTemplate implements InitializingBean {
}
public String getName() {
return "temporaryReplyChannel";
}
public Message receive() {
return this.receive(-1);
}