From 1ada398c7de992a399c92760a1d0c685f034a93e Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Wed, 7 Oct 2015 09:56:35 -0400
Subject: [PATCH] INT-3841: Fix NPE in the `AmqpChannelFactoryBean`

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

Previously the `isPubSub` was as `Boolean` object and `null` by default.
Convert it to the primitive to achieve the `false` logic by default as expected.

INT-3841: Fix New Test

New test channel remains as a listener on the connection factory.

Conflicts:
	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java
Resolved.
---
 .../amqp/config/AmqpChannelFactoryBean.java   | 23 +++---
 .../amqp/channel/ChannelTests.java            | 70 +++++++++++++++++++
 2 files changed, 80 insertions(+), 13 deletions(-)
 create mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java

diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
index 934afde5ad..04dced1e77 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
@@ -58,9 +58,11 @@ import org.springframework.util.StringUtils;
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.1
  */
-public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel> implements SmartLifecycle, DisposableBean, BeanNameAware {
+public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel> implements SmartLifecycle,
+		DisposableBean, BeanNameAware {
 
 	private volatile AbstractAmqpChannel channel;
 
@@ -70,8 +72,6 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private final AmqpTemplate amqpTemplate = new RabbitTemplate();
 
-	private volatile SimpleMessageListenerContainer container;
-
 	private volatile AmqpAdmin amqpAdmin;
 
 	private volatile FanoutExchange exchange;
@@ -96,7 +96,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private volatile Integer prefetchCount;
 
-	private volatile Boolean isPubSub;
+	private volatile boolean isPubSub;
 
 	private volatile Long receiveTimeout;
 
@@ -297,13 +297,13 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 	@Override
 	protected AbstractAmqpChannel createInstance() throws Exception {
 		if (this.messageDriven) {
-			this.container = this.createContainer();
+			SimpleMessageListenerContainer container = this.createContainer();
 			if (this.amqpTemplate instanceof InitializingBean) {
 				((InitializingBean) this.amqpTemplate).afterPropertiesSet();
 			}
 			if (this.isPubSub) {
 				PublishSubscribeAmqpChannel pubsub = new PublishSubscribeAmqpChannel(
-						this.beanName, this.container, this.amqpTemplate);
+						this.beanName, container, this.amqpTemplate);
 				if (this.exchange != null) {
 					pubsub.setExchange(this.exchange);
 				}
@@ -314,7 +314,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 			}
 			else {
 				PointToPointSubscribableAmqpChannel p2p = new PointToPointSubscribableAmqpChannel(
-						this.beanName, this.container, this.amqpTemplate);
+						this.beanName, container, this.amqpTemplate);
 				if (StringUtils.hasText(this.queueName)) {
 					p2p.setQueueName(this.queueName);
 				}
@@ -325,8 +325,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 			}
 		}
 		else {
-			Assert.isTrue(!Boolean.TRUE.equals(this.isPubSub),
-					"An AMQP 'publish-subscribe-channel' must be message-driven.");
+			Assert.isTrue(!this.isPubSub, "An AMQP 'publish-subscribe-channel' must be message-driven.");
 			PollableAmqpChannel pollable = new PollableAmqpChannel(this.beanName, this.amqpTemplate);
 			if (this.amqpAdmin != null) {
 				pollable.setAmqpAdmin(this.amqpAdmin);
@@ -403,8 +402,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 	 */
 
 	public boolean isAutoStartup() {
-		return (this.channel instanceof SmartLifecycle) ?
-				((SmartLifecycle) this.channel).isAutoStartup() : false;
+		return (this.channel instanceof SmartLifecycle) && ((SmartLifecycle) this.channel).isAutoStartup();
 	}
 
 	public int getPhase() {
@@ -413,8 +411,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 	}
 
 	public boolean isRunning() {
-		return (this.channel instanceof Lifecycle) ?
-				((Lifecycle) this.channel).isRunning() : false;
+		return (this.channel instanceof Lifecycle) && ((Lifecycle) this.channel).isRunning();
 	}
 
 	public void start() {
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java
new file mode 100644
index 0000000000..eba79e502f
--- /dev/null
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2014-2015 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.amqp.channel;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+
+import org.springframework.amqp.rabbit.connection.Connection;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.impl.AMQImpl;
+
+/**
+ * @author Gary Russell
+ * @author Artem Bilan
+ * @since 3.0.8
+ *
+ */
+public class ChannelTests {
+
+
+	@Test
+	public void testAmqpChannelFactoryBean() throws Exception {
+		ConnectionFactory factory = mock(ConnectionFactory.class);
+		Connection connection = mock(Connection.class);
+		when(factory.createConnection()).thenReturn(connection);
+		Channel mockChannel = mock(Channel.class);
+		when(connection.createChannel(false)).thenReturn(mockChannel);
+		when(mockChannel.queueDeclare()).thenReturn(new AMQImpl.Queue.DeclareOk("foo", 0, 0));
+
+		AmqpChannelFactoryBean channelFactoryBean = new AmqpChannelFactoryBean();
+		channelFactoryBean.setBeanFactory(mock(BeanFactory.class));
+		channelFactoryBean.setConnectionFactory(factory);
+		channelFactoryBean.setBeanName("testChannel");
+		channelFactoryBean.afterPropertiesSet();
+		AbstractAmqpChannel channel = channelFactoryBean.getObject();
+		assertThat(channel, instanceOf(PointToPointSubscribableAmqpChannel.class));
+
+		channelFactoryBean = new AmqpChannelFactoryBean();
+		channelFactoryBean.setBeanFactory(mock(BeanFactory.class));
+		channelFactoryBean.setConnectionFactory(factory);
+		channelFactoryBean.setBeanName("testChannel");
+		channelFactoryBean.setPubSub(true);
+		channelFactoryBean.afterPropertiesSet();
+		channel = channelFactoryBean.getObject();
+		assertThat(channel, instanceOf(PublishSubscribeAmqpChannel.class));
+	}
+
+}