From 472aa0f436515963a42f114d4fe9b2b0bc3ecd4d Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 31 Jan 2008 16:11:02 +0000 Subject: [PATCH] Added basic ChannelInterceptor functionality and 'unregister' method for ChannelRegistry. --- .../integration/bus/MessageBus.java | 11 + .../channel/AbstractMessageChannel.java | 237 ++++++++++++++++++ .../channel/ChannelInterceptor.java | 38 +++ .../channel/ChannelInterceptorAdapter.java | 43 ++++ .../integration/channel/ChannelRegistry.java | 2 + .../channel/DefaultChannelRegistry.java | 4 + .../integration/channel/SimpleChannel.java | 116 ++------- .../channel/ChannelInterceptorTests.java | 201 +++++++++++++++ .../channel/DefaultChannelRegistryTests.java | 92 +++++++ 9 files changed, 642 insertions(+), 102 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAdapter.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelInterceptorTests.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 08882084e1..54acaf017a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -191,6 +191,17 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } + public MessageChannel unregisterChannel(String name) { + MessageChannel removedChannel = this.channelRegistry.unregisterChannel(name); + if (removedChannel != null) { + MessageDispatcher removedDispatcher = this.dispatchers.remove(removedChannel); + if (removedDispatcher != null && removedDispatcher.isRunning()) { + removedDispatcher.stop(); + } + } + return removedChannel; + } + public void registerHandler(String name, MessageHandler handler, Subscription subscription) { this.registerHandler(name, handler, subscription, null); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java new file mode 100644 index 0000000000..ec41550b4f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -0,0 +1,237 @@ +/* + * Copyright 2002-2007 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 java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.integration.dispatcher.DispatcherPolicy; +import org.springframework.integration.message.Message; + +/** + * Base class for {@link MessageChannel} implementations providing common + * properties such as the channel name and {@link DispatcherPolicy}. Also + * provides the common functionality for sending and receiving + * {@link Message Messages} including the invocation of any + * {@link ChannelInterceptor ChannelInterceptors}. + * + * @author Mark Fisher + */ +public abstract class AbstractMessageChannel implements MessageChannel, BeanNameAware { + + private String name; + + private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); + + private final DispatcherPolicy dispatcherPolicy; + + + /** + * Create a channel with the given dispatcher policy. + */ + public AbstractMessageChannel(DispatcherPolicy dispatcherPolicy) { + this.dispatcherPolicy = dispatcherPolicy; + } + + + /** + * Set the name of this channel. + */ + public void setName(String name) { + this.name = name; + } + + /** + * Return the name of this channel. + */ + public String getName() { + return this.name; + } + + /** + * Set the name of this channel to its bean name. This will be invoked + * automatically whenever the channel is configured explicitly with a bean + * definition. + */ + public void setBeanName(String beanName) { + this.setName(beanName); + } + + /** + * Set the list of channel interceptors. This will clear any existing + * interceptors. + */ + public void setInterceptors(List interceptors) { + this.interceptors.set(interceptors); + } + + /** + * Add a channel interceptor to the end of the list. + */ + public void addInterceptor(ChannelInterceptor interceptor) { + this.interceptors.add(interceptor); + } + + /** + * Return the dispatcher policy for this channel. + */ + public DispatcherPolicy getDispatcherPolicy() { + return this.dispatcherPolicy; + } + + /** + * Send a message on this channel. If the channel is at capacity, this + * method will block until either space becomes available or the sending + * thread is interrupted. + * + * @param message the Message to send + * + * @return true if the message is sent successfully or + * false if the sending thread is interrupted. + */ + public final boolean send(Message message) { + return this.send(message, -1); + } + + /** + * Send a message on this channel. If the channel is at capacity, this + * method will block until either the timeout occurs or the sending thread + * is interrupted. If the specified timeout is 0, the method will return + * immediately. If less than zero, it will block indefinitely (see + * {@link #send(Message)}). + * + * @param message the Message to send + * @param timeout the timeout in milliseconds + * + * @return true if the message is sent successfully, + * false if the message cannot be sent within the allotted + * time or the sending thread is interrupted. + */ + public final boolean send(Message message, long timeout) { + if (!this.interceptors.preSend(message, this)) { + return false; + } + boolean sent = this.doSend(message, timeout); + this.interceptors.postSend(message, this, sent); + return sent; + } + + /** + * Receive the first available message from this channel. If the channel + * contains no messages, this method will block. + * + * @return the first available message or null if the + * receiving thread is interrupted. + */ + public final Message receive() { + return this.receive(-1); + } + + /** + * Receive the first available message from this channel. If the channel + * contains no messages, this method will block until the allotted timeout + * elapses. If the specified timeout is 0, the method will return + * immediately. If less than zero, it will block indefinitely (see + * {@link #receive()}). + * + * @param timeout the timeout in milliseconds + * + * @return the first available message or null if no message + * is available within the allotted time or the receiving thread is + * interrupted. + */ + public final Message receive(long timeout) { + if (!this.interceptors.preReceive(this)) { + return null; + } + Message message = this.doReceive(timeout); + this.interceptors.postReceive(message, this); + return message; + } + + + /** + * Subclasses must implement this method. A non-negative timeout indicates + * how long to wait if the channel is at capacity (if the value is 0, it + * must return immediately with or without success). A negative timeout + * value indicates that the method should block until either the message is + * accepted or the blocking thread is interrupted. + */ + public abstract boolean doSend(Message message, long timeout); + + /** + * Subclasses must implement this method. A non-negative timeout indicates + * how long to wait if the channel is empty (if the value is 0, it must + * return immediately with or without success). A negative timeout value + * indicates that the method should block until either a message is + * available or the blocking thread is interrupted. + */ + public abstract Message doReceive(long timeout); + + + /** + * A convenience wrapper class for the list of ChannelInterceptors. + */ + private static class ChannelInterceptorList { + + private final List interceptors = new CopyOnWriteArrayList(); + + + public boolean set(List interceptors) { + synchronized (this.interceptors) { + this.interceptors.clear(); + return this.interceptors.addAll(interceptors); + } + } + + public boolean add(ChannelInterceptor interceptor) { + return this.interceptors.add(interceptor); + } + + public boolean preSend(Message message, MessageChannel channel) { + for (ChannelInterceptor interceptor : interceptors) { + if (!interceptor.preSend(message, channel)) { + return false; + } + } + return true; + } + + public void postSend(Message message, MessageChannel channel, boolean sent) { + for (ChannelInterceptor interceptor : interceptors) { + interceptor.postSend(message, channel, sent); + } + } + + public boolean preReceive(MessageChannel channel) { + for (ChannelInterceptor interceptor : interceptors) { + if (!interceptor.preReceive(channel)) { + return false; + } + } + return true; + } + + public void postReceive(Message message, MessageChannel channel) { + for (ChannelInterceptor interceptor : interceptors) { + interceptor.postReceive(message, channel); + } + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java new file mode 100644 index 0000000000..000c4517e4 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java @@ -0,0 +1,38 @@ +/* + * Copyright 2002-2007 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; + +/** + * Interface for interceptors that are able to view and/or modify the + * {@link Message Messages} being sent-to and/or received-from a + * {@link MessageChannel}. + * + * @author Mark Fisher + */ +public interface ChannelInterceptor { + + boolean preSend(Message message, MessageChannel channel); + + void postSend(Message message, MessageChannel channel, boolean sent); + + boolean preReceive(MessageChannel channel); + + void postReceive(Message message, MessageChannel channel); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAdapter.java new file mode 100644 index 0000000000..7ef293efe0 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAdapter.java @@ -0,0 +1,43 @@ +/* + * Copyright 2002-2007 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; + +/** + * A {@link ChannelInterceptor} with no-op method implementations so that + * subclasses do not have to implement all of the interface's methods. + * + * @author Mark Fisher + */ +public class ChannelInterceptorAdapter implements ChannelInterceptor { + + public boolean preSend(Message message, MessageChannel channel) { + return true; + } + + public void postSend(Message message, MessageChannel channel, boolean sent) { + } + + public boolean preReceive(MessageChannel channel) { + return true; + } + + public void postReceive(Message message, MessageChannel channel) { + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java index 307bc4f532..98e31f6af1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java @@ -25,6 +25,8 @@ public interface ChannelRegistry { void registerChannel(String name, MessageChannel channel); + MessageChannel unregisterChannel(String name); + MessageChannel lookupChannel(String channelName); void setErrorChannel(MessageChannel errorChannel); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java index 145c753d28..4e95c9b071 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java @@ -51,4 +51,8 @@ public class DefaultChannelRegistry implements ChannelRegistry { this.channels.put(name, channel); } + public MessageChannel unregisterChannel(String name) { + return (name != null) ? this.channels.remove(name) : null; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java index 64f6665b01..98d3c8b0b8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java @@ -20,7 +20,6 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; -import org.springframework.beans.factory.BeanNameAware; import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.message.Message; import org.springframework.util.Assert; @@ -32,15 +31,11 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class SimpleChannel implements MessageChannel, BeanNameAware { +public class SimpleChannel extends AbstractMessageChannel { public static final int DEFAULT_CAPACITY = 100; - private String name; - - private final DispatcherPolicy dispatcherPolicy; - private BlockingQueue> queue; @@ -48,8 +43,8 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { * Create a channel with the specified queue capacity and dispatcher policy. */ public SimpleChannel(int capacity, DispatcherPolicy dispatcherPolicy) { + super((dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy()); this.queue = new LinkedBlockingQueue>(capacity); - this.dispatcherPolicy = (dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy(); } /** @@ -74,46 +69,15 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { } - /** - * Set the name of this channel. - */ - public void setName(String name) { - this.name = name; - } - - /** - * Return the name of this channel. - */ - public String getName() { - return this.name; - } - - public DispatcherPolicy getDispatcherPolicy() { - return this.dispatcherPolicy; - } - - /** - * Set the name of this channel to its bean name. This will be invoked - * automatically whenever the channel is configured explicitly with a bean - * definition. - */ - public void setBeanName(String beanName) { - this.setName(beanName); - } - - /** - * Send a message on this channel. If the queue is full, this method will - * block until either space becomes available or the sending thread is - * interrupted. - * - * @param message the Message to send - * - * @return true if the message is sent successfully or - * false if the sending thread is interrupted. - */ - public boolean send(Message message) { + public boolean doSend(Message message, long timeout) { Assert.notNull(message, "'message' must not be null"); try { + if (timeout > 0) { + return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS); + } + if (timeout == 0) { + return this.queue.offer(message); + } queue.put(message); return true; } @@ -123,42 +87,14 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { } } - /** - * Send a message on this channel. If the queue is full, this method will - * block until either the timeout occurs or the sending thread is - * interrupted. If the specified timeout is less than 1, the method will - * return immediately. - * - * @param message the Message to send - * @param timeout the timeout in milliseconds - * - * @return true if the message is sent successfully, - * false if the message cannot be sent within the allotted - * time or the sending thread is interrupted. - */ - public boolean send(Message message, long timeout) { - Assert.notNull(message, "'message' must not be null"); + public Message doReceive(long timeout) { try { if (timeout > 0) { - return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS); + return queue.poll(timeout, TimeUnit.MILLISECONDS); + } + if (timeout == 0) { + return queue.poll(); } - return this.queue.offer(message); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return false; - } - } - - /** - * Receive the message at the head of the queue. If the queue is empty, this - * method will block. - * - * @return the Message at the head of the queue or null if - * the receiving thread is interrupted. - */ - public Message receive() { - try { return queue.take(); } catch (InterruptedException e) { @@ -167,28 +103,4 @@ public class SimpleChannel implements MessageChannel, BeanNameAware { } } - /** - * Receive the message at the head of the queue. If the queue is empty, this - * method will block until the allotted timeout elapses. If the specified - * timeout is less than 1, the method will return immediately. - * - * @param timeout the timeout in milliseconds - * - * @return the message at the head of the queue or null in - * case no message is available within the allotted time or the receiving - * thread is interrupted. - */ - public Message receive(long timeout) { - try { - if (timeout > 0) { - return queue.poll(timeout, TimeUnit.MILLISECONDS); - } - return queue.poll(); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return null; - } - } - } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelInterceptorTests.java new file mode 100644 index 0000000000..94b9ce3373 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelInterceptorTests.java @@ -0,0 +1,201 @@ +/* + * Copyright 2002-2007 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class ChannelInterceptorTests { + + private final SimpleChannel channel = new SimpleChannel(); + + + @Test + public void testPreSendInterceptorReturnsTrue() { + channel.addInterceptor(new PreSendReturnsTrueInterceptor()); + channel.send(new StringMessage("test")); + Message result = channel.receive(0); + assertNotNull(result); + assertEquals("test", result.getPayload()); + assertEquals(1, result.getHeader().getAttribute(PreSendReturnsTrueInterceptor.class.getName())); + } + + @Test + public void testPreSendInterceptorReturnsFalse() { + channel.addInterceptor(new PreSendReturnsFalseInterceptor()); + Message message = new StringMessage("test"); + channel.send(message); + assertEquals(1, message.getHeader().getAttribute(PreSendReturnsFalseInterceptor.class.getName())); + Message result = channel.receive(0); + assertNull(result); + } + + @Test + public void testPostSendInterceptorWithSentMessage() { + final AtomicBoolean invoked = new AtomicBoolean(false); + channel.addInterceptor(new ChannelInterceptorAdapter() { + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + assertNotNull(message); + assertNotNull(channel); + assertSame(ChannelInterceptorTests.this.channel, channel); + assertTrue(sent); + invoked.set(true); + } + }); + channel.send(new StringMessage("test")); + assertTrue(invoked.get()); + } + + @Test + public void testPostSendInterceptorWithUnsentMessage() { + final AtomicInteger invokedCounter = new AtomicInteger(0); + final AtomicInteger sentCounter = new AtomicInteger(0); + final SimpleChannel singleItemChannel = new SimpleChannel(1); + singleItemChannel.addInterceptor(new ChannelInterceptorAdapter() { + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + assertNotNull(message); + assertNotNull(channel); + assertSame(singleItemChannel, channel); + if (sent) { + sentCounter.incrementAndGet(); + } + invokedCounter.incrementAndGet(); + } + }); + assertEquals(0, invokedCounter.get()); + assertEquals(0, sentCounter.get()); + singleItemChannel.send(new StringMessage("test1")); + assertEquals(1, invokedCounter.get()); + assertEquals(1, sentCounter.get()); + singleItemChannel.send(new StringMessage("test2"), 0); + assertEquals(2, invokedCounter.get()); + assertEquals(1, sentCounter.get()); + } + + @Test + public void testPreReceiveInterceptorReturnsTrue() { + channel.addInterceptor(new PreReceiveReturnsTrueInterceptor()); + Message message = new StringMessage("test"); + channel.send(message); + Message result = channel.receive(0); + assertEquals(1, PreReceiveReturnsTrueInterceptor.counter.get()); + assertNotNull(result); + } + + @Test + public void testPreReceiveInterceptorReturnsFalse() { + channel.addInterceptor(new PreReceiveReturnsFalseInterceptor()); + Message message = new StringMessage("test"); + channel.send(message); + Message result = channel.receive(0); + assertEquals(1, PreReceiveReturnsFalseInterceptor.counter.get()); + assertNull(result); + } + + @Test + public void testPostReceiveInterceptor() { + final AtomicInteger invokedCount = new AtomicInteger(); + final AtomicInteger messageCount = new AtomicInteger(); + channel.addInterceptor(new ChannelInterceptorAdapter() { + @Override + public void postReceive(Message message, MessageChannel channel) { + assertNotNull(channel); + assertSame(ChannelInterceptorTests.this.channel, channel); + if (message != null) { + messageCount.incrementAndGet(); + } + invokedCount.incrementAndGet(); + } + }); + channel.receive(0); + assertEquals(1, invokedCount.get()); + assertEquals(0, messageCount.get()); + channel.send(new StringMessage("test")); + Message result = channel.receive(0); + assertNotNull(result); + assertEquals(2, invokedCount.get()); + assertEquals(1, messageCount.get()); + } + + + private static class PreSendReturnsTrueInterceptor extends ChannelInterceptorAdapter { + + private static AtomicInteger counter = new AtomicInteger(); + + @Override + public boolean preSend(Message message, MessageChannel channel) { + assertNotNull(message); + message.getHeader().setAttribute(this.getClass().getName(), counter.incrementAndGet()); + return true; + } + } + + + private static class PreSendReturnsFalseInterceptor extends ChannelInterceptorAdapter { + + private static AtomicInteger counter = new AtomicInteger(); + + @Override + public boolean preSend(Message message, MessageChannel channel) { + assertNotNull(message); + message.getHeader().setAttribute(this.getClass().getName(), counter.incrementAndGet()); + return false; + } + } + + + private static class PreReceiveReturnsTrueInterceptor extends ChannelInterceptorAdapter { + + private static AtomicInteger counter = new AtomicInteger(); + + @Override + public boolean preReceive(MessageChannel channel) { + counter.incrementAndGet(); + return true; + } + } + + + private static class PreReceiveReturnsFalseInterceptor extends ChannelInterceptorAdapter { + + private static AtomicInteger counter = new AtomicInteger(); + + @Override + public boolean preReceive(MessageChannel channel) { + counter.incrementAndGet(); + return false; + } + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java new file mode 100644 index 0000000000..e66c5c9767 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java @@ -0,0 +1,92 @@ +/* + * Copyright 2002-2007 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 static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; + +import org.junit.Test; + +/** + * @author Mark Fisher + */ +public class DefaultChannelRegistryTests { + + @Test + public void testLookupRegisteredChannel() { + SimpleChannel testChannel = new SimpleChannel(); + DefaultChannelRegistry registry = new DefaultChannelRegistry(); + registry.registerChannel("testChannel", testChannel); + MessageChannel lookedUpChannel = registry.lookupChannel("testChannel"); + assertNotNull(testChannel); + assertSame(testChannel, lookedUpChannel); + } + + @Test + public void testLookupNeverRegisteredChannel() { + DefaultChannelRegistry registry = new DefaultChannelRegistry(); + MessageChannel noSuchChannel = registry.lookupChannel("noSuchChannel"); + assertNull(noSuchChannel); + } + + @Test + public void testLookupUnregisteredChannel() { + SimpleChannel testChannel = new SimpleChannel(); + DefaultChannelRegistry registry = new DefaultChannelRegistry(); + registry.registerChannel("testChannel", testChannel); + MessageChannel lookedUpChannel1 = registry.lookupChannel("testChannel"); + assertNotNull(lookedUpChannel1); + assertSame(testChannel, lookedUpChannel1); + MessageChannel unregisteredChannel = registry.unregisterChannel("testChannel"); + assertNotNull(unregisteredChannel); + assertSame(testChannel, unregisteredChannel); + MessageChannel lookedUpChannel2 = registry.lookupChannel("testChannel"); + assertNull(lookedUpChannel2); + } + + @Test + public void testUnregisteringChannelThatIsNotInRegistry() { + DefaultChannelRegistry registry = new DefaultChannelRegistry(); + MessageChannel unregisteredChannel = registry.unregisterChannel("noSuchChannel"); + assertNull(unregisteredChannel); + } + + @Test + public void testUnregisteringNullDoesNotThrowException() { + DefaultChannelRegistry registry = new DefaultChannelRegistry(); + MessageChannel unregisteredChannel = registry.unregisterChannel(null); + assertNull(unregisteredChannel); + } + + @Test + public void testNoErrorChannelByDefault() { + DefaultChannelRegistry registry = new DefaultChannelRegistry(); + assertNull(registry.getErrorChannel()); + } + + @Test + public void testRegisteredErrorChannel() { + SimpleChannel errorChannel = new SimpleChannel(); + DefaultChannelRegistry registry = new DefaultChannelRegistry(); + registry.setErrorChannel(errorChannel); + MessageChannel result = registry.getErrorChannel(); + assertNotNull(result); + assertSame(errorChannel, result); + } + +}