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.

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java

@@ -60,9 +60,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;
 
@@ -72,8 +74,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;
@@ -98,7 +98,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private volatile Integer prefetchCount;
 
-	private volatile Boolean isPubSub;
+	private volatile boolean isPubSub;
 
 	private volatile Long receiveTimeout;
 
@@ -315,13 +315,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);
 				}
@@ -332,7 +332,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);
 				}
@@ -343,8 +343,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);
@@ -428,8 +427,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	@Override
 	public boolean isAutoStartup() {
-		return (this.channel instanceof SmartLifecycle) ?
-				((SmartLifecycle) this.channel).isAutoStartup() : false;
+		return (this.channel instanceof SmartLifecycle) && ((SmartLifecycle) this.channel).isAutoStartup();
 	}
 
 	@Override
@@ -440,8 +438,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	@Override
 	public boolean isRunning() {
-		return (this.channel instanceof Lifecycle) ?
-				((Lifecycle) this.channel).isRunning() : false;
+		return (this.channel instanceof Lifecycle) && ((Lifecycle) this.channel).isRunning();
 	}
 
 	@Override

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * 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.
@@ -16,8 +16,10 @@
 
 package org.springframework.integration.amqp.channel;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
 import static org.mockito.Mockito.mock;
 
 import java.util.Collection;
@@ -31,10 +33,14 @@ import org.junit.runner.RunWith;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
 import org.springframework.integration.amqp.rule.BrokerRunning;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
@@ -46,6 +52,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 4.0
  *
  */
@@ -114,4 +121,24 @@ public class ChannelTests {
 		admin.deleteQueue("explicit");
 	}
 
+	@Test
+	public void testAmqpChannelFactoryBean() throws Exception {
+		AmqpChannelFactoryBean channelFactoryBean = new AmqpChannelFactoryBean();
+		channelFactoryBean.setBeanFactory(mock(BeanFactory.class));
+		channelFactoryBean.setConnectionFactory(this.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(this.factory);
+		channelFactoryBean.setBeanName("testChannel");
+		channelFactoryBean.setPubSub(true);
+		channelFactoryBean.afterPropertiesSet();
+		channel = channelFactoryBean.getObject();
+		assertThat(channel, instanceOf(PublishSubscribeAmqpChannel.class));
+	}
+
 }