From 134aa5923e58df69867532ab3965e15e183e5a62 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 26 Sep 2014 11:15:02 +0300 Subject: [PATCH] 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 --- .../amqp/channel/PollableAmqpChannel.java | 49 ++++-- .../channel/AbstractMessageChannel.java | 63 +++++++- .../channel/AbstractPollableChannel.java | 32 +++- .../interceptor/ChannelInterceptorTests.java | 150 +++++++++++++++++- .../integration/jms/PollableJmsChannel.java | 63 +++++--- src/reference/docbook/channel.xml | 12 ++ src/reference/docbook/whats-new.xml | 9 ++ 7 files changed, 326 insertions(+), 52 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 8bfca97c37..75ee1be4d5 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/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 interceptorStack = null; + try { + if (interceptorList.getInterceptors().size() > 0) { + interceptorStack = new ArrayDeque(); + + 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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 600763ab95..0f8a69fc48 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/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 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(); + 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 interceptors = new CopyOnWriteArrayList(); - public boolean set(List 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 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 interceptorStack) { + for (Iterator 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 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 interceptorStack) { + for (Iterator 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 getInterceptors() { return Collections.unmodifiableList(this.interceptors); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java index ce86b755dc..ae82e61cf7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java +++ b/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 interceptorStack = null; + try { + if (interceptorList.getInterceptors().size() > 0) { + interceptorStack = new ArrayDeque(); + + 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; } /** diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java index 765db5721c..62de0377df 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java +++ b/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("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("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; + } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java index 7fd3a82f35..dcb3be7063 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java +++ b/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 interceptorStack = null; + try { + if (interceptorList.getInterceptors().size() > 0) { + interceptorStack = new ArrayDeque(); - 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) { diff --git a/src/reference/docbook/channel.xml b/src/reference/docbook/channel.xml index 082ec98df2..ea1db67b87 100644 --- a/src/reference/docbook/channel.xml +++ b/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); }]]> After implementing the interface, registering the interceptor with a channel is just a matter of calling: channel.addInterceptor(someChannelInterceptor); @@ -363,6 +367,14 @@ precede postReceive. + + Starting with Spring Framework 4.1 and Spring Integration 4.1, the + ChannelInterceptor provides new methods - afterSendCompletion() + and afterReceiveCompletion(). They are invoked after send()/receive() 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 preSend()/preReceive() calls. +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 0741657b9e..4ecf70eade 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -272,5 +272,14 @@ See .
+
+ ChannelInterceptor Changes + + The ChannelInterceptor now supports additional + afterSendCompletion() and afterReceiveCompletion() methods. + See . + +
+