DSL: Add Amqp.*channel() support

* Upgrade to the latest Boot snapshot
* Polishing and workarounds for `Proxy` beans. Caused by `@EnableJmx`

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java

@@ -77,4 +77,33 @@ public abstract class Amqp {
 		return new AmqpOutboundEndpointSpec(endpoint, expectReply);
 	}
 
+	public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S>
+	pollableChannel(ConnectionFactory connectionFactory) {
+		return pollableChannel(null, connectionFactory);
+	}
+
+	public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S>
+	pollableChannel(String id, ConnectionFactory connectionFactory) {
+		return new AmqpPollableMessageChannelSpec<S>(connectionFactory).id(id);
+	}
+
+	public static
+	<S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(ConnectionFactory connectionFactory) {
+		return channel(null, connectionFactory);
+	}
+
+	public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(String id,
+			ConnectionFactory connectionFactory) {
+		return new AmqpMessageChannelSpec<S>(connectionFactory).id(id);
+	}
+
+	public static AmqpPublishSubscribeMessageChannelSpec publishSubscribeChannel(ConnectionFactory connectionFactory) {
+		return publishSubscribeChannel(null, connectionFactory);
+	}
+
+	public static AmqpPublishSubscribeMessageChannelSpec publishSubscribeChannel(String id,
+			ConnectionFactory connectionFactory) {
+		return new AmqpPublishSubscribeMessageChannelSpec(connectionFactory).id(id);
+	}
+
 }

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpMessageChannelSpec.java

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2014 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.dsl.amqp;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
+import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.TransactionAttribute;
+import org.springframework.util.ErrorHandler;
+
+/**
+ * @author Artem Bilan
+ */
+public class AmqpMessageChannelSpec<S extends AmqpMessageChannelSpec<S>> extends AmqpPollableMessageChannelSpec<S> {
+
+	private final List<Advice> adviceChain = new LinkedList<Advice>();
+
+	AmqpMessageChannelSpec(ConnectionFactory connectionFactory) {
+		super(new AmqpChannelFactoryBean(true), connectionFactory);
+	}
+
+	public S maxSubscribers(int maxSubscribers) {
+		this.amqpChannelFactoryBean.setMaxSubscribers(maxSubscribers);
+		return _this();
+	}
+
+	public S acknowledgeMode(AcknowledgeMode acknowledgeMode) {
+		this.amqpChannelFactoryBean.setAcknowledgeMode(acknowledgeMode);
+		return _this();
+	}
+
+	public S advice(Advice... advice) {
+		this.adviceChain.addAll(Arrays.asList(advice));
+		return _this();
+	}
+
+	public S autoStartup(boolean autoStartup) {
+		this.amqpChannelFactoryBean.setAutoStartup(autoStartup);
+		return _this();
+	}
+
+	public S concurrentConsumers(int concurrentConsumers) {
+		this.amqpChannelFactoryBean.setConcurrentConsumers(concurrentConsumers);
+		return _this();
+	}
+
+	public S errorHandler(ErrorHandler errorHandler) {
+		this.amqpChannelFactoryBean.setErrorHandler(errorHandler);
+		return _this();
+	}
+
+	public S exposeListenerChannel(boolean exposeListenerChannel) {
+		this.amqpChannelFactoryBean.setExposeListenerChannel(exposeListenerChannel);
+		return _this();
+	}
+
+	public S phase(int phase) {
+		this.amqpChannelFactoryBean.setPhase(phase);
+		return _this();
+	}
+
+	public S prefetchCount(int prefetchCount) {
+		this.amqpChannelFactoryBean.setPrefetchCount(prefetchCount);
+		return _this();
+	}
+
+	public S receiveTimeout(long receiveTimeout) {
+		this.amqpChannelFactoryBean.setReceiveTimeout(receiveTimeout);
+		return _this();
+	}
+
+	public S recoveryInterval(long recoveryInterval) {
+		this.amqpChannelFactoryBean.setRecoveryInterval(recoveryInterval);
+		return _this();
+	}
+
+	public S shutdownTimeout(long shutdownTimeout) {
+		this.amqpChannelFactoryBean.setShutdownTimeout(shutdownTimeout);
+		return _this();
+	}
+
+	public S taskExecutor(Executor taskExecutor) {
+		this.amqpChannelFactoryBean.setTaskExecutor(taskExecutor);
+		return _this();
+	}
+
+	public S transactionAttribute(TransactionAttribute transactionAttribute) {
+		this.amqpChannelFactoryBean.setTransactionAttribute(transactionAttribute);
+		return _this();
+	}
+
+	public S transactionManager(PlatformTransactionManager transactionManager) {
+		this.amqpChannelFactoryBean.setTransactionManager(transactionManager);
+		return _this();
+	}
+
+	public S txSize(int txSize) {
+		this.amqpChannelFactoryBean.setTxSize(txSize);
+		return _this();
+	}
+
+	@Override
+	protected AbstractAmqpChannel doGet() {
+		this.amqpChannelFactoryBean.setAdviceChain(this.adviceChain.toArray(new Advice[this.adviceChain.size()]));
+		return super.doGet();
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPollableMessageChannelSpec.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2014 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.dsl.amqp;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
+import org.springframework.integration.amqp.config.AmqpChannelFactoryBean;
+import org.springframework.integration.dsl.channel.MessageChannelSpec;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+public class AmqpPollableMessageChannelSpec<S extends AmqpPollableMessageChannelSpec<S>>
+		extends MessageChannelSpec<S, AbstractAmqpChannel> {
+
+	protected final AmqpChannelFactoryBean amqpChannelFactoryBean;
+
+	AmqpPollableMessageChannelSpec(ConnectionFactory connectionFactory) {
+		this(new AmqpChannelFactoryBean(false), connectionFactory);
+	}
+
+	AmqpPollableMessageChannelSpec(AmqpChannelFactoryBean amqpChannelFactoryBean, ConnectionFactory connectionFactory) {
+		this.amqpChannelFactoryBean = amqpChannelFactoryBean;
+		this.amqpChannelFactoryBean.setConnectionFactory(connectionFactory);
+		this.amqpChannelFactoryBean.setSingleton(false);
+		this.amqpChannelFactoryBean.setPubSub(false);
+	}
+
+	@Override
+	protected S id(String id) {
+		this.amqpChannelFactoryBean.setBeanName(id);
+		return super.id(id);
+	}
+
+	public S queueName(String queueName) {
+		if (this.id == null) {
+			id(queueName + ".channel");
+		}
+		this.amqpChannelFactoryBean.setQueueName(queueName);
+		return _this();
+	}
+
+	public S encoding(String encoding) {
+		this.amqpChannelFactoryBean.setEncoding(encoding);
+		return _this();
+	}
+
+	public S amqpMessageConverter(MessageConverter messageConverter) {
+		this.amqpChannelFactoryBean.setMessageConverter(messageConverter);
+		return _this();
+	}
+
+	public S channelTransacted(boolean channelTransacted) {
+		this.amqpChannelFactoryBean.setChannelTransacted(channelTransacted);
+		return _this();
+	}
+
+	public S messagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) {
+		this.amqpChannelFactoryBean.setMessagePropertiesConverter(messagePropertiesConverter);
+		return _this();
+	}
+
+	@Override
+	protected AbstractAmqpChannel doGet() {
+		Assert.notNull(this.id, "The 'id' or 'queueName' must be specified");
+		try {
+			this.channel = this.amqpChannelFactoryBean.getObject();
+		}
+		catch (Exception e) {
+			throw new BeanCreationException("Cannot create the AMQP MessageChannel", e);
+		}
+		return super.doGet();
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpPublishSubscribeMessageChannelSpec.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2014 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.dsl.amqp;
+
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+
+/**
+ * @author Artem Bilan
+ */
+public class AmqpPublishSubscribeMessageChannelSpec
+		extends AmqpMessageChannelSpec<AmqpPublishSubscribeMessageChannelSpec> {
+
+	AmqpPublishSubscribeMessageChannelSpec(ConnectionFactory connectionFactory) {
+		super(connectionFactory);
+		this.amqpChannelFactoryBean.setPubSub(true);
+	}
+
+	public AmqpPublishSubscribeMessageChannelSpec exchange(FanoutExchange exchange) {
+		this.amqpChannelFactoryBean.setExchange(exchange);
+		return _this();
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java

@@ -18,6 +18,7 @@ package org.springframework.integration.dsl.core;
 
 import java.util.Collection;
 
+import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -31,6 +32,7 @@ import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.config.IntegrationConfigUtils;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.support.MessageChannelReference;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 
@@ -60,7 +62,7 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
 	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
 		if (bean instanceof IntegrationFlow) {
 			IntegrationFlow flow = (IntegrationFlow) bean;
-			String flowNamePrefix = beanName + ":";
+			String flowNamePrefix = beanName + ".";
 			int channelNameIndex = 0;
 			for (Object component : flow.getIntegrationComponents()) {
 				if (component instanceof ConsumerEndpointSpec) {
@@ -69,8 +71,8 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
 					ConsumerEndpointFactoryBean endpoint = endpointSpec.get().getT1();
 					String id = endpointSpec.getId();
 
-					Collection<?> messageHandlers =
-							this.beanFactory.getBeansOfType(messageHandler.getClass(), false, false).values();
+					Collection<?> messageHandlers = this.beanFactory.getBeansOfType(MessageHandler.class, false,
+							false).values();
 
 					if (!messageHandlers.contains(messageHandler)) {
 						String handlerBeanName = generateBeanName(messageHandler);
@@ -93,35 +95,42 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
 					registerComponent(endpoint, endpointBeanName);
 				}
 				else {
-					Collection<?> values = this.beanFactory.getBeansOfType(component.getClass(), false, false).values();
-					if (!values.contains(component)) {
-						if (component instanceof AbstractMessageChannel) {
-							String channelBeanName = ((AbstractMessageChannel) component).getComponentName();
-							if (channelBeanName == null) {
-								channelBeanName = flowNamePrefix + "channel" +
-										BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
+					//TODO workaround until SF will fix 'TypeDescriptor.forObject'
+					if (component instanceof MessageChannel) {
+						Collection<?> messageChannels =
+								this.beanFactory.getBeansOfType(MessageChannel.class, false, false).values();
+						if (!messageChannels.contains(component)) {
+							if (component instanceof AbstractMessageChannel) {
+								String channelBeanName = ((AbstractMessageChannel) component).getComponentName();
+								if (channelBeanName == null) {
+									channelBeanName = flowNamePrefix + "channel" +
+											BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
+								}
+								registerComponent(component, channelBeanName);
 							}
-							registerComponent(component, channelBeanName);
-						}
-						else if (component instanceof MessageChannelReference) {
-							String channelBeanName = ((MessageChannelReference) component).getName();
-							if (!this.beanFactory.containsBean(channelBeanName)) {
-								DirectChannel directChannel = new DirectChannel();
-								registerComponent(directChannel, channelBeanName);
+							else if (component instanceof MessageChannelReference) {
+								String channelBeanName = ((MessageChannelReference) component).getName();
+								if (!this.beanFactory.containsBean(channelBeanName)) {
+									DirectChannel directChannel = new DirectChannel();
+									registerComponent(directChannel, channelBeanName);
+								}
+							}
+							else if (component instanceof FixedSubscriberChannel) {
+								FixedSubscriberChannel fixedSubscriberChannel = (FixedSubscriberChannel) component;
+								String channelBeanName = fixedSubscriberChannel.getComponentName();
+								if ("Unnamed fixed subscriber channel".equals(channelBeanName)) {
+									channelBeanName = flowNamePrefix + "channel" +
+											BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
+								}
+								registerComponent(component, channelBeanName);
 							}
 						}
-						else if (component instanceof FixedSubscriberChannel) {
-							FixedSubscriberChannel fixedSubscriberChannel = (FixedSubscriberChannel) component;
-							String channelBeanName = fixedSubscriberChannel.getComponentName();
-							if ("Unnamed fixed subscriber channel".equals(channelBeanName)) {
-								channelBeanName = flowNamePrefix + "channel" +
-										BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
-							}
-							registerComponent(component, channelBeanName);
-						}
-						else {
-							registerComponent(component, generateBeanName(component));
-						}
+					}
+					else if (!this.beanFactory
+							.getBeansOfType(AopUtils.getTargetClass(component), false, false)
+							.values()
+							.contains(component)) {
+						registerComponent(component, generateBeanName(component));
 					}
 				}
 			}

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java

@@ -54,11 +54,16 @@ import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.AnonymousQueue;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.aop.TargetSource;
+import org.springframework.aop.framework.Advised;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.ConfigFileApplicationContextInitializer;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -79,7 +84,6 @@ import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.AbstractPollableChannel;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
-import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.GlobalChannelInterceptor;
@@ -113,10 +117,12 @@ import org.springframework.integration.xml.transformer.support.XPathExpressionEv
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
+import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.messaging.support.ChannelInterceptorAdapter;
 import org.springframework.messaging.support.ErrorMessage;
@@ -130,7 +136,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 /**
  * @author Artem Bilan
  */
-@ContextConfiguration
+@ContextConfiguration(initializers = ConfigFileApplicationContextInitializer.class)
 @RunWith(SpringJUnit4ClassRunner.class)
 @DirtiesContext
 public class IntegrationFlowTests {
@@ -158,11 +164,11 @@ public class IntegrationFlowTests {
 
 	@Autowired
 	@Qualifier("inputChannel")
-	private DirectChannel inputChannel;
+	private MessageChannel inputChannel;
 
 	@Autowired
 	@Qualifier("foo")
-	private PublishSubscribeChannel foo;
+	private SubscribableChannel foo;
 
 	@Autowired
 	@Qualifier("successChannel")
@@ -170,7 +176,7 @@ public class IntegrationFlowTests {
 
 	@Autowired
 	@Qualifier("flow3Input")
-	private DirectChannel flow3Input;
+	private MessageChannel flow3Input;
 
 	@Autowired
 	private AtomicReference<Object> eventHolder;
@@ -185,7 +191,7 @@ public class IntegrationFlowTests {
 
 	@Autowired
 	@Qualifier("bridgeFlow2Input")
-	private DirectChannel bridgeFlow2Input;
+	private MessageChannel bridgeFlow2Input;
 
 	@Autowired
 	@Qualifier("bridgeFlow2Output")
@@ -193,15 +199,15 @@ public class IntegrationFlowTests {
 
 	@Autowired
 	@Qualifier("fileFlow1Input")
-	private DirectChannel fileFlow1Input;
+	private MessageChannel fileFlow1Input;
 
 	@Autowired
 	@Qualifier("fileWritingMessageHandler")
-	private FileWritingMessageHandler fileWritingMessageHandler;
+	private MessageHandler fileWritingMessageHandler;
 
 	@Autowired
 	@Qualifier("methodInvokingInput")
-	private DirectChannel methodInvokingInput;
+	private MessageChannel methodInvokingInput;
 
 	@Autowired
 	@Qualifier("delayedAdvice")
@@ -213,11 +219,11 @@ public class IntegrationFlowTests {
 
 	@Autowired
 	@Qualifier("splitInput")
-	private DirectChannel splitInput;
+	private MessageChannel splitInput;
 
 	@Autowired
 	@Qualifier("xpathHeaderEnricherInput")
-	private DirectChannel xpathHeaderEnricherInput;
+	private MessageChannel xpathHeaderEnricherInput;
 
 	@Autowired
 	@Qualifier("splitAggregateInput")
@@ -265,7 +271,7 @@ public class IntegrationFlowTests {
 
 	@Autowired
 	@Qualifier("defaultOutputChannel")
-	private QueueChannel defaultOutputChannel;
+	private PollableChannel defaultOutputChannel;
 
 	@Autowired
 	private MessageStore messageStore;
@@ -384,8 +390,9 @@ public class IntegrationFlowTests {
 		assertNotNull(reply);
 		assertEquals("test", reply.getPayload());
 
-		assertTrue(this.beanFactory.containsBean("bridgeFlow2:channel#0"));
-		assertThat(this.beanFactory.getBean("bridgeFlow2:channel#0"), Matchers.instanceOf(FixedSubscriberChannel.class));
+		assertTrue(this.beanFactory.containsBean("bridgeFlow2.channel#0"));
+		assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0"), Matchers.instanceOf(FixedSubscriberChannel
+				.class));
 
 		try {
 			this.bridgeFlow2Input.send(message);
@@ -443,8 +450,7 @@ public class IntegrationFlowTests {
 
 
 	@Test
-	public void testFileHandler() {
-		assertEquals(1, this.beanFactory.getBeansOfType(FileWritingMessageHandler.class).size());
+	public void testFileHandler() throws Exception {
 		Message<?> message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build();
 		try {
 			this.fileFlow1Input.send(message);
@@ -456,7 +462,15 @@ public class IntegrationFlowTests {
 		}
 		DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
 		fileNameGenerator.setBeanFactory(this.beanFactory);
-		this.fileWritingMessageHandler.setFileNameGenerator(fileNameGenerator);
+		Object targetFileWritingMessageHandler = this.fileWritingMessageHandler;
+		if (this.fileWritingMessageHandler instanceof Advised) {
+			TargetSource targetSource = ((Advised) this.fileWritingMessageHandler).getTargetSource();
+			if (targetSource != null) {
+				targetFileWritingMessageHandler = targetSource.getTarget();
+			}
+		}
+		DirectFieldAccessor dfa = new DirectFieldAccessor(targetFileWritingMessageHandler);
+		dfa.setPropertyValue("fileNameGenerator", fileNameGenerator);
 		this.fileFlow1Input.send(message);
 
 		assertTrue(new File(tmpDir, "foo").exists());
@@ -901,7 +915,7 @@ public class IntegrationFlowTests {
 	private MessageChannel amqpOutboundInput;
 
 	@Autowired
-	@Qualifier("amqpReplyChannel")
+	@Qualifier("amqpReplyChannel.channel")
 	private PollableChannel amqpReplyChannel;
 
 	@Test
@@ -909,13 +923,22 @@ public class IntegrationFlowTests {
 		this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp")
 				.setHeader("routingKey", "foo")
 				.build());
-		Message<?> receive = this.amqpReplyChannel.receive(5000);
+		Message<?> receive = null;
+		int i = 0;
+		do {
+			receive = this.amqpReplyChannel.receive();
+			if (receive != null) {
+				break;
+			}
+			Thread.sleep(100);
+			i++;
+		} while (i < 10);
+
 		assertNotNull(receive);
 		assertEquals("HELLO THROUGH THE AMQP", receive.getPayload());
 	}
 
 
-
 	@MessagingGateway(defaultRequestChannel = "controlBus")
 	private static interface ControlBusGateway {
 
@@ -955,12 +978,12 @@ public class IntegrationFlowTests {
 		}
 
 		@Bean
-		public DirectChannel inputChannel() {
+		public MessageChannel inputChannel() {
 			return MessageChannels.direct().get();
 		}
 
 		@Bean
-		public PublishSubscribeChannel foo() {
+		public MessageChannel foo() {
 			return MessageChannels.publishSubscribe().get();
 		}
 
@@ -1106,17 +1129,17 @@ public class IntegrationFlowTests {
 		}
 
 		@Bean(name = "foo-channel")
-		public QueueChannel fooChannel() {
+		public MessageChannel fooChannel() {
 			return new QueueChannel();
 		}
 
 		@Bean(name = "bar-channel")
-		public QueueChannel barChannel() {
+		public MessageChannel barChannel() {
 			return new QueueChannel();
 		}
 
 		@Bean
-		public QueueChannel defaultOutputChannel() {
+		public MessageChannel defaultOutputChannel() {
 			return new QueueChannel();
 		}
 
@@ -1125,7 +1148,7 @@ public class IntegrationFlowTests {
 			return IntegrationFlows.from("recipientListInput")
 					.<String, String>transform(p -> p.replaceFirst("Payload", ""))
 					.recipientListRoute(r ->
-									r.defaultOutputChannel(defaultOutputChannel())
+									r.defaultOutputChannel(this.defaultOutputChannel())
 											.recipient("foo-channel", "'foo' == payload")
 											.recipient("bar-channel", m ->
 													m.getHeaders().containsKey("recipient")
@@ -1156,28 +1179,28 @@ public class IntegrationFlowTests {
 	public static class ContextConfiguration4 {
 
 		@Bean
-		public FileWritingMessageHandler fileWritingMessageHandler() {
-			return new FileWritingMessageHandler(tmpDir);
+		public MessageHandler fileWritingMessageHandler() {
+			FileWritingMessageHandler fileWritingMessageHandler = new FileWritingMessageHandler(tmpDir);
+			fileWritingMessageHandler.setFileNameGenerator(message -> null);
+			fileWritingMessageHandler.setExpectReply(false);
+			return fileWritingMessageHandler;
 		}
 
 		@Bean
 		public IntegrationFlow fileFlow1() {
 			return IntegrationFlows.from("fileFlow1Input")
-					.handle(this.fileWritingMessageHandler(), c -> {
-						FileWritingMessageHandler handler = c.get().getT2();
-						handler.setFileNameGenerator(message -> null);
-						handler.setExpectReply(false);
-					})
+					.handle(this.fileWritingMessageHandler())
 					.get();
 		}
 
 		@Autowired
-		private GreetingService greetingService;
+		@Qualifier("integrationFlowTests.GreetingService")
+		private MessageHandler greetingService;
 
 		@Bean
 		public IntegrationFlow methodInvokingFlow() {
 			return IntegrationFlows.from("methodInvokingInput")
-					.handle(Message.class, (p, h) -> this.greetingService.handleRequestMessage(p))
+					.handle(this.greetingService)
 					.get();
 		}
 
@@ -1352,7 +1375,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow amqpOutboundFlow() {
-			return IntegrationFlows.from("amqpOutboundInput")
+			return IntegrationFlows.from(Amqp.channel("amqpOutboundInput", this.rabbitConnectionFactory))
 					.handle(Amqp.outboundAdapter(this.amqpTemplate).routingKeyExpression("headers.routingKey").get())
 					.get();
 		}
@@ -1362,11 +1385,18 @@ public class IntegrationFlowTests {
 			return new Queue("foo");
 		}
 
+		@Bean
+		public Queue amqpReplyChannel() {
+			return new Queue("amqpReplyChannel");
+		}
+
 		@Bean
 		public IntegrationFlow amqpInboundFlow() {
 			return IntegrationFlows.from(Amqp.inboundAdapter(this.rabbitConnectionFactory, fooQueue()))
 					.transform(String.class, String::toUpperCase)
-					.channel(MessageChannels.queue("amqpReplyChannel"))
+					.channel(Amqp.pollableChannel(this.rabbitConnectionFactory)
+							.queueName("amqpReplyChannel")
+							.channelTransacted(true))
 					.get();
 		}
 

spring-integration-java-dsl/src/test/resources/application.properties

@@ -0,0 +1 @@
+#spring.jmx.enabled=false