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

Removed BlockingChannel interface. The send(message, timeout) method is now defined directly in the top-level MessageChannel interface.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-15 16:49:00 +00:00
parent 0afc835679
commit 11e3fbb584
9 changed files with 39 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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());

53
org.springframework.integration/src/main/java/org/springframework/integration/channel/BlockingChannel.java
View File

@@ -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 <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);
}

8
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java
View File

@@ -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;

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

@@ -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));

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

@@ -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<Message<?>> 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");
}

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

@@ -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 + "'");

26
org.springframework.integration/src/main/java/org/springframework/integration/core/MessageChannel.java
View File

@@ -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'.
*
* <p>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 <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);
}

10
org.springframework.integration/src/test/java/org/springframework/integration/channel/MessageChannelTemplateTests.java
View File

@@ -206,12 +206,10 @@ public class MessageChannelTemplateTests {
public void sendWithReturnAddress() throws InterruptedException {
final List<String> replies = new ArrayList<String>(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;
}

3
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java
View File

@@ -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<String>("test"), 10);
assertTrue(result);