DSL: Add Jms.inboundGateway() infrastructure

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java

@@ -158,7 +158,8 @@ public final class IntegrationFlowBuilder {
 				(isLambda(genericTransformer)
 						? new MethodInvokingTransformer(new LambdaMessageProcessor(genericTransformer, payloadType))
 						: new MethodInvokingTransformer(genericTransformer));
-		return this.handle(new MessageTransformingHandler(transformer), endpointConfigurer);
+		return addComponent(transformer)
+				.handle(new MessageTransformingHandler(transformer), endpointConfigurer);
 	}
 
 	public IntegrationFlowBuilder filter(String expression) {

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java

@@ -19,6 +19,8 @@ package org.springframework.integration.dsl.jms;
 import javax.jms.ConnectionFactory;
 
 import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
 
 /**
  * @author Artem Bilan
@@ -72,4 +74,25 @@ public abstract class Jms {
 		return new JmsOutboundGatewaySpec(connectionFactory);
 	}
 
+	public static <S extends JmsInboundGatewaySpec<S>> JmsInboundGatewaySpec<S> inboundGateway(AbstractMessageListenerContainer listenerContainer) {
+		return new JmsInboundGatewaySpec<S>(listenerContainer);
+	}
+
+	public static JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<DefaultMessageListenerContainer> inboundGateway(ConnectionFactory connectionFactory) {
+		return inboundGateway(connectionFactory, DefaultMessageListenerContainer.class);
+	}
+
+	public static <C extends AbstractMessageListenerContainer>
+	JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<C> inboundGateway(ConnectionFactory connectionFactory,
+			Class<C> containerClass) {
+		try {
+			JmsListenerContainerSpec<C> spec = new JmsListenerContainerSpec<C>(containerClass)
+					.connectionFactory(connectionFactory);
+			return new JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<C>(spec);
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+	}
+
 }

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java

@@ -0,0 +1,75 @@
+/*
+ * 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.jms;
+
+import javax.jms.ConnectionFactory;
+
+import org.springframework.integration.dsl.core.IntegrationComponentSpec;
+import org.springframework.jms.support.destination.DestinationResolver;
+import org.springframework.jms.support.destination.JmsDestinationAccessor;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class JmsDestinationAccessorSpec<S extends JmsDestinationAccessorSpec<S, A>, A extends JmsDestinationAccessor>
+		extends IntegrationComponentSpec<S, A> {
+
+	protected JmsDestinationAccessorSpec(A accessor) {
+		this.target = accessor;
+	}
+
+	S connectionFactory(ConnectionFactory connectionFactory) {
+		this.target.setConnectionFactory(connectionFactory);
+		return _this();
+	}
+
+	public S destinationResolver(DestinationResolver destinationResolver) {
+		this.target.setDestinationResolver(destinationResolver);
+		return _this();
+	}
+
+	public S pubSubDomain(boolean pubSubDomain) {
+		target.setPubSubDomain(pubSubDomain);
+		return _this();
+	}
+
+	/**
+	 * @param sessionAcknowledgeMode the acknowledgement mode constant
+	 * @return the current {@link org.springframework.integration.dsl.channel.MessageChannelSpec}
+	 * @see javax.jms.Session#AUTO_ACKNOWLEDGE etc.
+	 */
+	public S sessionAcknowledgeMode(int sessionAcknowledgeMode) {
+		this.target.setSessionAcknowledgeMode(sessionAcknowledgeMode);
+		return _this();
+	}
+
+	public S sessionAcknowledgeModeName(String constantName) {
+		target.setSessionAcknowledgeModeName(constantName);
+		return _this();
+	}
+
+	public S sessionTransacted(boolean sessionTransacted) {
+		this.target.setSessionTransacted(sessionTransacted);
+		return _this();
+	}
+
+	@Override
+	protected A doGet() {
+		throw new UnsupportedOperationException();
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGateway.java

@@ -0,0 +1,128 @@
+/*
+ * 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.jms;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.ApplicationContext;
+import org.springframework.integration.context.OrderlyShutdownCapable;
+import org.springframework.integration.gateway.MessagingGatewaySupport;
+import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
+import org.springframework.integration.jms.JmsMessageDrivenEndpoint;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Artem Bilan
+ */
+public class JmsInboundGateway extends MessagingGatewaySupport implements
+		DisposableBean, OrderlyShutdownCapable {
+
+	private final JmsMessageDrivenEndpoint endpoint;
+
+	private final ChannelPublishingJmsMessageListener listener;
+
+	public JmsInboundGateway(AbstractMessageListenerContainer listenerContainer,
+			ChannelPublishingJmsMessageListener listener) {
+		this.endpoint = new JmsMessageDrivenEndpoint(listenerContainer, listener);
+		this.listener = listener;
+
+	}
+
+	@Override
+	public void setRequestChannel(MessageChannel requestChannel) {
+		this.listener.setRequestChannel(requestChannel);
+	}
+
+	@Override
+	public void setReplyChannel(MessageChannel replyChannel) {
+		this.listener.setReplyChannel(replyChannel);
+	}
+
+	@Override
+	public void setErrorChannel(MessageChannel errorChannel) {
+		this.listener.setErrorChannel(errorChannel);
+	}
+
+	@Override
+	public void setRequestTimeout(long requestTimeout) {
+		this.listener.setRequestTimeout(requestTimeout);
+	}
+
+	@Override
+	public void setReplyTimeout(long replyTimeout) {
+		this.listener.setReplyTimeout(replyTimeout);
+	}
+
+	@Override
+	public void setShouldTrack(boolean shouldTrack) {
+		this.listener.setShouldTrack(shouldTrack);
+	}
+
+	@Override
+	public String getComponentType() {
+		return "jms:message-driven-channel-adapter";
+	}
+
+	@Override
+	public void setComponentName(String componentName) {
+		super.setComponentName(componentName);
+		this.endpoint.setComponentName(getComponentName());
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		super.setApplicationContext(applicationContext);
+		this.endpoint.setApplicationContext(applicationContext);
+		this.listener.setBeanFactory(applicationContext);
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		this.endpoint.afterPropertiesSet();
+	}
+
+	ChannelPublishingJmsMessageListener getListener() {
+		return this.listener;
+	}
+
+	@Override
+	protected void doStart() {
+		this.endpoint.start();
+	}
+
+	@Override
+	protected void doStop() {
+		this.endpoint.stop();
+	}
+
+	@Override
+	public void destroy() throws Exception {
+		this.endpoint.destroy();
+	}
+
+	@Override
+	public int beforeShutdown() {
+		return this.endpoint.beforeShutdown();
+	}
+
+	@Override
+	public int afterShutdown() {
+		return this.endpoint.afterShutdown();
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java

@@ -0,0 +1,136 @@
+/*
+ * 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.jms;
+
+import javax.jms.Destination;
+
+import org.springframework.integration.dsl.core.MessagingGatewaySpec;
+import org.springframework.integration.dsl.support.ComponentConfigurer;
+import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
+import org.springframework.integration.jms.JmsHeaderMapper;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.support.converter.MessageConverter;
+import org.springframework.jms.support.destination.DestinationResolver;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
+		extends MessagingGatewaySpec<S, JmsInboundGateway> {
+
+	JmsInboundGatewaySpec(AbstractMessageListenerContainer listenerContainer) {
+		super(new JmsInboundGateway(listenerContainer, new ChannelPublishingJmsMessageListener()));
+		this.target.getListener().setExpectReply(true);
+	}
+
+	public S defaultReplyDestination(Destination defaultReplyDestination) {
+		this.target.getListener().setDefaultReplyDestination(defaultReplyDestination);
+		return _this();
+	}
+
+	public S defaultReplyQueueName(String destinationName) {
+		this.target.getListener().setDefaultReplyQueueName(destinationName);
+		return _this();
+	}
+
+	public S defaultReplyTopicName(String destinationName) {
+		this.target.getListener().setDefaultReplyTopicName(destinationName);
+		return _this();
+	}
+
+	public S replyTimeToLive(long replyTimeToLive) {
+		this.target.getListener().setReplyTimeToLive(replyTimeToLive);
+		return _this();
+	}
+
+	public S replyPriority(int replyPriority) {
+		this.target.getListener().setReplyPriority(replyPriority);
+		return _this();
+	}
+
+	public S replyDeliveryPersistent(boolean replyDeliveryPersistent) {
+		this.target.getListener().setReplyDeliveryPersistent(replyDeliveryPersistent);
+		return _this();
+	}
+
+	public S correlationKey(String correlationKey) {
+		this.target.getListener().setCorrelationKey(correlationKey);
+		return _this();
+	}
+
+	public S explicitQosEnabledForReplies(boolean explicitQosEnabledForReplies) {
+		this.target.getListener().setExplicitQosEnabledForReplies(explicitQosEnabledForReplies);
+		return _this();
+	}
+
+	public S destinationResolver(DestinationResolver destinationResolver) {
+		this.target.getListener().setDestinationResolver(destinationResolver);
+		return _this();
+	}
+
+	public S jmsMessageConverter(MessageConverter messageConverter) {
+		this.target.getListener().setMessageConverter(messageConverter);
+		return _this();
+	}
+
+	public S setHeaderMapper(JmsHeaderMapper headerMapper) {
+		this.target.getListener().setHeaderMapper(headerMapper);
+		return _this();
+	}
+
+	public S extractRequestPayload(boolean extractRequestPayload) {
+		this.target.getListener().setExtractRequestPayload(extractRequestPayload);
+		return _this();
+	}
+
+	public S extractReplyPayload(boolean extractReplyPayload) {
+		this.target.getListener().setExtractReplyPayload(extractReplyPayload);
+		return _this();
+	}
+
+	public static class JmsInboundGatewayListenerContainerSpec<C extends AbstractMessageListenerContainer> extends
+			JmsInboundGatewaySpec<JmsInboundGatewayListenerContainerSpec<C>> {
+
+		private final JmsListenerContainerSpec<C> spec;
+
+		JmsInboundGatewayListenerContainerSpec(JmsListenerContainerSpec<C> spec) {
+			super(spec.get());
+			this.spec = spec;
+			this.spec.get().setAutoStartup(false);
+		}
+
+		public JmsInboundGatewayListenerContainerSpec<C> destination(Destination destination) {
+			spec.destination(destination);
+			return _this();
+		}
+
+		public JmsInboundGatewayListenerContainerSpec<C> destination(String destinationName) {
+			spec.destination(destinationName);
+			return _this();
+		}
+
+		public JmsInboundGatewayListenerContainerSpec<C> configureListenerContainer
+				(ComponentConfigurer<JmsListenerContainerSpec<C>> configurer) {
+			Assert.notNull(configurer);
+			configurer.configure(this.spec);
+			return _this();
+		}
+
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsListenerContainerSpec.java

@@ -0,0 +1,85 @@
+/*
+ * 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.jms;
+
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.util.ErrorHandler;
+
+/**
+ * @author Artem Bilan
+ */
+public class JmsListenerContainerSpec<C extends AbstractMessageListenerContainer>
+		extends JmsDestinationAccessorSpec<JmsListenerContainerSpec<C>, C> {
+
+	JmsListenerContainerSpec(Class<C> aClass) throws Exception {
+		super(aClass.newInstance());
+	}
+
+	JmsListenerContainerSpec<C> destination(Destination destination) {
+		target.setDestination(destination);
+		return _this();
+	}
+
+	JmsListenerContainerSpec<C> destination(String destinationName) {
+		target.setDestinationName(destinationName);
+		return _this();
+	}
+
+	public JmsListenerContainerSpec<C> messageSelector(String messageSelector) {
+		target.setMessageSelector(messageSelector);
+		return _this();
+	}
+
+	public JmsListenerContainerSpec<C> subscriptionDurable(boolean subscriptionDurable) {
+		target.setSubscriptionDurable(subscriptionDurable);
+		return _this();
+	}
+
+	public JmsListenerContainerSpec<C> durableSubscriptionName(String durableSubscriptionName) {
+		target.setDurableSubscriptionName(durableSubscriptionName);
+		return _this();
+	}
+
+	public JmsListenerContainerSpec<C> exceptionListener(ExceptionListener exceptionListener) {
+		target.setExceptionListener(exceptionListener);
+		return _this();
+	}
+
+	public JmsListenerContainerSpec<C> errorHandler(ErrorHandler errorHandler) {
+		target.setErrorHandler(errorHandler);
+		return _this();
+	}
+
+	public JmsListenerContainerSpec<C> exposeListenerSession(boolean exposeListenerSession) {
+		target.setExposeListenerSession(exposeListenerSession);
+		return _this();
+	}
+
+	public JmsListenerContainerSpec<C> acceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) {
+		target.setAcceptMessagesWhileStopping(acceptMessagesWhileStopping);
+		return _this();
+	}
+
+	public JmsListenerContainerSpec<C> clientId(String clientId) {
+		target.setClientId(clientId);
+		return _this();
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsTemplateSpec.java

@@ -16,35 +16,16 @@
 
 package org.springframework.integration.dsl.jms;
 
-import javax.jms.ConnectionFactory;
-
-import org.springframework.integration.dsl.core.IntegrationComponentSpec;
 import org.springframework.integration.jms.DynamicJmsTemplate;
 import org.springframework.jms.support.converter.MessageConverter;
-import org.springframework.jms.support.destination.DestinationResolver;
 
 /**
  * @author Artem Bilan
  */
-public class JmsTemplateSpec extends IntegrationComponentSpec<JmsTemplateSpec, DynamicJmsTemplate> {
+public class JmsTemplateSpec extends JmsDestinationAccessorSpec<JmsTemplateSpec, DynamicJmsTemplate> {
 
-	public JmsTemplateSpec() {
-		this.target = new DynamicJmsTemplate();
-	}
-
-	JmsTemplateSpec connectionFactory(ConnectionFactory connectionFactory) {
-		this.target.setConnectionFactory(connectionFactory);
-		return _this();
-	}
-
-	public JmsTemplateSpec destinationResolver(DestinationResolver destinationResolver) {
-		this.target.setDestinationResolver(destinationResolver);
-		return _this();
-	}
-
-	public JmsTemplateSpec pubSubDomain(boolean pubSubDomain) {
-		this.target.setPubSubDomain(pubSubDomain);
-		return _this();
+	JmsTemplateSpec() {
+		super(new DynamicJmsTemplate());
 	}
 
 	public JmsTemplateSpec jmsMessageConverter(MessageConverter messageConverter) {
@@ -77,24 +58,4 @@ public class JmsTemplateSpec extends IntegrationComponentSpec<JmsTemplateSpec, D
 		return _this();
 	}
 
-	/**
-	 * @param sessionAcknowledgeMode the acknowledgement mode constant
-	 * @return the current {@link org.springframework.integration.dsl.channel.MessageChannelSpec}
-	 * @see javax.jms.Session#AUTO_ACKNOWLEDGE etc.
-	 */
-	public JmsTemplateSpec sessionAcknowledgeMode(int sessionAcknowledgeMode) {
-		this.target.setSessionAcknowledgeMode(sessionAcknowledgeMode);
-		return _this();
-	}
-
-	public JmsTemplateSpec sessionTransacted(boolean sessionTransacted) {
-		this.target.setSessionTransacted(sessionTransacted);
-		return _this();
-	}
-
-	@Override
-	protected DynamicJmsTemplate doGet() {
-		throw new UnsupportedOperationException();
-	}
-
 }

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

@@ -149,8 +149,6 @@ public class IntegrationFlowTests {
 
 	private static MongodExecutable mongodExe;
 
-	private static MongodProcess mongod;
-
 	@Autowired
 	private ListableBeanFactory beanFactory;
 
@@ -321,12 +319,11 @@ public class IntegrationFlowTests {
 						.version(Version.Main.PRODUCTION)
 						.net(new Net(mongoPort, Network.localhostIsIPv6()))
 						.build());
-		mongod = mongodExe.start();
+		mongodExe.start();
 	}
 
 	@AfterClass
 	public static void tearDown() {
-		mongod.stop();
 		mongodExe.stop();
 	}
 
@@ -862,6 +859,8 @@ public class IntegrationFlowTests {
 		receive = this.priorityReplyChannel.receive(2000);
 		assertNotNull(receive);
 		assertEquals("none", receive.getPayload());
+
+		this.controlBus.send("@priorityChannelBridge.stop()");
 	}
 
 	@Test
@@ -951,15 +950,32 @@ public class IntegrationFlowTests {
 	private PollableChannel jmsOutboundInboundReplyChannel;
 
 	@Test
-	public void testJmsOutboundInboundFlow() throws Exception {
-		this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello through the amqp").build());
+	public void testJmsOutboundInboundFlow() {
+		this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello through the jms").build());
 
 		Message<?> receive = this.jmsOutboundInboundReplyChannel.receive(5000);
 
 		assertNotNull(receive);
-		assertEquals("HELLO THROUGH THE AMQP", receive.getPayload());
+		assertEquals("HELLO THROUGH THE JMS", receive.getPayload());
 	}
 
+	@Autowired
+	@Qualifier("jmsOutboundGatewayChannel")
+	private MessageChannel jmsOutboundGatewayChannel;
+
+	@Test
+	public void testJmsPipelineFlow() {
+		PollableChannel replyChannel = new QueueChannel();
+		Message<String> message = MessageBuilder.withPayload("hello through the jms pipeline")
+				.setReplyChannel(replyChannel)
+				.build();
+		this.jmsOutboundGatewayChannel.send(message);
+
+		Message<?> receive = replyChannel.receive(5000);
+
+		assertNotNull(receive);
+		assertEquals("HELLO THROUGH THE JMS PIPELINE", receive.getPayload());
+	}
 
 	@MessagingGateway(defaultRequestChannel = "controlBus")
 	private static interface ControlBusGateway {
@@ -1039,6 +1055,22 @@ public class IntegrationFlowTests {
 					.get();
 		}
 
+		@Bean
+		public IntegrationFlow jmsOutboundGatewayFlow() {
+			return IntegrationFlows.from("jmsOutboundGatewayChannel")
+					.handle(Jms.outboundGateway(this.jmsConnectionFactory)
+							.replyContainer()
+							.requestDestination("jmsPipelineTest"))
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow jmsInboundGatewayFlow() {
+			return IntegrationFlows.from(Jms.inboundGateway(this.jmsConnectionFactory)
+					.destination("jmsPipelineTest"))
+					.<String, String>transform(String::toUpperCase)
+					.get();
+		}
 	}
 
 	@Configuration
@@ -1098,7 +1130,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
 			return IntegrationFlows.from(MessageChannels.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup"))
-					.bridge(s -> s.poller(Pollers.fixedDelay(1000, 5000)))
+					.bridge(s -> s.poller(Pollers.fixedDelay(1000, 5000)).id("priorityChannelBridge"))
 					.channel(MessageChannels.queue("priorityReplyChannel"))
 					.get();
 		}

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

@@ -4,5 +4,5 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=%d %c{1} [%t] : %m%n
 
-log4j.category.org.springframework.integration=WARN
+log4j.category.org.springframework=WARN
 log4j.category.org.springframework.integration.dsl=INFO