From 045f96e2c05d7c91853194cbfd1bb4357eb2eff6 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 19 Sep 2008 12:33:22 +0000 Subject: [PATCH] Added AbstractSubscribablChannel base class for DirectChannel and PublishSubscribeChannel. --- .../channel/AbstractSubscribableChannel.java | 59 +++++++++++++++++++ .../integration/channel/DirectChannel.java | 21 +------ .../channel/PublishSubscribeChannel.java | 29 +++------ 3 files changed, 69 insertions(+), 40 deletions(-) create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java new file mode 100644 index 0000000000..70a905c527 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java @@ -0,0 +1,59 @@ +/* + * 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.dispatcher.MessageDispatcher; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageConsumer; +import org.springframework.integration.message.Subscribable; +import org.springframework.util.Assert; + +/** + * Base implementation of {@link MessageChannel} that invokes the subscribed + * {@link MessageConsumer consumer(s)} by delegating to a {@link MessageDispatcher}. + * + * @author Mark Fisher + */ +public class AbstractSubscribableChannel extends AbstractMessageChannel implements Subscribable { + + private final T dispatcher; + + + public AbstractSubscribableChannel(T dispatcher) { + Assert.notNull(dispatcher, "dispatcher must not be null"); + this.dispatcher = dispatcher; + } + + + protected T getDispatcher() { + return this.dispatcher; + } + + public boolean subscribe(MessageConsumer consumer) { + return this.dispatcher.subscribe(consumer); + } + + public boolean unsubscribe(MessageConsumer consumer) { + return this.dispatcher.unsubscribe(consumer); + } + + @Override + protected boolean doSend(Message message, long timeout) { + return this.dispatcher.dispatch(message); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java index 505ec2690d..4ba2c3c9ea 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java @@ -18,9 +18,6 @@ package org.springframework.integration.channel; import org.springframework.integration.dispatcher.SimpleDispatcher; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageConsumer; -import org.springframework.integration.message.Subscribable; /** * A channel that invokes the subscribed {@link MessageEndpoint endpoint(s)} @@ -29,22 +26,10 @@ import org.springframework.integration.message.Subscribable; * @author Dave Syer * @author Mark Fisher */ -public class DirectChannel extends AbstractMessageChannel implements Subscribable { +public class DirectChannel extends AbstractSubscribableChannel { - private final SimpleDispatcher dispatcher = new SimpleDispatcher(); - - - public boolean subscribe(MessageConsumer consumer) { - return this.dispatcher.subscribe(consumer); - } - - public boolean unsubscribe(MessageConsumer consumer) { - return this.dispatcher.unsubscribe(consumer); - } - - @Override - protected boolean doSend(Message message, long timeout) { - return this.dispatcher.dispatch(message); + public DirectChannel() { + super(new SimpleDispatcher()); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 95cafb14fd..a52c3fdb5f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -18,47 +18,32 @@ package org.springframework.integration.channel; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.dispatcher.BroadcastingDispatcher; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageConsumer; -import org.springframework.integration.message.Subscribable; /** + * A channel that sends Messages to each of its subscribers. + * * @author Mark Fisher */ -public class PublishSubscribeChannel extends AbstractMessageChannel implements Subscribable { - - private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher(); - +public class PublishSubscribeChannel extends AbstractSubscribableChannel { /** * Create a PublishSubscribeChannel that will use a {@link TaskExecutor} * to publish its Messages. */ public PublishSubscribeChannel(TaskExecutor taskExecutor) { + super(new BroadcastingDispatcher()); if (taskExecutor != null) { - this.dispatcher.setTaskExecutor(taskExecutor); + this.getDispatcher().setTaskExecutor(taskExecutor); } } public PublishSubscribeChannel() { + this(null); } public void setApplySequence(boolean applySequence) { - this.dispatcher.setApplySequence(applySequence); - } - - public boolean subscribe(MessageConsumer consumer) { - return this.dispatcher.subscribe(consumer); - } - - public boolean unsubscribe(MessageConsumer consumer) { - return this.dispatcher.unsubscribe(consumer); - } - - @Override - protected boolean doSend(Message message, long timeout) { - return this.dispatcher.dispatch(message); + this.getDispatcher().setApplySequence(applySequence); } }