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.

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

@@ -59,9 +59,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;
 
@@ -71,8 +73,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;
@@ -97,7 +97,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	private volatile Integer prefetchCount;
 
-	private volatile Boolean isPubSub;
+	private volatile boolean isPubSub;
 
 	private volatile Long receiveTimeout;
 
@@ -305,13 +305,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);
 				}
@@ -322,7 +322,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);
 				}
@@ -333,8 +333,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 			}
 		}
 		else {
-			Assert.isTrue(!Boolean.TRUE.equals(this.isPubSub),
+			Assert.isTrue(!this.isPubSub, "An AMQP 'publish-subscribe-channel' must be message-driven.");
-					"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);
@@ -415,8 +414,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	@Override
 	public boolean isAutoStartup() {
-		return (this.channel instanceof SmartLifecycle) ?
+		return (this.channel instanceof SmartLifecycle) && ((SmartLifecycle) this.channel).isAutoStartup();
-				((SmartLifecycle) this.channel).isAutoStartup() : false;
 	}
 
 	@Override
@@ -427,8 +425,7 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
 
 	@Override
 	public boolean isRunning() {
-		return (this.channel instanceof Lifecycle) ?
+		return (this.channel instanceof Lifecycle) && ((Lifecycle) this.channel).isRunning();
-				((Lifecycle) this.channel).isRunning() : false;
 	}
 
 	@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,7 +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.assertThat;
+import static org.mockito.Mockito.mock;
 
 import java.util.Collection;
 import java.util.concurrent.CyclicBarrier;
@@ -27,23 +30,29 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+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;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.annotation.DirtiesContext.ClassMode;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 4.0
  *
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext(classMode=ClassMode.AFTER_EACH_TEST_METHOD)
 public class ChannelTests {
 
 	@ClassRule
@@ -79,4 +88,24 @@ public class ChannelTests {
 		assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size());
 	}
 
+	@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));
+	}
+
 }