INT-3521 completions for ChannelInterceptorList

JIRA: https://jira.spring.io/browse/INT-3521

INT-3521: Rework invocation `index` to the `Deque` of invoked interceptors

Conflicts:
	src/reference/docbook/whats-new.xml

INT-3521: Address PR comments

Create an `interceptorStack` only if there are `interceptor` on the channel.
Invoke `afterSend(Receive)Completion` only `if (interceptorStack != null)`

Minor Doc Polishing

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java

@@ -16,6 +16,9 @@
 
 package org.springframework.integration.amqp.channel;
 
+import java.util.ArrayDeque;
+import java.util.Deque;
+
 import org.springframework.amqp.core.AmqpAdmin;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.Queue;
@@ -23,6 +26,7 @@ import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.util.Assert;
 
 /**
@@ -31,6 +35,7 @@ import org.springframework.util.Assert;
  * name as the routing key.
  *
  * @author Mark Fisher
+ * @author Artem Bilan
  * @since 2.1
  */
 public class PollableAmqpChannel extends AbstractAmqpChannel implements PollableChannel {
@@ -93,21 +98,39 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
 
 	@Override
 	public Message<?> receive() {
-		if (!this.getInterceptors().preReceive(this)) {
- 			return null;
- 		}
-		Object object = this.getAmqpTemplate().receiveAndConvert(this.queueName);
-		if (object == null) {
-			return null;
+		ChannelInterceptorList interceptorList = getInterceptors();
+		Deque<ChannelInterceptor> interceptorStack = null;
+		try {
+			if (interceptorList.getInterceptors().size() > 0) {
+				interceptorStack = new ArrayDeque<ChannelInterceptor>();
+
+				if (!interceptorList.preReceive(this, interceptorStack)) {
+					 return null;
+				}
+			}
+			Object object = getAmqpTemplate().receiveAndConvert(this.queueName);
+			if (object == null) {
+				return null;
+			}
+			Message<?> message = null;
+			if (object instanceof Message<?>) {
+				message = (Message<?>) object;
+			}
+			else {
+				message = getMessageBuilderFactory().withPayload(object).build();
+			}
+			message = interceptorList.postReceive(message, this);
+			if (interceptorStack != null) {
+				interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
+			}
+			return message;
 		}
-		Message<?> replyMessage = null;
-		if (object instanceof Message<?>) {
-			replyMessage = (Message<?>) object;
+		catch (RuntimeException e) {
+			if (interceptorStack != null) {
+				interceptorList.afterReceiveCompletion(null, this, e, interceptorStack);
+			}
+			throw e;
 		}
-		else {
-			replyMessage = this.getMessageBuilderFactory().withPayload(object).build();
-		}
-		return this.getInterceptors().postReceive(replyMessage, this) ;
 	}
 
 	@Override

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java

@@ -16,8 +16,11 @@
 
 package org.springframework.integration.channel;
 
+import java.util.ArrayDeque;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.Deque;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 
@@ -257,19 +260,31 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		if (this.shouldTrack) {
 			message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
 		}
+
+		Deque<ChannelInterceptor> interceptorStack = null;
+		boolean sent = false;
 		try {
 			if (this.datatypes.length > 0) {
 				message = this.convertPayloadIfNecessary(message);
 			}
-			message = this.interceptors.preSend(message, this);
-			if (message == null) {
-				return false;
+			if (this.interceptors.getInterceptors().size() > 0) {
+				interceptorStack = new ArrayDeque<ChannelInterceptor>();
+				message = this.interceptors.preSend(message, this, interceptorStack);
+				if (message == null) {
+					return false;
+				}
 			}
-			boolean sent = this.doSend(message, timeout);
+			sent = this.doSend(message, timeout);
 			this.interceptors.postSend(message, this, sent);
+			if (interceptorStack != null) {
+				this.interceptors.afterSendCompletion(message, this, sent, null, interceptorStack);
+			}
 			return sent;
 		}
 		catch (Exception e) {
+			if (interceptorStack != null) {
+				this.interceptors.afterSendCompletion(message, this, sent, e, interceptorStack);
+			}
 			if (e instanceof MessagingException) {
 				throw (MessagingException) e;
 			}
@@ -326,7 +341,6 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 		private final List<ChannelInterceptor> interceptors = new CopyOnWriteArrayList<ChannelInterceptor>();
 
-
 		public boolean set(List<ChannelInterceptor> interceptors) {
 			synchronized (this.interceptors) {
 				this.interceptors.clear();
@@ -342,7 +356,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 			this.interceptors.add(index, interceptor);
 		}
 
-		public Message<?> preSend(Message<?> message, MessageChannel channel) {
+		public Message<?> preSend(Message<?> message, MessageChannel channel,
+				Deque<ChannelInterceptor> interceptorStack) {
 			if (logger.isDebugEnabled()) {
 				logger.debug("preSend on channel '" + channel + "', message: " + message);
 			}
@@ -350,8 +365,14 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 				for (ChannelInterceptor interceptor : this.interceptors) {
 					message = interceptor.preSend(message, channel);
 					if (message == null) {
+						if (logger.isDebugEnabled()) {
+							logger.debug(interceptor.getClass().getSimpleName()
+									+ " returned null from preSend, i.e. precluding the send.");
+						}
+						afterSendCompletion(null, channel, false, null, interceptorStack);
 						return null;
 					}
+					interceptorStack.add(interceptor);
 				}
 			}
 			return message;
@@ -368,15 +389,30 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 			}
 		}
 
-		public boolean preReceive(MessageChannel channel) {
+		public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex,
+				Deque<ChannelInterceptor> interceptorStack) {
+			for (Iterator<ChannelInterceptor> iterator = interceptorStack.descendingIterator(); iterator.hasNext(); ) {
+				ChannelInterceptor interceptor = iterator.next();
+				try {
+					interceptor.afterSendCompletion(message, channel, sent, ex);
+				}
+				catch (Throwable ex2) {
+					logger.error("Exception from afterSendCompletion in " + interceptor, ex2);
+				}
+			}
+		}
+
+		public boolean preReceive(MessageChannel channel, Deque<ChannelInterceptor> interceptorStack) {
 			if (logger.isTraceEnabled()) {
 				logger.trace("preReceive on channel '" + channel + "'");
 			}
 			if (this.interceptors.size() > 0) {
 				for (ChannelInterceptor interceptor : interceptors) {
 					if (!interceptor.preReceive(channel)) {
+						afterReceiveCompletion(null, channel, null, interceptorStack);
 						return false;
 					}
+					interceptorStack.add(interceptor);
 				}
 			}
 			return true;
@@ -400,6 +436,19 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 			return message;
 		}
 
+		public void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex,
+				Deque<ChannelInterceptor> interceptorStack) {
+			for (Iterator<ChannelInterceptor> iterator = interceptorStack.descendingIterator(); iterator.hasNext(); ) {
+				ChannelInterceptor interceptor = iterator.next();
+				try {
+					interceptor.afterReceiveCompletion(message, channel, ex);
+				}
+				catch (Throwable ex2) {
+					logger.error("Exception from afterReceiveCompletion in " + interceptor, ex2);
+				}
+			}
+		}
+
 		public List<ChannelInterceptor> getInterceptors() {
 			return Collections.unmodifiableList(this.interceptors);
 		}

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java

@@ -16,13 +16,18 @@
 
 package org.springframework.integration.channel;
 
+import java.util.ArrayDeque;
+import java.util.Deque;
+
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
 
 /**
  * Base class for all pollable channels.
  *
  * @author Mark Fisher
+ * @author Artem Bilan
  */
 public abstract class AbstractPollableChannel extends AbstractMessageChannel implements PollableChannel {
 
@@ -53,12 +58,29 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 	 */
 	@Override
 	public final Message<?> receive(long timeout) {
-		if (!this.getInterceptors().preReceive(this)) {
-			return null;
+		ChannelInterceptorList interceptorList = this.getInterceptors();
+		Deque<ChannelInterceptor> interceptorStack = null;
+		try {
+			if (interceptorList.getInterceptors().size() > 0) {
+				interceptorStack = new ArrayDeque<ChannelInterceptor>();
+
+				if (!interceptorList.preReceive(this, interceptorStack)) {
+					return null;
+				}
+			}
+			Message<?> message = this.doReceive(timeout);
+			message = interceptorList.postReceive(message, this);
+			if (interceptorStack != null) {
+				interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
+			}
+			return message;
+		}
+		catch (RuntimeException e) {
+			if (interceptorStack != null) {
+				interceptorList.afterReceiveCompletion(null, this, e, interceptorStack);
+			}
+			throw e;
 		}
-		Message<?> message = this.doReceive(timeout);
-		message = this.getInterceptors().postReceive(message, this);
-		return message;
 	}
 
 	/**

spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.channel.interceptor;
 
 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;
@@ -55,12 +56,14 @@ public class ChannelInterceptorTests {
 
 	@Test
 	public void testPreSendInterceptorReturnsMessage() {
-		channel.addInterceptor(new PreSendReturnsMessageInterceptor());
+		PreSendReturnsMessageInterceptor interceptor = new PreSendReturnsMessageInterceptor();
+		channel.addInterceptor(interceptor);
 		channel.send(new GenericMessage<String>("test"));
 		Message<?> result = channel.receive(0);
 		assertNotNull(result);
 		assertEquals("test", result.getPayload());
 		assertEquals(1, result.getHeaders().get(PreSendReturnsMessageInterceptor.class.getSimpleName()));
+		assertTrue(interceptor.wasAfterCompletionInvoked());
 	}
 
 	@Test
@@ -130,14 +133,56 @@ public class ChannelInterceptorTests {
 		assertEquals(1, sentCounter.get());
 	}
 
+	@Test
+	public void afterCompletionWithSendException() {
+		final AbstractMessageChannel testChannel = new AbstractMessageChannel() {
+
+			@Override
+			protected boolean doSend(Message<?> message, long timeout) {
+				throw new RuntimeException("Simulated exception");
+			}
+		};
+		AfterCompletionTestInterceptor interceptor1 = new AfterCompletionTestInterceptor();
+		AfterCompletionTestInterceptor interceptor2 = new AfterCompletionTestInterceptor();
+		testChannel.addInterceptor(interceptor1);
+		testChannel.addInterceptor(interceptor2);
+		try {
+			testChannel.send(MessageBuilder.withPayload("test").build());
+		}
+		catch (Exception ex) {
+			assertEquals("Simulated exception", ex.getCause().getMessage());
+		}
+		assertTrue(interceptor1.wasAfterCompletionInvoked());
+		assertTrue(interceptor2.wasAfterCompletionInvoked());
+	}
+
+	@Test
+	public void afterCompletionWithPreSendException() {
+		AfterCompletionTestInterceptor interceptor1 = new AfterCompletionTestInterceptor();
+		AfterCompletionTestInterceptor interceptor2 = new AfterCompletionTestInterceptor();
+		interceptor2.setExceptionToRaise(new RuntimeException("Simulated exception"));
+		this.channel.addInterceptor(interceptor1);
+		this.channel.addInterceptor(interceptor2);
+		try {
+			this.channel.send(MessageBuilder.withPayload("test").build());
+		}
+		catch (Exception ex) {
+			assertEquals("Simulated exception", ex.getCause().getMessage());
+		}
+		assertTrue(interceptor1.wasAfterCompletionInvoked());
+		assertFalse(interceptor2.wasAfterCompletionInvoked());
+	}
+
 	@Test
 	public void testPreReceiveInterceptorReturnsTrue() {
-		channel.addInterceptor(new PreReceiveReturnsTrueInterceptor());
+		PreReceiveReturnsTrueInterceptor interceptor = new PreReceiveReturnsTrueInterceptor();
+		channel.addInterceptor(interceptor);
 		Message<?> message = new GenericMessage<String>("test");
 		channel.send(message);
 		Message<?> result = channel.receive(0);
-		assertEquals(1, PreReceiveReturnsTrueInterceptor.counter.get());
+		assertEquals(1, interceptor.getCounter().get());
 		assertNotNull(result);
+		assertTrue(interceptor.wasAfterCompletionInvoked());
 	}
 
 	@Test
@@ -175,6 +220,25 @@ public class ChannelInterceptorTests {
 		assertEquals(2, invokedCount.get());
 		assertEquals(1, messageCount.get());
 	}
+
+	@Test
+	public void afterCompletionWithReceiveException() {
+		PreReceiveReturnsTrueInterceptor interceptor1 = new PreReceiveReturnsTrueInterceptor();
+		PreReceiveReturnsTrueInterceptor interceptor2 = new PreReceiveReturnsTrueInterceptor();
+		interceptor2.setExceptionToRaise(new RuntimeException("Simulated exception"));
+		channel.addInterceptor(interceptor1);
+		channel.addInterceptor(interceptor2);
+
+		try {
+			channel.receive(0);
+		}
+		catch (Exception ex) {
+			assertEquals("Simulated exception", ex.getMessage());
+		}
+		assertTrue(interceptor1.wasAfterCompletionInvoked());
+		assertFalse(interceptor2.wasAfterCompletionInvoked());
+	}
+
 	@Test
 	public void testInterceptorBeanWithPNamespace(){
 		ConfigurableApplicationContext ac =
@@ -195,12 +259,14 @@ public class ChannelInterceptorTests {
 
 		private static AtomicInteger counter = new AtomicInteger();
 
+		private volatile boolean afterCompletionInvoked;
+
 		@Override
 		public Message<?> preSend(Message<?> message, MessageChannel channel) {
 			assertNotNull(message);
-			Message<?> reply = MessageBuilder.fromMessage(message)
-					.setHeader(this.getClass().getSimpleName(), counter.incrementAndGet()).build();
-			return reply;
+			return MessageBuilder.fromMessage(message)
+					.setHeader(this.getClass().getSimpleName(), counter.incrementAndGet())
+					.build();
 		}
 		public String getFoo() {
 			return foo;
@@ -209,6 +275,16 @@ public class ChannelInterceptorTests {
 		public void setFoo(String foo) {
 			this.foo = foo;
 		}
+
+		public boolean wasAfterCompletionInvoked() {
+			return this.afterCompletionInvoked;
+		}
+
+		@Override
+		public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
+			this.afterCompletionInvoked = true;
+		}
+
 	}
 
 
@@ -228,16 +304,76 @@ public class ChannelInterceptorTests {
 		}
 	}
 
+	private static class AfterCompletionTestInterceptor extends ChannelInterceptorAdapter {
+
+		private AtomicInteger counter = new AtomicInteger();
+
+		private volatile boolean afterCompletionInvoked;
+
+		private RuntimeException exceptionToRaise;
+
+		public void setExceptionToRaise(RuntimeException exception) {
+			this.exceptionToRaise = exception;
+		}
+
+		public AtomicInteger getCounter() {
+			return this.counter;
+		}
+
+		public boolean wasAfterCompletionInvoked() {
+			return this.afterCompletionInvoked;
+		}
+
+		@Override
+		public Message<?> preSend(Message<?> message, MessageChannel channel) {
+			assertNotNull(message);
+			counter.incrementAndGet();
+			if (this.exceptionToRaise != null) {
+				throw this.exceptionToRaise;
+			}
+			return message;
+		}
+
+		@Override
+		public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
+			this.afterCompletionInvoked = true;
+		}
+
+	}
 
 	private static class PreReceiveReturnsTrueInterceptor extends ChannelInterceptorAdapter {
 
-		private static AtomicInteger counter = new AtomicInteger();
+		private AtomicInteger counter = new AtomicInteger();
+
+		private volatile boolean afterCompletionInvoked;
+
+		private RuntimeException exceptionToRaise;
+
+		public void setExceptionToRaise(RuntimeException exception) {
+			this.exceptionToRaise = exception;
+		}
+
+		public AtomicInteger getCounter() {
+			return this.counter;
+		}
 
 		@Override
 		public boolean preReceive(MessageChannel channel) {
 			counter.incrementAndGet();
+			if (this.exceptionToRaise != null) {
+				throw this.exceptionToRaise;
+			}
 			return true;
 		}
+
+		public boolean wasAfterCompletionInvoked() {
+			return this.afterCompletionInvoked;
+		}
+
+		@Override
+		public void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex) {
+			this.afterCompletionInvoked = true;
+		}
 	}
 
 

spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -16,14 +16,19 @@
 
 package org.springframework.integration.jms;
 
+import java.util.ArrayDeque;
+import java.util.Deque;
+
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
 
 /**
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.0
  */
 public class PollableJmsChannel extends AbstractJmsChannel implements PollableChannel {
@@ -39,28 +44,46 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
 	}
 
 	public Message<?> receive() {
-		if (!this.getInterceptors().preReceive(this)) {
- 			return null;
- 		}
-		Object object;
-		if (this.messageSelector == null) {
-			object = this.getJmsTemplate().receiveAndConvert();
-		}
-		else {
-			object = this.getJmsTemplate().receiveSelectedAndConvert(this.messageSelector);
-		}
+		ChannelInterceptorList interceptorList = getInterceptors();
+		Deque<ChannelInterceptor> interceptorStack = null;
+		try {
+			if (interceptorList.getInterceptors().size() > 0) {
+				interceptorStack = new ArrayDeque<ChannelInterceptor>();
 
-		if (object == null) {
-			return null;
+				if (!interceptorList.preReceive(this, interceptorStack)) {
+					return null;
+				}
+			}
+			Object object;
+			if (this.messageSelector == null) {
+				object = getJmsTemplate().receiveAndConvert();
+			}
+			else {
+				object = getJmsTemplate().receiveSelectedAndConvert(this.messageSelector);
+			}
+
+			if (object == null) {
+				return null;
+			}
+			Message<?> message = null;
+			if (object instanceof Message<?>) {
+				message = (Message<?>) object;
+			}
+			else {
+				message = getMessageBuilderFactory().withPayload(object).build();
+			}
+			message = interceptorList.postReceive(message, this);
+			if (interceptorStack != null) {
+				interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
+			}
+			return message;
 		}
-		Message<?> replyMessage = null;
-		if (object instanceof Message<?>) {
-			replyMessage = (Message<?>) object;
+		catch (RuntimeException e) {
+			if (interceptorStack != null) {
+				interceptorList.afterReceiveCompletion(null, this, e, interceptorStack);
+			}
+			throw e;
 		}
-		else {
-			replyMessage = this.getMessageBuilderFactory().withPayload(object).build();
-		}
-		return this.getInterceptors().postReceive(replyMessage, this) ;
 	}
 
 	public Message<?> receive(long timeout) {

src/reference/docbook/channel.xml

@@ -306,9 +306,13 @@
 
     void postSend(Message<?> message, MessageChannel channel, boolean sent);
 
+    void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex);
+
     boolean preReceive(MessageChannel channel);
 
     Message<?> postReceive(Message<?> message, MessageChannel channel);
+
+    void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex);
 }]]></programlisting>
       After implementing the interface, registering the interceptor with a channel is just a matter of calling:
       <programlisting language="java">channel.addInterceptor(someChannelInterceptor);</programlisting>
@@ -363,6 +367,14 @@
         precede postReceive.
       </tip>
     </para>
+    <para>
+        Starting with <emphasis>Spring Framework 4.1</emphasis> and Spring Integration 4.1, the
+        <interfacename>ChannelInterceptor</interfacename> provides new methods - <code>afterSendCompletion()</code>
+        and <code>afterReceiveCompletion()</code>. They are invoked after <code>send()/receive()</code> calls,
+        regardless of any exception that is raised, thus allowing for resource cleanup. Note, the
+        Channel invokes these methods on the ChannelInterceptor List in the reverse order of the
+        initial <code>preSend()/preReceive()</code> calls.
+    </para>
   </section>
 
   <section id="channel-template">

src/reference/docbook/whats-new.xml

@@ -272,5 +272,14 @@
 				See <xref linkend="channel-configuration-queuechannel"/>.
 			</para>
 		</section>
+		<section id="4.1-channel-interceptor">
+			<title>ChannelInterceptor Changes</title>
+			<para>
+				The <interfacename>ChannelInterceptor</interfacename> now supports additional
+				<code>afterSendCompletion()</code> and <code>afterReceiveCompletion()</code> methods.
+				See <xref linkend="channel-interceptors"/>.
+			</para>
+		</section>
+
 	</section>
 </chapter>