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 c3f1906c0c..63415250a6 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -369,7 +369,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport * false if the sending thread is interrupted. */ @Override - public final boolean send(Message message) { + public boolean send(Message message) { return this.send(message, -1); } @@ -388,7 +388,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport * time or the sending thread is interrupted. */ @Override - public final boolean send(Message message, long timeout) { + public boolean send(Message message, long timeout) { Assert.notNull(message, "message must not be null"); Assert.notNull(message.getPayload(), "message payload must not be null"); if (this.shouldTrack) { 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 563fc992c2..90642506b8 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2017 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. @@ -68,7 +68,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel * receiving thread is interrupted. */ @Override - public final Message receive() { + public Message receive() { return receive(-1); } @@ -86,7 +86,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel * interrupted. */ @Override - public final Message receive(long timeout) { + public Message receive(long timeout) { ChannelInterceptorList interceptorList = getInterceptors(); Deque interceptorStack = null; boolean counted = false; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index 04fa375183..b1ed21ff00 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java @@ -105,7 +105,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo * If {@link #componentName} was not set this method will default to the 'beanName' of this component; */ @Override - public final String getComponentName() { + public String getComponentName() { return StringUtils.hasText(this.componentName) ? this.componentName : this.beanName; } @@ -190,7 +190,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo protected void onInit() throws Exception { } - protected final BeanFactory getBeanFactory() { + protected BeanFactory getBeanFactory() { return this.beanFactory; } @@ -213,7 +213,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo this.taskScheduler = taskScheduler; } - public final ConversionService getConversionService() { + public ConversionService getConversionService() { if (this.conversionService == null && this.beanFactory != null) { synchronized (this) { if (this.conversionService == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index aa4c9d6d42..929cfbdb45 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2017 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. @@ -136,7 +136,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im } @Override - public final void handleMessage(Message message) { + public void handleMessage(Message message) { Assert.notNull(message, "Message must not be null"); Assert.notNull(message.getPayload(), "Message payload must not be null"); //NOSONAR - false positive if (this.loggingEnabled && this.logger.isDebugEnabled()) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/CGLibProxyChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/CGLibProxyChannelTests.java new file mode 100644 index 0000000000..ea426a7941 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/CGLibProxyChannelTests.java @@ -0,0 +1,144 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.channel; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.aop.framework.ProxyFactoryBean; +import org.springframework.aop.support.AopUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 4.3.10 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class CGLibProxyChannelTests { + + @Autowired + private DirectChannel directChannel; + + @Autowired + private QueueChannel queueChannel; + + @Autowired + private ExecutorChannel executorChannel; + + @Autowired + private PublishSubscribeChannel publishSubscribeChannel; + + @Test + public void testProxyDirect() { + assertTrue(AopUtils.isCglibProxy(this.directChannel)); + final AtomicReference> message = new AtomicReference<>(); + this.directChannel.subscribe(m -> message.set(m)); + this.directChannel.send(new GenericMessage<>("foo")); + assertNotNull(message.get()); + } + + @Test + public void testProxyQueue() { + assertTrue(AopUtils.isCglibProxy(this.queueChannel)); + this.queueChannel.send(new GenericMessage<>("foo")); + assertNotNull(this.queueChannel.receive(0)); + } + + @Test + public void testProxyExecutor() throws Exception { + assertTrue(AopUtils.isCglibProxy(this.executorChannel)); + final AtomicReference> message = new AtomicReference<>(); + final CountDownLatch latch = new CountDownLatch(1); + this.executorChannel.subscribe(m -> { + message.set(m); + latch.countDown(); + }); + this.executorChannel.send(new GenericMessage<>("foo")); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertNotNull(message.get()); + } + + @Test + public void testProxyPubSubWithExec() throws Exception { + assertTrue(AopUtils.isCglibProxy(this.publishSubscribeChannel)); + final AtomicReference> message = new AtomicReference<>(); + final CountDownLatch latch = new CountDownLatch(1); + this.publishSubscribeChannel.subscribe(m -> { + message.set(m); + latch.countDown(); + }); + this.publishSubscribeChannel.send(new GenericMessage<>("foo")); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertNotNull(message.get()); + } + + @Configuration + public static class Config { + + @Bean + public ProxyFactoryBean directChannel() { + return createProxyFactory(new DirectChannel()); + } + + @Bean + public ProxyFactoryBean queueChannel() { + return createProxyFactory(new QueueChannel()); + } + + @Bean + public ProxyFactoryBean executorChannel() { + return createProxyFactory(new ExecutorChannel(executor())); + } + + @Bean + public ProxyFactoryBean publishSubscribeChannel() { + return createProxyFactory(new PublishSubscribeChannel(executor())); + } + + @Bean + public ExecutorService executor() { + return Executors.newCachedThreadPool(); + } + + private ProxyFactoryBean createProxyFactory(MessageChannel channel) { + ProxyFactoryBean fb = new ProxyFactoryBean(); + fb.setProxyTargetClass(true); + fb.setTarget(channel); + return fb; + } + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/CGLibProxyHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/CGLibProxyHandlerTests.java new file mode 100644 index 0000000000..afd7043ca7 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/CGLibProxyHandlerTests.java @@ -0,0 +1,96 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.handler; + +import static org.hamcrest.CoreMatchers.equalTo; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.aop.framework.ProxyFactoryBean; +import org.springframework.aop.support.AopUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableMessageHistory; +import org.springframework.integration.history.MessageHistory; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 4.3.10 + * + */ +@RunWith(SpringRunner.class) +public class CGLibProxyHandlerTests { + + @Autowired + private BridgeHandler bridge; + + @Autowired + private QueueChannel out; + + @Test + public void testProxyBridge() { + assertTrue(AopUtils.isCglibProxy(this.bridge)); + this.bridge.handleMessage(new GenericMessage<>("foo")); + Message received = this.out.receive(0); + assertNotNull(received); + assertNotNull(received.getHeaders().get(MessageHistory.HEADER_NAME)); + MessageHistory history = received.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class); + assertThat(history.size(), equalTo(2)); + } + + @Configuration + @EnableMessageHistory + public static class Config { + + @Bean + public BridgeHandler bridgeTarget() { // separate bean so setters and aPS are called + BridgeHandler bridgeHandler = new BridgeHandler(); + bridgeHandler.setOutputChannel(out()); + bridgeHandler.setShouldTrack(true); + return bridgeHandler; + } + + @Bean + public ProxyFactoryBean bridge() { + return createProxyFactory(bridgeTarget()); + } + + @Bean + public QueueChannel out() { + return new QueueChannel(); + } + + private ProxyFactoryBean createProxyFactory(MessageHandler handler) { + ProxyFactoryBean fb = new ProxyFactoryBean(); + fb.setProxyTargetClass(true); + fb.setTarget(handler); + return fb; + } + + } + +}