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

Added basic ChannelInterceptor functionality and 'unregister' method for ChannelRegistry.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-31 16:11:02 +00:00
parent 7d8566ee2e
commit 472aa0f436
9 changed files with 642 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

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

237
spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java Normal file
View File

@@ -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<ChannelInterceptor> 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 <code>true</code> if the message is sent successfully or
* <code>false</code> 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 <code>true</code> if the message is sent successfully,
* <code>false</code> 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 <code>null</code> 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 <code>null</code> 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<ChannelInterceptor> interceptors = new CopyOnWriteArrayList<ChannelInterceptor>();
public boolean set(List<ChannelInterceptor> 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);
}
}
}
}

38
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java Normal file
View File

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

43
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAdapter.java Normal file
View File

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

2
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelRegistry.java
View File

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

4
spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultChannelRegistry.java
View File

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

116
spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java
View File

@@ -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<Message<?>> 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<Message<?>>(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 <code>true</code> if the message is sent successfully or
* <code>false</code> 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 <code>true</code> if the message is sent successfully,
* <code>false</code> 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 <code>null</code> 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 <code>null</code> 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;
}
}
}