DSL: Add Jms.messageDriverAdapter

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

@@ -27,16 +27,19 @@ import org.springframework.jms.listener.DefaultMessageListenerContainer;
  */
 public abstract class Jms {
 
-	public static <S extends JmsPollableMessageChannelSpec<S>> JmsPollableMessageChannelSpec<S>	pollableChannel(ConnectionFactory connectionFactory) {
+	public static <S extends JmsPollableMessageChannelSpec<S>> JmsPollableMessageChannelSpec<S>
+	pollableChannel(ConnectionFactory connectionFactory) {
 		return pollableChannel(null, connectionFactory);
 	}
 
-	public static <S extends JmsPollableMessageChannelSpec<S>> JmsPollableMessageChannelSpec<S> pollableChannel(String id,
+	public static
+	<S extends JmsPollableMessageChannelSpec<S>> JmsPollableMessageChannelSpec<S> pollableChannel(String id,
 			ConnectionFactory connectionFactory) {
 		return new JmsPollableMessageChannelSpec<S>(connectionFactory).id(id);
 	}
 
-	public static <S extends JmsMessageChannelSpec<S>> JmsMessageChannelSpec<S> channel(ConnectionFactory connectionFactory) {
+	public static
+	<S extends JmsMessageChannelSpec<S>> JmsMessageChannelSpec<S> channel(ConnectionFactory connectionFactory) {
 		return channel(null, connectionFactory);
 	}
 
@@ -54,19 +57,23 @@ public abstract class Jms {
 		return new JmsPublishSubscribeMessageChannelSpec(connectionFactory).id(id);
 	}
 
-	public static <S extends JmsOutboundChannelAdapterSpec<S>> JmsOutboundChannelAdapterSpec<S> outboundAdapter(JmsTemplate jmsTemplate) {
+	public static <S extends JmsOutboundChannelAdapterSpec<S>> JmsOutboundChannelAdapterSpec<S>
+	outboundAdapter(JmsTemplate jmsTemplate) {
 		return new JmsOutboundChannelAdapterSpec<S>(jmsTemplate);
 	}
 
-	public static JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware outboundAdapter(ConnectionFactory connectionFactory) {
+	public static JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware
+	outboundAdapter(ConnectionFactory connectionFactory) {
 		return new JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware(connectionFactory);
 	}
 
-	public static <S extends JmsInboundChannelAdapterSpec<S>> JmsInboundChannelAdapterSpec<S> inboundAdapter(JmsTemplate jmsTemplate) {
+	public static <S extends JmsInboundChannelAdapterSpec<S>> JmsInboundChannelAdapterSpec<S>
+	inboundAdapter(JmsTemplate jmsTemplate) {
 		return new JmsInboundChannelAdapterSpec<S>(jmsTemplate);
 	}
 
-	public static JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware inboundAdapter(ConnectionFactory connectionFactory) {
+	public static JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware
+	inboundAdapter(ConnectionFactory connectionFactory) {
 		return new JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware(connectionFactory);
 	}
 
@@ -74,11 +81,13 @@ public abstract class Jms {
 		return new JmsOutboundGatewaySpec(connectionFactory);
 	}
 
-	public static <S extends JmsInboundGatewaySpec<S>> JmsInboundGatewaySpec<S> inboundGateway(AbstractMessageListenerContainer listenerContainer) {
+	public static <S extends JmsInboundGatewaySpec<S>> JmsInboundGatewaySpec<S>
+	inboundGateway(AbstractMessageListenerContainer listenerContainer) {
 		return new JmsInboundGatewaySpec<S>(listenerContainer);
 	}
 
-	public static JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<DefaultMessageListenerContainer> inboundGateway(ConnectionFactory connectionFactory) {
+	public static JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<DefaultMessageListenerContainer>
+	inboundGateway(ConnectionFactory connectionFactory) {
 		return inboundGateway(connectionFactory, DefaultMessageListenerContainer.class);
 	}
 
@@ -95,4 +104,28 @@ public abstract class Jms {
 		}
 	}
 
+	public static <S extends JmsMessageDrivenChannelAdapterSpec<S>> JmsMessageDrivenChannelAdapterSpec<S>
+	messageDriverChannelAdapter(AbstractMessageListenerContainer listenerContainer) {
+		return new JmsMessageDrivenChannelAdapterSpec<S>(listenerContainer);
+	}
+
+	public static
+	JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<DefaultMessageListenerContainer>
+	messageDriverAdapter(ConnectionFactory connectionFactory) {
+		return messageDriverAdapter(connectionFactory, DefaultMessageListenerContainer.class);
+	}
+
+	public static <C extends AbstractMessageListenerContainer>
+	JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<C>
+	messageDriverAdapter(ConnectionFactory connectionFactory, Class<C> containerClass) {
+		try {
+			JmsListenerContainerSpec<C> spec = new JmsListenerContainerSpec<C>(containerClass)
+					.connectionFactory(connectionFactory);
+			return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<C>(spec);
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+	}
+
 }

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

@@ -40,7 +40,6 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements
 			ChannelPublishingJmsMessageListener listener) {
 		this.endpoint = new JmsMessageDrivenEndpoint(listenerContainer, listener);
 		this.listener = listener;
-
 	}
 
 	@Override
@@ -75,7 +74,7 @@ public class JmsInboundGateway extends MessagingGatewaySupport implements
 
 	@Override
 	public String getComponentType() {
-		return "jms:message-driven-channel-adapter";
+		return "jms:inbound-gateway";
 	}
 
 	@Override

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

@@ -0,0 +1,117 @@
+/*
+ * 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.endpoint.MessageProducerSupport;
+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 JmsMessageDrivenChannelAdapter extends MessageProducerSupport implements
+		DisposableBean, OrderlyShutdownCapable {
+
+	private final JmsMessageDrivenEndpoint endpoint;
+
+	private final ChannelPublishingJmsMessageListener listener;
+
+	public JmsMessageDrivenChannelAdapter(AbstractMessageListenerContainer listenerContainer,
+			ChannelPublishingJmsMessageListener listener) {
+		this.endpoint = new JmsMessageDrivenEndpoint(listenerContainer, listener);
+		this.listener = listener;
+	}
+
+	@Override
+	public void setOutputChannel(MessageChannel requestChannel) {
+		this.listener.setRequestChannel(requestChannel);
+	}
+
+	@Override
+	public void setErrorChannel(MessageChannel errorChannel) {
+		this.listener.setErrorChannel(errorChannel);
+	}
+
+	@Override
+	public void setSendTimeout(long requestTimeout) {
+		this.listener.setRequestTimeout(requestTimeout);
+	}
+
+	@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() {
+		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/JmsMessageDrivenChannelAdapterSpec.java

@@ -0,0 +1,86 @@
+/*
+ * 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.MessagingProducerSpec;
+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.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChannelAdapterSpec<S>>
+		extends MessagingProducerSpec<S, JmsMessageDrivenChannelAdapter> {
+
+	JmsMessageDrivenChannelAdapterSpec(AbstractMessageListenerContainer listenerContainer) {
+		super(new JmsMessageDrivenChannelAdapter(listenerContainer, new ChannelPublishingJmsMessageListener()));
+		this.target.getListener().setExpectReply(false);
+	}
+
+	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 extractPayload(boolean extractRequestPayload) {
+		this.target.getListener().setExtractRequestPayload(extractRequestPayload);
+		return _this();
+	}
+
+
+	public static class JmsMessageDrivenChannelAdapterListenerContainerSpec<C extends AbstractMessageListenerContainer> extends
+			JmsMessageDrivenChannelAdapterSpec<JmsMessageDrivenChannelAdapterListenerContainerSpec<C>> {
+
+		private final JmsListenerContainerSpec<C> spec;
+
+		JmsMessageDrivenChannelAdapterListenerContainerSpec(JmsListenerContainerSpec<C> spec) {
+			super(spec.get());
+			this.spec = spec;
+			this.spec.get().setAutoStartup(false);
+		}
+
+		public JmsMessageDrivenChannelAdapterListenerContainerSpec<C> destination(Destination destination) {
+			spec.destination(destination);
+			return _this();
+		}
+
+		public JmsMessageDrivenChannelAdapterListenerContainerSpec<C> destination(String destinationName) {
+			spec.destination(destinationName);
+			return _this();
+		}
+
+		public JmsMessageDrivenChannelAdapterListenerContainerSpec<C> configureListenerContainer
+				(ComponentConfigurer<JmsListenerContainerSpec<C>> configurer) {
+			Assert.notNull(configurer);
+			configurer.configure(this.spec);
+			return _this();
+		}
+
+	}
+
+}

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

@@ -35,7 +35,6 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import com.mongodb.MongoClient;
 import de.flapdoodle.embed.mongo.MongodExecutable;
-import de.flapdoodle.embed.mongo.MongodProcess;
 import de.flapdoodle.embed.mongo.MongodStarter;
 import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
 import de.flapdoodle.embed.mongo.config.Net;
@@ -124,6 +123,7 @@ 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.simp.SimpMessageHeaderAccessor;
 import org.springframework.messaging.support.ChannelInterceptorAdapter;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
@@ -951,12 +951,23 @@ public class IntegrationFlowTests {
 
 	@Test
 	public void testJmsOutboundInboundFlow() {
-		this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello through the jms").build());
+		this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS")
+				.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsInbound")
+				.build());
 
 		Message<?> receive = this.jmsOutboundInboundReplyChannel.receive(5000);
 
 		assertNotNull(receive);
 		assertEquals("HELLO THROUGH THE JMS", receive.getPayload());
+
+		this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS")
+				.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsMessageDriver")
+				.build());
+
+		receive = this.jmsOutboundInboundReplyChannel.receive(5000);
+
+		assertNotNull(receive);
+		assertEquals("hello through the jms", receive.getPayload());
 	}
 
 	@Autowired
@@ -968,6 +979,7 @@ public class IntegrationFlowTests {
 		PollableChannel replyChannel = new QueueChannel();
 		Message<String> message = MessageBuilder.withPayload("hello through the jms pipeline")
 				.setReplyChannel(replyChannel)
+				.setHeader("destination", "jmsPipelineTest")
 				.build();
 		this.jmsOutboundGatewayChannel.send(message);
 
@@ -1043,15 +1055,33 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow jmsOutboundFlow() {
 			return IntegrationFlows.from("jmsOutboundInboundChannel")
-					.handle(Jms.outboundAdapter(this.jmsConnectionFactory).destination("jmsOutboundInboundChannel"))
+					.handle(Jms.outboundAdapter(this.jmsConnectionFactory)
+							.destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER))
 					.get();
 		}
 
+		@Bean
+		public MessageChannel jmsOutboundInboundReplyChannel() {
+			return MessageChannels.queue().get();
+		}
+
 		@Bean
 		public IntegrationFlow jmsInboundFlow() {
-			return IntegrationFlows.from(Jms.inboundAdapter(this.jmsConnectionFactory).destination("jmsOutboundInboundChannel"))
+			return IntegrationFlows
+					.from(Jms.inboundAdapter(this.jmsConnectionFactory)
+							.destination("jmsInbound"))
 					.<String, String>transform(String::toUpperCase)
-					.channel(MessageChannels.queue("jmsOutboundInboundReplyChannel"))
+					.channel(this.jmsOutboundInboundReplyChannel())
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow jmsMessageDriverFlow() {
+			return IntegrationFlows
+					.from(Jms.messageDriverAdapter(this.jmsConnectionFactory)
+							.destination("jmsMessageDriver"))
+					.<String, String>transform(String::toLowerCase)
+					.channel(this.jmsOutboundInboundReplyChannel())
 					.get();
 		}