Add ChannelInterceptor to spring-messaging module

Issue: SPR-10866

spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessageHeaderAccessor.java

@@ -68,7 +68,6 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
 	 */
 	protected SimpMessageHeaderAccessor(Message<?> message) {
 		super(message);
-		Assert.notNull(message, "message is required");
 	}
 
 

spring-messaging/src/main/java/org/springframework/messaging/support/NativeMessageHeaderAccessor.java

@@ -107,7 +107,10 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor {
 		return result;
 	}
 
-	protected List<String> getNativeHeader(String headerName) {
+	/**
+	 * Return all values for the specified native header or {@code null}.
+	 */
+	public List<String> getNativeHeader(String headerName) {
 		if (this.nativeHeaders.containsKey(headerName)) {
 			return this.nativeHeaders.get(headerName);
 		}
@@ -117,23 +120,28 @@ public class NativeMessageHeaderAccessor extends MessageHeaderAccessor {
 		return null;
 	}
 
+	/**
+	 * Return the first value for the specified native header of {@code null}.
+	 */
 	public String getFirstNativeHeader(String headerName) {
 		List<String> values = getNativeHeader(headerName);
 		return CollectionUtils.isEmpty(values) ? null : values.get(0);
 	}
 
 	/**
-	 * Set the value for the given header name. If the provided value is {@code null} the
-	 * header will be removed.
+	 * Set the specified native header value.
 	 */
-	protected void putNativeHeader(String name, List<String> value) {
+	public void setNativeHeader(String name, String value) {
 		if (!ObjectUtils.nullSafeEquals(value, getHeader(name))) {
-			this.nativeHeaders.put(name, value);
+			this.nativeHeaders.set(name, value);
 		}
 	}
 
-	public void setNativeHeader(String name, String value) {
-		this.nativeHeaders.set(name, value);
+	/**
+	 * Add the specified native header value.
+	 */
+	public void addNativeHeader(String name, String value) {
+		this.nativeHeaders.add(name, value);
 	}
 
 }

spring-messaging/src/main/java/org/springframework/messaging/support/channel/AbstractMessageChannel.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2002-2013 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.messaging.support.channel;
+
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageDeliveryException;
+import org.springframework.messaging.MessagingException;
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+
+/**
+ * Abstract base class for {@link MessageChannel} implementations.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public abstract class AbstractMessageChannel implements MessageChannel, BeanNameAware  {
+
+	protected Log logger = LogFactory.getLog(getClass());
+
+	private String beanName;
+
+	private final ChannelInterceptorChain interceptorChain = new ChannelInterceptorChain();
+
+
+	public AbstractMessageChannel() {
+		this.beanName = getClass().getSimpleName() + "@" + ObjectUtils.getIdentityHexString(this);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 * <p>Used primarily for logging purposes.
+	 */
+	@Override
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
+
+	/**
+	 * @return the name for this channel.
+	 */
+	public String getBeanName() {
+		return this.beanName;
+	}
+
+	/**
+	 * Set the list of channel interceptors. This will clear any existing interceptors.
+	 */
+	public void setInterceptors(List<ChannelInterceptor> interceptors) {
+		this.interceptorChain.set(interceptors);
+	}
+
+	/**
+	 * Add a channel interceptor to the end of the list.
+	 */
+	public void addInterceptor(ChannelInterceptor interceptor) {
+		this.interceptorChain.add(interceptor);
+	}
+
+	/**
+	 * Return a read-only list of the configured interceptors.
+	 */
+	public List<ChannelInterceptor> getInterceptors() {
+		return this.interceptorChain.getInterceptors();
+	}
+
+	/**
+	 * Exposes the interceptor list for subclasses.
+	 */
+	protected ChannelInterceptorChain getInterceptorChain() {
+		return this.interceptorChain;
+	}
+
+
+	@Override
+	public final boolean send(Message<?> message) {
+		return send(message, INDEFINITE_TIMEOUT);
+	}
+
+	@Override
+	public final boolean send(Message<?> message, long timeout) {
+
+		Assert.notNull(message, "Message must not be null");
+		if (logger.isTraceEnabled()) {
+			logger.trace("[" + this.beanName + "] sending message " + message);
+		}
+
+		message = this.interceptorChain.preSend(message, this);
+		if (message == null) {
+			return false;
+		}
+
+		try {
+			boolean sent = sendInternal(message, timeout);
+			this.interceptorChain.postSend(message, this, sent);
+			return sent;
+		}
+		catch (Exception e) {
+			if (e instanceof MessagingException) {
+				throw (MessagingException) e;
+			}
+			throw new MessageDeliveryException(message,
+					"Failed to send message to channel '" + this.getBeanName() + "'", e);
+		}
+	}
+
+	protected abstract boolean sendInternal(Message<?> message, long timeout);
+
+}

spring-messaging/src/main/java/org/springframework/messaging/support/channel/AbstractSubscribableChannel.java

@@ -16,14 +16,8 @@
 
 package org.springframework.messaging.support.channel;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.beans.factory.BeanNameAware;
-import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.SubscribableChannel;
-import org.springframework.util.Assert;
-import org.springframework.util.ObjectUtils;
 
 
 /**
@@ -32,57 +26,17 @@ import org.springframework.util.ObjectUtils;
  * @author Rossen Stoyanchev
  * @since 4.0
  */
-public abstract class AbstractSubscribableChannel implements SubscribableChannel, BeanNameAware  {
+public abstract class AbstractSubscribableChannel extends AbstractMessageChannel implements SubscribableChannel {
 
-	protected Log logger = LogFactory.getLog(getClass());
-
-	private String beanName;
-
-
-	public AbstractSubscribableChannel() {
-		this.beanName = getClass().getSimpleName() + "@" + ObjectUtils.getIdentityHexString(this);
-	}
-
-	/**
-	 * {@inheritDoc}
-	 * <p>Used primarily for logging purposes.
-	 */
-	@Override
-	public void setBeanName(String name) {
-		this.beanName = name;
-	}
-
-	/**
-	 * @return the name for this channel.
-	 */
-	public String getBeanName() {
-		return this.beanName;
-	}
-
-	@Override
-	public final boolean send(Message<?> message) {
-		return send(message, INDEFINITE_TIMEOUT);
-	}
-
-	@Override
-	public final boolean send(Message<?> message, long timeout) {
-		Assert.notNull(message, "Message must not be null");
-		if (logger.isTraceEnabled()) {
-			logger.trace("[" + this.beanName + "] sending message " + message);
-		}
-		return sendInternal(message, timeout);
-	}
-
-	protected abstract boolean sendInternal(Message<?> message, long timeout);
 
 	@Override
 	public final boolean subscribe(MessageHandler handler) {
 		if (hasSubscription(handler)) {
-			logger.warn("[" + this.beanName + "] handler already subscribed " + handler);
+			logger.warn("[" + getBeanName() + "] handler already subscribed " + handler);
 			return false;
 		}
 		if (logger.isDebugEnabled()) {
-			logger.debug("[" + this.beanName + "] subscribing " + handler);
+			logger.debug("[" + getBeanName() + "] subscribing " + handler);
 		}
 		return subscribeInternal(handler);
 	}
@@ -94,7 +48,7 @@ public abstract class AbstractSubscribableChannel implements SubscribableChannel
 	@Override
 	public final boolean unsubscribe(MessageHandler handler) {
 		if (logger.isDebugEnabled()) {
-			logger.debug("[" + this.beanName + "] unsubscribing " + handler);
+			logger.debug("[" + getBeanName() + "] unsubscribing " + handler);
 		}
 		return unsubscribeInternal(handler);
 	}

spring-messaging/src/main/java/org/springframework/messaging/support/channel/ChannelInterceptor.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2002-2010 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.messaging.support.channel;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * 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
+ * @since 4.0
+ */
+public interface ChannelInterceptor {
+
+	/**
+	 * Invoked before the Message is actually sent to the channel.
+	 * This allows for modification of the Message if necessary.
+	 * If this method returns <code>null</code>, then the actual
+	 * send invocation will not occur.
+	 */
+	Message<?> preSend(Message<?> message, MessageChannel channel);
+
+	/**
+	 * Invoked immediately after the send invocation. The boolean
+	 * value argument represents the return value of that invocation.
+	 */
+	void postSend(Message<?> message, MessageChannel channel, boolean sent);
+
+	/**
+	 * Invoked as soon as receive is called and before a Message is
+	 * actually retrieved. If the return value is 'false', then no
+	 * Message will be retrieved. This only applies to PollableChannels.
+	 */
+	boolean preReceive(MessageChannel channel);
+
+	/**
+	 * Invoked immediately after a Message has been retrieved but before
+	 * it is returned to the caller. The Message may be modified if
+	 * necessary. This only applies to PollableChannels.
+	 */
+	Message<?> postReceive(Message<?> message, MessageChannel channel);
+
+}

spring-messaging/src/main/java/org/springframework/messaging/support/channel/ChannelInterceptorAdapter.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2002-2013 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.messaging.support.channel;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * A {@link ChannelInterceptor} with empty method implementations as a convenience.
+ *
+ * @author Mark Fisher
+ * @since 4.0
+ */
+public class ChannelInterceptorAdapter implements ChannelInterceptor {
+
+
+	public Message<?> preSend(Message<?> message, MessageChannel channel) {
+		return message;
+	}
+
+	public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
+	}
+
+	public boolean preReceive(MessageChannel channel) {
+		return true;
+	}
+
+	public Message<?> postReceive(Message<?> message, MessageChannel channel) {
+		return message;
+	}
+
+}

spring-messaging/src/main/java/org/springframework/messaging/support/channel/ChannelInterceptorChain.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2002-2013 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.messaging.support.channel;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+
+/**
+ * A convenience wrapper class for invoking a list of {@link ChannelInterceptor}s.
+ *
+ * @author Mark Fisher
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+class ChannelInterceptorChain {
+
+	private static final Log logger = LogFactory.getLog(ChannelInterceptorChain.class);
+
+	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 List<ChannelInterceptor> getInterceptors() {
+		return Collections.unmodifiableList(this.interceptors);
+	}
+
+
+	public Message<?> preSend(Message<?> message, MessageChannel channel) {
+		if (logger.isTraceEnabled()) {
+			logger.trace("preSend on channel '" + channel + "', message: " + message);
+		}
+		for (ChannelInterceptor interceptor : this.interceptors) {
+			message = interceptor.preSend(message, channel);
+			if (message == null) {
+				return null;
+			}
+		}
+		return message;
+	}
+
+	public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
+		if (logger.isTraceEnabled()) {
+			logger.trace("postSend (sent=" + sent + ") on channel '" + channel + "', message: " + message);
+		}
+		for (ChannelInterceptor interceptor : this.interceptors) {
+			interceptor.postSend(message, channel, sent);
+		}
+	}
+
+	public boolean preReceive(MessageChannel channel) {
+		if (logger.isTraceEnabled()) {
+			logger.trace("preReceive on channel '" + channel + "'");
+		}
+		for (ChannelInterceptor interceptor : this.interceptors) {
+			if (!interceptor.preReceive(channel)) {
+				return false;
+			}
+		}
+		return true;
+	}
+
+	public Message<?> postReceive(Message<?> message, MessageChannel channel) {
+		if (message != null && logger.isTraceEnabled()) {
+			logger.trace("postReceive on channel '" + channel + "', message: " + message);
+		}
+		else if (logger.isTraceEnabled()) {
+			logger.trace("postReceive on channel '" + channel + "', message is null");
+		}
+		for (ChannelInterceptor interceptor : this.interceptors) {
+			message = interceptor.postReceive(message, channel);
+			if (message == null) {
+				return null;
+			}
+		}
+		return message;
+	}
+
+
+}