INT-4270: CGLib Compatibility

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

Initial Commit

Remove `final` modifier from certain critical framework methods that prevent CGLib proxies working.

Since Spring Boot 2.0 now uses proxyTargetClass=true by default (and has done that for
transactional proxies since 1.4), we must relax these restrictions so that CGLib can
proxy channels (which is often done to make subflows run in a transaction).

When CGLib can't override a `final` method, the fields used within those methods are uninitialized.
In the case of `AbstractMessageChannel` this causes an NPE on the unitiallized `dataTypes` field.
With `MessageHandler`, fields like `shouldTrack` are always false.

Further, methods on `IntegrationObjectSupport` - such as `getComponentName()` return null.

I have left setters and `afterPropertiesSet` (on IOS) `final` since these will typically be called
before the object is proxied.

I suspect there will be other methods we need to open up, but perhaps we should only do them on-demand.

__cherry-pick to 4.3.x__

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
 	 * <code>false</code> 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) {

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<ChannelInterceptor> interceptorStack = null;
 		boolean counted = false;

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) {

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()) {

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<?>> 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<?>> 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<?>> 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;
+		}
+
+	}
+
+}

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;
+		}
+
+	}
+
+}