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

GH-2749: Deprecate ChannelInterceptorAware (#2751)

Browse Source 
* GH-2749: Deprecate ChannelInterceptorAware

Fixes https://github.com/spring-projects/spring-integration/issues/2749

The `org.springframework.messaging.support.InterceptableChannel` provides
exact functionality as `ChannelInterceptorAware`

* Make `ChannelInterceptorAware extends InterceptableChannel`
* Suppress deprecation warning whenever we need to keep backward
compatibility
* Fix all other places to deal with `InterceptableChannel` already

* GH-2749: Deprecate ChannelInterceptorAware

Fixes https://github.com/spring-projects/spring-integration/issues/2749

The `org.springframework.messaging.support.InterceptableChannel` provides
exact functionality as `ChannelInterceptorAware`

* Make `ChannelInterceptorAware extends InterceptableChannel`
* Suppress deprecation warning whenever we need to keep backward
compatibility
* Fix all other places to deal with `InterceptableChannel` already

* * Fix `channel.adoc` for the current version
This commit is contained in:
Artem Bilan
2019-02-22 10:22:58 -05:00
committed by Gary Russell
parent d6ac866871
commit 82ecd5a75d
20 changed files with 174 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -178,7 +178,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel
protected Message<?> doReceive(Long timeout) {
ChannelInterceptorList interceptorList = getInterceptors();
ChannelInterceptorList interceptorList = getIChannelInterceptorList();
Deque<ChannelInterceptor> interceptorStack = null;
boolean counted = false;
boolean countsEnabled = isCountsEnabled();

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

@@ -66,6 +66,7 @@ import org.springframework.util.StringUtils;
* @author Artem Bilan
*/
@IntegrationManagedResource
@SuppressWarnings("deprecation")
public abstract class AbstractMessageChannel extends IntegrationObjectSupport
implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics,
ConfigurableMetricsAware<AbstractMessageChannelMetrics> {
@@ -243,7 +244,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
* Return a read-only list of the configured interceptors.
*/
@Override
public List<ChannelInterceptor> getChannelInterceptors() {
public List<ChannelInterceptor> getInterceptors() {
return this.interceptors.getInterceptors();
}
@@ -259,10 +260,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
}
/**
* Exposes the interceptor list for subclasses.
* Exposes the interceptor list instance for subclasses.
* @return The channel interceptor list.
*/
protected ChannelInterceptorList getInterceptors() {
protected ChannelInterceptorList getIChannelInterceptorList() {
return this.interceptors;
}

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

@@ -92,7 +92,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel
@Override
@Nullable
public Message<?> receive(long timeout) {
ChannelInterceptorList interceptorList = getInterceptors();
ChannelInterceptorList interceptorList = getIChannelInterceptorList();
Deque<ChannelInterceptor> interceptorStack = null;
boolean counted = false;
boolean countsEnabled = isCountsEnabled();

49
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAware.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2018 the original author or authors.
* Copyright 2014-2019 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.
@@ -18,8 +18,8 @@ package org.springframework.integration.channel;
import java.util.List;
import org.springframework.lang.Nullable;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
/**
* A marker interface providing the ability to configure {@link ChannelInterceptor}s
@@ -30,48 +30,21 @@ import org.springframework.messaging.support.ChannelInterceptor;
* *
* @author Artem Bilan
* @author Gary Russell
*
* @since 4.0
*
* @deprecated since 5.2 in favor of {@link InterceptableChannel}.
* Will be removed in the next 5.3 version.
*/
public interface ChannelInterceptorAware {
/**
* Populate the {@link ChannelInterceptor}s to the target implementation.
* @param interceptors the {@link ChannelInterceptor}s to populate.
*/
void setInterceptors(List<ChannelInterceptor> interceptors);
/**
* And a {@link ChannelInterceptor} to the target implementation.
* @param interceptor the {@link ChannelInterceptor} to add.
*/
void addInterceptor(ChannelInterceptor interceptor);
/**
* And a {@link ChannelInterceptor} to the target implementation for the specific index.
* @param index the index for {@link ChannelInterceptor} to add.
* @param interceptor the {@link ChannelInterceptor} to add.
*/
void addInterceptor(int index, ChannelInterceptor interceptor);
@Deprecated
public interface ChannelInterceptorAware extends InterceptableChannel {
/**
* return the {@link ChannelInterceptor} list.
* @return the {@link ChannelInterceptor} list.
*/
List<ChannelInterceptor> getChannelInterceptors();
/**
* Remove the provided {@link ChannelInterceptor} from the target implementation.
* @param interceptor {@link ChannelInterceptor} to remove.
* @return the {@code boolean} if {@link ChannelInterceptor} has been removed.
*/
boolean removeInterceptor(ChannelInterceptor interceptor);
/**
* Remove a {@link ChannelInterceptor} from the target implementation for specific index.
* @param index the index for the {@link org.springframework.messaging.support.ChannelInterceptor} to remove.
* @return the {@code boolean} if the {@link ChannelInterceptor} has been removed.
*/
@Nullable
ChannelInterceptor removeInterceptor(int index);
default List<ChannelInterceptor> getChannelInterceptors() {
return getInterceptors();
}
}

5
spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannelInterceptorAware.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* Copyright 2015-2019 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.
@@ -23,9 +23,12 @@ package org.springframework.integration.channel;
* and the implementors require to know if they should make the
* {@link org.springframework.messaging.support.ExecutorChannelInterceptor}
* or not.
*
* @author Artem Bilan
*
* @since 4.2
*/
@SuppressWarnings("deprecation")
public interface ExecutorChannelInterceptorAware extends ChannelInterceptorAware {
boolean hasExecutorInterceptors();

15
spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/VetoCapableInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2018 the original author or authors.
* Copyright 2014-2019 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,16 +16,17 @@
package org.springframework.integration.channel.interceptor;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.messaging.support.InterceptableChannel;
/**
* {@link org.springframework.messaging.support.ChannelInterceptor}s implementing this
* interface can veto
* global interception of a particular channel. Could be used, for example,
* when an interceptor itself writes to an output channel (which should
* not be intercepted with this interceptor).
* interface can veto global interception of a particular channel.
* Could be used, for example, when an interceptor itself writes to an output channel
* (which should not be intercepted with this interceptor).
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.0
*
*/
@@ -36,6 +37,6 @@ public interface VetoCapableInterceptor {
* @param channel The channel that is about to be intercepted.
* @return false if the intercept wishes to veto the interception.
*/
boolean shouldIntercept(String beanName, ChannelInterceptorAware channel);
boolean shouldIntercept(String beanName, InterceptableChannel channel);
}

7
spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -23,7 +23,6 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.jmx.export.annotation.ManagedAttribute;
@@ -32,6 +31,7 @@ import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.util.Assert;
/**
@@ -173,7 +173,8 @@ public class WireTap implements ChannelInterceptor, Lifecycle, VetoCapableInterc
}
@Override
public boolean shouldIntercept(String beanName, ChannelInterceptorAware channel) {
public boolean shouldIntercept(String beanName, InterceptableChannel channel) {
return !getChannel().equals(channel);
}

81
spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -18,11 +18,8 @@ package org.springframework.integration.config;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.apache.commons.logging.Log;
@@ -35,11 +32,11 @@ import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.OrderComparator;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.interceptor.GlobalChannelInterceptorWrapper;
import org.springframework.integration.channel.interceptor.VetoCapableInterceptor;
import org.springframework.integration.support.utils.PatternMatchUtils;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
@@ -79,6 +76,7 @@ public final class GlobalChannelInterceptorProcessor
}
@Override
@SuppressWarnings("deprecation")
public void afterSingletonsInstantiated() {
Collection<GlobalChannelInterceptorWrapper> interceptors =
this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class).values();
@@ -86,20 +84,17 @@ public final class GlobalChannelInterceptorProcessor
logger.debug("No global channel interceptors.");
}
else {
for (GlobalChannelInterceptorWrapper channelInterceptor : interceptors) {
if (channelInterceptor.getOrder() >= 0) {
this.positiveOrderInterceptors.add(channelInterceptor);
interceptors.forEach(interceptor -> {
if (interceptor.getOrder() >= 0) {
this.positiveOrderInterceptors.add(interceptor);
}
else {
this.negativeOrderInterceptors.add(channelInterceptor);
this.negativeOrderInterceptors.add(interceptor);
}
}
});
Map<String, ChannelInterceptorAware> channels =
this.beanFactory.getBeansOfType(ChannelInterceptorAware.class);
for (Entry<String, ChannelInterceptorAware> entry : channels.entrySet()) {
addMatchingInterceptors(entry.getValue(), entry.getKey());
}
this.beanFactory.getBeansOfType(InterceptableChannel.class)
.forEach((key, value) -> addMatchingInterceptors(value, key));
}
this.singletonsInstantiated = true;
@@ -107,8 +102,8 @@ public final class GlobalChannelInterceptorProcessor
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (this.singletonsInstantiated && bean instanceof ChannelInterceptorAware) {
addMatchingInterceptors((ChannelInterceptorAware) bean, beanName);
if (this.singletonsInstantiated && bean instanceof InterceptableChannel) {
addMatchingInterceptors((InterceptableChannel) bean, beanName);
}
return bean;
}
@@ -118,41 +113,34 @@ public final class GlobalChannelInterceptorProcessor
* @param channel the message channel to add interceptors.
* @param beanName the message channel bean name to match the pattern.
*/
public void addMatchingInterceptors(ChannelInterceptorAware channel, String beanName) {
public void addMatchingInterceptors(InterceptableChannel channel, String beanName) {
if (logger.isDebugEnabled()) {
logger.debug("Applying global interceptors on channel '" + beanName + "'");
}
List<GlobalChannelInterceptorWrapper> tempInterceptors = new ArrayList<>();
for (GlobalChannelInterceptorWrapper globalChannelInterceptorWrapper : this.positiveOrderInterceptors) {
String[] patterns = globalChannelInterceptorWrapper.getPatterns();
patterns = StringUtils.trimArrayElements(patterns);
if (beanName != null && Boolean.TRUE.equals(PatternMatchUtils.smartMatch(beanName, patterns))) {
tempInterceptors.add(globalChannelInterceptorWrapper);
}
}
Collections.sort(tempInterceptors, this.comparator);
this.positiveOrderInterceptors
.forEach(interceptorWrapper ->
addMatchingInterceptors(beanName, tempInterceptors, interceptorWrapper));
for (GlobalChannelInterceptorWrapper next : tempInterceptors) {
ChannelInterceptor channelInterceptor = next.getChannelInterceptor();
if (!(channelInterceptor instanceof VetoCapableInterceptor)
|| ((VetoCapableInterceptor) channelInterceptor).shouldIntercept(beanName, channel)) {
channel.addInterceptor(channelInterceptor);
}
}
tempInterceptors.sort(this.comparator);
tempInterceptors
.stream()
.map(GlobalChannelInterceptorWrapper::getChannelInterceptor)
.filter(interceptor -> !(interceptor instanceof VetoCapableInterceptor)
|| ((VetoCapableInterceptor) interceptor).shouldIntercept(beanName, channel))
.forEach(channel::addInterceptor);
tempInterceptors.clear();
for (GlobalChannelInterceptorWrapper globalChannelInterceptorWrapper : this.negativeOrderInterceptors) {
String[] patterns = globalChannelInterceptorWrapper.getPatterns();
patterns = StringUtils.trimArrayElements(patterns);
if (beanName != null && Boolean.TRUE.equals(PatternMatchUtils.smartMatch(beanName, patterns))) {
tempInterceptors.add(globalChannelInterceptorWrapper);
}
}
this.negativeOrderInterceptors
.forEach(interceptorWrapper ->
addMatchingInterceptors(beanName, tempInterceptors, interceptorWrapper));
Collections.sort(tempInterceptors, this.comparator);
tempInterceptors.sort(this.comparator);
if (!tempInterceptors.isEmpty()) {
for (int i = tempInterceptors.size() - 1; i >= 0; i--) {
@@ -165,4 +153,15 @@ public final class GlobalChannelInterceptorProcessor
}
}
private static void addMatchingInterceptors(String beanName,
List<GlobalChannelInterceptorWrapper> tempInterceptors,
GlobalChannelInterceptorWrapper globalChannelInterceptorWrapper) {
String[] patterns = globalChannelInterceptorWrapper.getPatterns();
patterns = StringUtils.trimArrayElements(patterns);
if (beanName != null && Boolean.TRUE.equals(PatternMatchUtils.smartMatch(beanName, patterns))) {
tempInterceptors.add(globalChannelInterceptorWrapper);
}
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -35,7 +35,6 @@ import org.springframework.beans.factory.config.DestructionAwareBeanPostProcesso
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.FluxMessageChannel;
@@ -90,6 +89,7 @@ import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
@@ -430,12 +430,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
*/
public B wireTap(WireTapSpec wireTapSpec) {
WireTap interceptor = wireTapSpec.get();
if (!(this.currentMessageChannel instanceof ChannelInterceptorAware)) {
if (!(this.currentMessageChannel instanceof InterceptableChannel)) {
channel(new DirectChannel());
this.implicitChannel = true;
}
addComponent(wireTapSpec);
((ChannelInterceptorAware) this.currentMessageChannel).addInterceptor(interceptor);
((InterceptableChannel) this.currentMessageChannel).addInterceptor(interceptor);
return _this();
}

6
spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java
View File

@@ -20,7 +20,6 @@ import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
@@ -39,6 +38,7 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -150,8 +150,8 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
MessageChannel replyChannel = this.gatherChannel;
if (replyChannel instanceof ChannelInterceptorAware) {
((ChannelInterceptorAware) replyChannel)
if (replyChannel instanceof InterceptableChannel) {
((InterceptableChannel) replyChannel)
.addInterceptor(0,
new ChannelInterceptor() {

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

@@ -30,7 +30,6 @@ import org.junit.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.support.MessageBuilder;
@@ -41,6 +40,7 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.ExecutorChannelInterceptor;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.util.StringUtils;
/**
@@ -243,8 +243,8 @@ public class ChannelInterceptorTests {
public void testInterceptorBeanWithPNamespace() {
ConfigurableApplicationContext ac =
new ClassPathXmlApplicationContext("ChannelInterceptorTests-context.xml", ChannelInterceptorTests.class);
ChannelInterceptorAware channel = ac.getBean("input", AbstractMessageChannel.class);
List<ChannelInterceptor> interceptors = channel.getChannelInterceptors();
InterceptableChannel channel = ac.getBean("input", AbstractMessageChannel.class);
List<ChannelInterceptor> interceptors = channel.getInterceptors();
ChannelInterceptor channelInterceptor = interceptors.get(0);
assertThat(channelInterceptor).isInstanceOf(PreSendReturnsMessageInterceptor.class);
String foo = ((PreSendReturnsMessageInterceptor) channelInterceptor).getFoo();

16
spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/GlobalChannelInterceptorTests.java
View File

@@ -32,11 +32,11 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -57,20 +57,20 @@ public class GlobalChannelInterceptorTests {
@Autowired
@Qualifier("inputC")
ChannelInterceptorAware inputCChannel;
InterceptableChannel inputCChannel;
@Test
public void validateGlobalInterceptor() throws Exception {
Map<String, ChannelInterceptorAware> channels = applicationContext.getBeansOfType(ChannelInterceptorAware.class);
public void validateGlobalInterceptor() {
Map<String, InterceptableChannel> channels = applicationContext.getBeansOfType(InterceptableChannel.class);
for (String channelName : channels.keySet()) {
ChannelInterceptorAware channel = channels.get(channelName);
InterceptableChannel channel = channels.get(channelName);
if (channelName.equals("nullChannel")) {
continue;
}
ChannelInterceptor[] interceptors = channel.getChannelInterceptors()
.toArray(new ChannelInterceptor[channel.getChannelInterceptors().size()]);
ChannelInterceptor[] interceptors = channel.getInterceptors()
.toArray(new ChannelInterceptor[channel.getInterceptors().size()]);
if (channelName.equals("inputA")) { // 328741
assertThat(interceptors.length == 10).isTrue();
assertThat(interceptors[0].toString()).isEqualTo("interceptor-three");
@@ -133,7 +133,7 @@ public class GlobalChannelInterceptorTests {
@Test
public void testWildCardPatternMatch() {
List<ChannelInterceptor> channelInterceptors = this.inputCChannel.getChannelInterceptors();
List<ChannelInterceptor> channelInterceptors = this.inputCChannel.getInterceptors();
List<String> interceptorNames = new ArrayList<String>();
for (ChannelInterceptor interceptor : channelInterceptors) {
interceptorNames.add(interceptor.toString());

6
spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ImplicitConsumerChannelTests.java
View File

@@ -25,10 +25,10 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -99,7 +99,7 @@ public class ImplicitConsumerChannelTests {
}
@Override
public boolean shouldIntercept(String beanName, ChannelInterceptorAware channel) {
public boolean shouldIntercept(String beanName, InterceptableChannel channel) {
return !this.channel.equals(channel);
}
@@ -137,7 +137,7 @@ public class ImplicitConsumerChannelTests {
}
@Override
public boolean shouldIntercept(String beanName, ChannelInterceptorAware channel) {
public boolean shouldIntercept(String beanName, InterceptableChannel channel) {
return !this.channel.equals(channel);
}

30
spring-integration-core/src/test/java/org/springframework/integration/config/GlobalChannelInterceptorProcessorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2019 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.
@@ -29,9 +29,9 @@ import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.interceptor.GlobalChannelInterceptorWrapper;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
/**
* @author Meherzad Lahewala
@@ -59,24 +59,24 @@ public class GlobalChannelInterceptorProcessorTests {
verify(this.beanFactory)
.getBeansOfType(GlobalChannelInterceptorWrapper.class);
verify(this.beanFactory, Mockito.never())
.getBeansOfType(ChannelInterceptorAware.class);
.getBeansOfType(InterceptableChannel.class);
}
@Test
public void testProcessorWithInterceptorDefaultPattern() {
Map<String, GlobalChannelInterceptorWrapper> interceptors = new HashMap<>();
Map<String, ChannelInterceptorAware> channels = new HashMap<>();
Map<String, InterceptableChannel> channels = new HashMap<>();
ChannelInterceptor channelInterceptor = Mockito.mock(ChannelInterceptor.class);
GlobalChannelInterceptorWrapper globalChannelInterceptorWrapper =
new GlobalChannelInterceptorWrapper(channelInterceptor);
ChannelInterceptorAware channel = Mockito.mock(ChannelInterceptorAware.class);
InterceptableChannel channel = Mockito.mock(InterceptableChannel.class);
interceptors.put("Test-1", globalChannelInterceptorWrapper);
channels.put("Test-1", channel);
when(this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class))
.thenReturn(interceptors);
when(this.beanFactory.getBeansOfType(ChannelInterceptorAware.class))
when(this.beanFactory.getBeansOfType(InterceptableChannel.class))
.thenReturn(channels);
this.globalChannelInterceptorProcessor.afterSingletonsInstantiated();
@@ -88,19 +88,19 @@ public class GlobalChannelInterceptorProcessorTests {
@Test
public void testProcessorWithInterceptorMatchingPattern() {
Map<String, GlobalChannelInterceptorWrapper> interceptors = new HashMap<>();
Map<String, ChannelInterceptorAware> channels = new HashMap<>();
Map<String, InterceptableChannel> channels = new HashMap<>();
ChannelInterceptor channelInterceptor = Mockito.mock(ChannelInterceptor.class);
GlobalChannelInterceptorWrapper globalChannelInterceptorWrapper =
new GlobalChannelInterceptorWrapper(channelInterceptor);
ChannelInterceptorAware channel = Mockito.mock(ChannelInterceptorAware.class);
InterceptableChannel channel = Mockito.mock(InterceptableChannel.class);
globalChannelInterceptorWrapper.setPatterns(new String[] { "Te*" });
interceptors.put("Test-1", globalChannelInterceptorWrapper);
channels.put("Test-1", channel);
when(this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class))
.thenReturn(interceptors);
when(this.beanFactory.getBeansOfType(ChannelInterceptorAware.class))
when(this.beanFactory.getBeansOfType(InterceptableChannel.class))
.thenReturn(channels);
this.globalChannelInterceptorProcessor.afterSingletonsInstantiated();
@@ -111,19 +111,19 @@ public class GlobalChannelInterceptorProcessorTests {
@Test
public void testProcessorWithInterceptorNotMatchingPattern() {
Map<String, GlobalChannelInterceptorWrapper> interceptors = new HashMap<>();
Map<String, ChannelInterceptorAware> channels = new HashMap<>();
Map<String, InterceptableChannel> channels = new HashMap<>();
ChannelInterceptor channelInterceptor = Mockito.mock(ChannelInterceptor.class);
GlobalChannelInterceptorWrapper globalChannelInterceptorWrapper =
new GlobalChannelInterceptorWrapper(channelInterceptor);
ChannelInterceptorAware channel = Mockito.mock(ChannelInterceptorAware.class);
InterceptableChannel channel = Mockito.mock(InterceptableChannel.class);
globalChannelInterceptorWrapper.setPatterns(new String[] { "te*" });
interceptors.put("Test-1", globalChannelInterceptorWrapper);
channels.put("Test-1", channel);
when(this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class))
.thenReturn(interceptors);
when(this.beanFactory.getBeansOfType(ChannelInterceptorAware.class))
when(this.beanFactory.getBeansOfType(InterceptableChannel.class))
.thenReturn(channels);
this.globalChannelInterceptorProcessor.afterSingletonsInstantiated();
@@ -135,19 +135,19 @@ public class GlobalChannelInterceptorProcessorTests {
@Test
public void testProcessorWithInterceptorMatchingNegativePattern() {
Map<String, GlobalChannelInterceptorWrapper> interceptors = new HashMap<>();
Map<String, ChannelInterceptorAware> channels = new HashMap<>();
Map<String, InterceptableChannel> channels = new HashMap<>();
ChannelInterceptor channelInterceptor = Mockito.mock(ChannelInterceptor.class);
GlobalChannelInterceptorWrapper globalChannelInterceptorWrapper =
new GlobalChannelInterceptorWrapper(channelInterceptor);
ChannelInterceptorAware channel = Mockito.mock(ChannelInterceptorAware.class);
InterceptableChannel channel = Mockito.mock(InterceptableChannel.class);
globalChannelInterceptorWrapper.setPatterns(new String[] { "!te*", "!Te*" });
interceptors.put("Test-1", globalChannelInterceptorWrapper);
channels.put("Test-1", channel);
when(this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class))
.thenReturn(interceptors);
when(this.beanFactory.getBeansOfType(ChannelInterceptorAware.class))
when(this.beanFactory.getBeansOfType(InterceptableChannel.class))
.thenReturn(channels);
this.globalChannelInterceptorProcessor.afterSingletonsInstantiated();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -90,7 +90,7 @@ public class PollableJmsChannel extends AbstractJmsChannel
@Override
@Nullable
public Message<?> receive() {
ChannelInterceptorList interceptorList = getInterceptors();
ChannelInterceptorList interceptorList = getIChannelInterceptorList();
Deque<ChannelInterceptor> interceptorStack = null;
boolean counted = false;
boolean countsEnabled = isCountsEnabled();

20
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/GlobalChannelInterceptorTests.java
View File

@@ -25,8 +25,8 @@ import org.junit.Test;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
/**
* @author Oleg Zhurakousky
@@ -40,18 +40,20 @@ public class GlobalChannelInterceptorTests {
@Test
public void testJmsChannel() {
ActiveMqTestUtils.prepare();
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"GlobalChannelInterceptorTests-context.xml", GlobalChannelInterceptorTests.class);
ChannelInterceptorAware jmsChannel = context.getBean("jmsChannel", AbstractMessageChannel.class);
List<ChannelInterceptor> interceptors = jmsChannel.getChannelInterceptors();
assertThat(interceptors).isNotNull();
assertThat(interceptors.size()).isEqualTo(1);
assertThat(interceptors.get(0) instanceof SampleInterceptor).isTrue();
context.close();
try (ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"GlobalChannelInterceptorTests-context.xml", GlobalChannelInterceptorTests.class)) {
InterceptableChannel jmsChannel = context.getBean("jmsChannel", AbstractMessageChannel.class);
List<ChannelInterceptor> interceptors = jmsChannel.getInterceptors();
assertThat(interceptors).isNotNull();
assertThat(interceptors.size()).isEqualTo(1);
assertThat(interceptors.get(0) instanceof SampleInterceptor).isTrue();
}
}
public static class SampleInterceptor implements ChannelInterceptor {
}
}

41
spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java
View File

@@ -43,7 +43,6 @@ import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -74,6 +73,7 @@ import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.InterceptableChannel;
import org.springframework.stereotype.Component;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
@@ -160,8 +160,8 @@ public class JmsTests extends ActiveMQMultiContextTests {
}
this.controlBus.send("@'jmsTests.ContextConfiguration.integerMessageSource.inboundChannelAdapter'.stop()");
assertThat(((ChannelInterceptorAware) this.outputChannel).getChannelInterceptors()
.contains(this.testChannelInterceptor)).isTrue();
assertThat(((InterceptableChannel) this.outputChannel).getInterceptors())
.contains(this.testChannelInterceptor);
assertThat(this.testChannelInterceptor.invoked.get()).isGreaterThanOrEqualTo(5);
}
@@ -179,8 +179,10 @@ public class JmsTests extends ActiveMQMultiContextTests {
Message<?> receive = this.jmsOutboundInboundReplyChannel.receive(10000);
assertThat(receive).isNotNull();
assertThat(receive.getPayload()).isEqualTo("HELLO THROUGH THE JMS");
assertThat(receive)
.isNotNull()
.extracting(Message::getPayload)
.isEqualTo("HELLO THROUGH THE JMS");
this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS")
.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsMessageDriven")
@@ -188,8 +190,10 @@ public class JmsTests extends ActiveMQMultiContextTests {
receive = this.jmsOutboundInboundReplyChannel.receive(10000);
assertThat(receive).isNotNull();
assertThat(receive.getPayload()).isEqualTo("hello through the jms");
assertThat(receive)
.isNotNull()
.extracting(Message::getPayload)
.isEqualTo("hello through the jms");
assertThat(this.jmsMessageDrivenChannelCalled.get()).isTrue();
@@ -199,8 +203,10 @@ public class JmsTests extends ActiveMQMultiContextTests {
receive = this.jmsOutboundInboundReplyChannel.receive(10000);
assertThat(receive).isNotNull();
assertThat(receive.getPayload()).isEqualTo("foo");
assertThat(receive)
.isNotNull()
.extracting(Message::getPayload)
.isEqualTo("foo");
assertThat(this.jmsOutboundFlowTemplate).isNotNull();
}
@@ -218,8 +224,10 @@ public class JmsTests extends ActiveMQMultiContextTests {
Message<?> receive = replyChannel.receive(5000);
assertThat(receive).isNotNull();
assertThat(receive.getPayload()).isEqualTo("HELLO THROUGH THE JMS PIPELINE");
assertThat(receive)
.isNotNull()
.extracting(Message::getPayload)
.isEqualTo("HELLO THROUGH THE JMS PIPELINE");
assertThat(this.jmsInboundGatewayChannelCalled.get()).isTrue();
}
@@ -231,8 +239,10 @@ public class JmsTests extends ActiveMQMultiContextTests {
template.setDefaultDestinationName("pubsub");
template.convertAndSend("foo");
Message<?> received = this.jmsPubSubBridgeChannel.receive(5000);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("foo");
assertThat(received)
.isNotNull()
.extracting(Message::getPayload)
.isEqualTo("foo");
}
@Test
@@ -340,10 +350,11 @@ public class JmsTests extends ActiveMQMultiContextTests {
@Bean
public IntegrationFlow jmsMessageDrivenFlow() {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory(), DefaultMessageListenerContainer.class)
.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory(),
DefaultMessageListenerContainer.class)
.outputChannel(jmsMessageDrivenInputChannel())
.destination("jmsMessageDriven")
.configureListenerContainer(c -> c.clientId("foo")))
.configureListenerContainer(c -> c.clientId("foo")))
.<String, String>transform(String::toLowerCase)
.channel(jmsOutboundInboundReplyChannel())
.get();

70
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java
View File

@@ -30,13 +30,13 @@ import org.junit.Test;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.interceptor.WireTap;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.InterceptableChannel;
/**
* @author Dave Syer
@@ -74,16 +74,13 @@ public class MessageChannelsMonitorIntegrationTests {
@Test
public void testRates() throws Exception {
ClassPathXmlApplicationContext context = createContext("anonymous-channel.xml", "anonymous");
try {
try (ClassPathXmlApplicationContext context = createContext("anonymous-channel.xml")) {
this.channel = context.getBean("anonymous", MessageChannel.class);
int before = service.getCounter();
CountDownLatch latch = new CountDownLatch(50);
service.setLatch(latch);
for (int i = 0; i < 50; i++) {
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("bar"));
Thread.sleep(20L);
}
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -96,32 +93,27 @@ public class MessageChannelsMonitorIntegrationTests {
assertThat(sends).as("No send statistics for input channel").isEqualTo(sendsLong);
}
finally {
context.close();
}
}
@Test
public void testErrors() throws Exception {
ClassPathXmlApplicationContext context = createContext("anonymous-channel.xml", "anonymous");
try {
try (ClassPathXmlApplicationContext context = createContext("anonymous-channel.xml")) {
this.channel = context.getBean("anonymous", MessageChannel.class);
int before = service.getCounter();
CountDownLatch latch = new CountDownLatch(10);
service.setLatch(latch);
for (int i = 0; i < 5; i++) {
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("bar"));
Thread.sleep(20L);
}
try {
channel.send(new GenericMessage<String>("fail"));
channel.send(new GenericMessage<>("fail"));
}
catch (MessageHandlingException e) {
// ignore
}
for (int i = 0; i < 5; i++) {
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("bar"));
Thread.sleep(20L);
}
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -133,32 +125,27 @@ public class MessageChannelsMonitorIntegrationTests {
int errors = messageChannelsMonitor.getChannelErrorRate("" + channel).getCount();
assertThat(errors).as("No error statistics for input channel").isEqualTo(1);
}
finally {
context.close();
}
}
@Test
public void testQueues() throws Exception {
ClassPathXmlApplicationContext context = createContext("queue-channel.xml", "queue");
try {
try (ClassPathXmlApplicationContext context = createContext("queue-channel.xml")) {
this.channel = context.getBean("queue", MessageChannel.class);
int before = service.getCounter();
CountDownLatch latch = new CountDownLatch(10);
service.setLatch(latch);
for (int i = 0; i < 5; i++) {
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("bar"));
Thread.sleep(20L);
}
try {
channel.send(new GenericMessage<String>("fail"));
channel.send(new GenericMessage<>("fail"));
}
catch (MessageHandlingException e) {
// ignore
}
for (int i = 0; i < 5; i++) {
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("bar"));
Thread.sleep(20L);
}
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -172,22 +159,15 @@ public class MessageChannelsMonitorIntegrationTests {
int errors = messageChannelsMonitor.getChannelErrorRate("" + channel).getCount();
assertThat(errors).as("Expect no errors for input channel (handler fails)").isEqualTo(0);
}
finally {
context.close();
}
}
private void doTest(String config, String channelName) throws Exception {
ClassPathXmlApplicationContext context = createContext(config, channelName);
try {
try (ClassPathXmlApplicationContext context = createContext(config)) {
this.channel = context.getBean(channelName, MessageChannel.class);
int before = service.getCounter();
CountDownLatch latch = new CountDownLatch(1);
service.setLatch(latch);
channel.send(new GenericMessage<String>("bar"));
channel.send(new GenericMessage<>("bar"));
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(service.getCounter()).isEqualTo(before + 1);
@@ -195,23 +175,17 @@ public class MessageChannelsMonitorIntegrationTests {
int sends = messageChannelsMonitor.getChannelSendRate("" + channel).getCount();
assertThat(sends).as("No statistics for input channel").isEqualTo(1);
assertThat(channel).isInstanceOf(ChannelInterceptorAware.class);
List<ChannelInterceptor> channelInterceptors =
((ChannelInterceptorAware) channel).getChannelInterceptors();
assertThat(channel).isInstanceOf(InterceptableChannel.class);
List<ChannelInterceptor> channelInterceptors = ((InterceptableChannel) channel).getInterceptors();
assertThat(channelInterceptors.size()).isEqualTo(1);
assertThat(channelInterceptors.get(0)).isInstanceOf(WireTap.class);
}
finally {
context.close();
}
}
private ClassPathXmlApplicationContext createContext(String config, String channelName) {
private ClassPathXmlApplicationContext createContext(String config) {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(config, getClass());
context.getAutowireCapableBeanFactory().autowireBeanProperties(this,
AutowireCapableBeanFactory.AUTOWIRE_BY_TYPE, false);
channel = context.getBean(channelName, MessageChannel.class);
context.getAutowireCapableBeanFactory()
.autowireBeanProperties(this, AutowireCapableBeanFactory.AUTOWIRE_BY_TYPE, false);
return context;
}

1
src/reference/asciidoc/channel.adoc
View File

@@ -304,6 +304,7 @@ Also, starting with version 5.1, `ChannelInterceptor.postReceive()` is no longer
Previously, the method was called.
If you have an interceptor that relies on the previous behavior, implement `afterReceiveCompleted()` instead, since that method is invoked, regardless of whether a message is received or not.
NOTE: Starting with version 5.2, the `ChannelInterceptorAware` is deprecated in favor of `InterceptableChannel` from the Spring Messaging module, which it extends now for backward compatibility.
[[channel-template]]
==== `MessagingTemplate`

6
src/reference/asciidoc/dsl.adoc
View File

@@ -620,9 +620,9 @@ public IntegrationFlow loggingFlow() {
[IMPORTANT]
====
If the `MessageChannel` is an instance of `ChannelInterceptorAware`, the `log()` or `wireTap()` operators are applied to the current `MessageChannel`.
If the `MessageChannel` is an instance of `InterceptableChannel`, the `log()` or `wireTap()` operators are applied to the current `MessageChannel`.
Otherwise, an intermediate `DirectChannel` is injected into the flow for the currently configured endpoint.
In the following example, the `WireTap` interceptor is added to `myChannel` directly, because `DirectChannel` implements `ChannelInterceptorAware`:
In the following example, the `WireTap` interceptor is added to `myChannel` directly, because `DirectChannel` implements `InterceptableChannel`:
[source,java]
----
@@ -638,7 +638,7 @@ MessageChannel myChannel() {
----
====
When the current `MessageChannel` does not implement `ChannelInterceptorAware`, an implicit `DirectChannel` and `BridgeHandler` are injected into the `IntegrationFlow`, and the `WireTap` is added to this new `DirectChannel`.
When the current `MessageChannel` does not implement `InterceptableChannel`, an implicit `DirectChannel` and `BridgeHandler` are injected into the `IntegrationFlow`, and the `WireTap` is added to this new `DirectChannel`.
The following example does not have any channel declaration:
====