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

Added AbstractSubscribablChannel base class for DirectChannel and PublishSubscribeChannel.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-19 12:33:22 +00:00
parent e0318d6132
commit 045f96e2c0
3 changed files with 69 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java Normal file
View File

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

21
org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java
View File

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

29
org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -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<BroadcastingDispatcher> {
/**
* 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);
}
}