diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 7e2c3e7670..540617a2e7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -34,7 +34,7 @@ import org.springframework.integration.core.MessageChannel; * * @author Mark Fisher */ -public abstract class AbstractMessageChannel implements BlockingChannel, BeanNameAware { +public abstract class AbstractMessageChannel implements MessageChannel, BeanNameAware { private final Log logger = LogFactory.getLog(this.getClass()); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/BlockingChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/BlockingChannel.java deleted file mode 100644 index d2caf4ea01..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/BlockingChannel.java +++ /dev/null @@ -1,53 +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.channel; - -import org.springframework.integration.core.Message; -import org.springframework.integration.core.MessageChannel; - -/** - * 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 true if the message is sent successfully, - * 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 true if the message is sent successfully, - * false if the specified timeout period elapses or - * the send is interrupted - */ - boolean send(Message message, long timeout); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java index c21fef7ede..2c4dac235d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java @@ -217,8 +217,8 @@ public class MessageChannelTemplate implements InitializingBean { private boolean doSend(Message message, MessageChannel channel) { Assert.notNull(channel, "channel must not be null"); long timeout = this.sendTimeout; - boolean sent = (timeout >= 0 && channel instanceof BlockingChannel) - ? ((BlockingChannel) channel).send(message, timeout) + boolean sent = (timeout >= 0) + ? channel.send(message, timeout) : channel.send(message); if (!sent && this.logger.isTraceEnabled()) { this.logger.trace("failed to send message to channel '" + channel + "' within timeout: " + timeout); @@ -294,6 +294,10 @@ public class MessageChannelTemplate implements InitializingBean { } public boolean send(Message message) { + return this.send(message, -1); + } + + public boolean send(Message message, long timeout) { this.message = message; this.latch.countDown(); return true; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java index 5e7e69054d..8dc2872899 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java @@ -63,8 +63,8 @@ public class MessagePublishingErrorHandler implements ErrorHandler { } if (this.errorChannel != null) { try { - if (this.errorChannel instanceof BlockingChannel && this.sendTimeout >= 0) { - ((BlockingChannel) this.errorChannel).send(new ErrorMessage(t), this.sendTimeout); + if (this.sendTimeout >= 0) { + this.errorChannel.send(new ErrorMessage(t), this.sendTimeout); } else { this.errorChannel.send(new ErrorMessage(t)); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java index 2aaa634003..30b5235317 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java @@ -21,7 +21,6 @@ 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.core.Message; import org.springframework.integration.core.MessageChannel; @@ -71,8 +70,7 @@ public class MessageStoringInterceptor extends ChannelInterceptorAdapter { } List> storedMessages = this.messageStore.list(); for (Message message : storedMessages) { - boolean sent = (channel instanceof BlockingChannel) - ? ((BlockingChannel) channel).send(message, 0) : channel.send(message); + boolean sent = channel.send(message, 0); if (!sent) { throw new MessagingException("failed to initialize channel from MessageStore"); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java index 4e7d21928f..235ff33b1e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java @@ -20,7 +20,6 @@ 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.core.Message; import org.springframework.integration.core.MessageChannel; @@ -109,8 +108,8 @@ 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.channel instanceof BlockingChannel && this.timeout >= 0) - ? ((BlockingChannel) this.channel).send(message, this.timeout) + boolean sent = (this.timeout >= 0) + ? this.channel.send(message, this.timeout) : this.channel.send(message); if (!sent && logger.isWarnEnabled()) { logger.warn("failed to send message to WireTap channel '" + this.channel + "'"); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageChannel.java index 889d5b4999..ece40d5951 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageChannel.java @@ -29,13 +29,31 @@ public interface MessageChannel { 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'. + * 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'. + * + *

Depending on the implementation, this method may block indefinitely. + * To provide a maximum wait time, use {@link #send(Message, long)}. + * + * @param message the {@link Message} to send * - * @param message the Message to send * @return whether the Message has been sent successfully */ 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 true if the message is sent successfully, + * false if the specified timeout period elapses or + * the send is interrupted + */ + boolean send(Message message, long timeout); + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java index f9939fe5db..2189b9be27 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java @@ -206,12 +206,10 @@ public class MessageChannelTemplateTests { public void sendWithReturnAddress() throws InterruptedException { final List replies = new ArrayList(3); final CountDownLatch latch = new CountDownLatch(3); - MessageChannel replyChannel = new MessageChannel() { - public String getName() { - return "testReplyChannel"; - } - public boolean send(Message replyMessage) { - replies.add((String) replyMessage.getPayload()); + MessageChannel replyChannel = new AbstractMessageChannel() { + @Override + protected boolean doSend(Message message, long timeout) { + replies.add((String) message.getPayload()); latch.countDown(); return true; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java index a31df69e49..500e88356d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java @@ -26,7 +26,6 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.FatalBeanException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.BlockingChannel; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.PublishSubscribeChannel; @@ -53,7 +52,7 @@ public class ChannelParserTests { public void testChannelWithCapacity() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "channelParserTests.xml", this.getClass()); - BlockingChannel channel = (BlockingChannel) context.getBean("capacityChannel"); + MessageChannel channel = (MessageChannel) context.getBean("capacityChannel"); for (int i = 0; i < 10; i++) { boolean result = channel.send(new GenericMessage("test"), 10); assertTrue(result);