diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index 41a93ec..63d52af 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -81,7 +81,7 @@ dependencies { testCompile "de.flapdoodle.embed:de.flapdoodle.embed.mongo:$embedMongoVersion" testCompile "org.springframework.boot:spring-boot-autoconfigure:$springBootVersion" - testRuntime "org.slf4j:jcl-over-slf4j:$slf4jVersion" +// testRuntime "org.slf4j:jcl-over-slf4j:$slf4jVersion" testRuntime "org.slf4j:slf4j-log4j12:$slf4jVersion" testRuntime("org.apache.activemq:activemq-broker:$activeMqVersion") testRuntime("org.apache.activemq:activemq-kahadb-store:$activeMqVersion") { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java index 01c01e7..e1e9353 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java +++ b/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 > JmsPollableMessageChannelSpec pollableChannel(ConnectionFactory connectionFactory) { + public static > JmsPollableMessageChannelSpec + pollableChannel(ConnectionFactory connectionFactory) { return pollableChannel(null, connectionFactory); } - public static > JmsPollableMessageChannelSpec pollableChannel(String id, + public static + > JmsPollableMessageChannelSpec pollableChannel(String id, ConnectionFactory connectionFactory) { return new JmsPollableMessageChannelSpec(connectionFactory).id(id); } - public static > JmsMessageChannelSpec channel(ConnectionFactory connectionFactory) { + public static + > JmsMessageChannelSpec channel(ConnectionFactory connectionFactory) { return channel(null, connectionFactory); } @@ -54,19 +57,23 @@ public abstract class Jms { return new JmsPublishSubscribeMessageChannelSpec(connectionFactory).id(id); } - public static > JmsOutboundChannelAdapterSpec outboundAdapter(JmsTemplate jmsTemplate) { + public static > JmsOutboundChannelAdapterSpec + outboundAdapter(JmsTemplate jmsTemplate) { return new JmsOutboundChannelAdapterSpec(jmsTemplate); } - public static JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware outboundAdapter(ConnectionFactory connectionFactory) { + public static JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware + outboundAdapter(ConnectionFactory connectionFactory) { return new JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware(connectionFactory); } - public static > JmsInboundChannelAdapterSpec inboundAdapter(JmsTemplate jmsTemplate) { + public static > JmsInboundChannelAdapterSpec + inboundAdapter(JmsTemplate jmsTemplate) { return new JmsInboundChannelAdapterSpec(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 > JmsInboundGatewaySpec inboundGateway(AbstractMessageListenerContainer listenerContainer) { + public static > JmsInboundGatewaySpec + inboundGateway(AbstractMessageListenerContainer listenerContainer) { return new JmsInboundGatewaySpec(listenerContainer); } - public static JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec inboundGateway(ConnectionFactory connectionFactory) { + public static JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec + inboundGateway(ConnectionFactory connectionFactory) { return inboundGateway(connectionFactory, DefaultMessageListenerContainer.class); } @@ -95,4 +104,28 @@ public abstract class Jms { } } + public static > JmsMessageDrivenChannelAdapterSpec + messageDriverChannelAdapter(AbstractMessageListenerContainer listenerContainer) { + return new JmsMessageDrivenChannelAdapterSpec(listenerContainer); + } + + public static + JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec + messageDriverAdapter(ConnectionFactory connectionFactory) { + return messageDriverAdapter(connectionFactory, DefaultMessageListenerContainer.class); + } + + public static + JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec + messageDriverAdapter(ConnectionFactory connectionFactory, Class containerClass) { + try { + JmsListenerContainerSpec spec = new JmsListenerContainerSpec(containerClass) + .connectionFactory(connectionFactory); + return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec(spec); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGateway.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGateway.java index b357ffc..46c0baf 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGateway.java +++ b/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 diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapter.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapter.java new file mode 100644 index 0000000..7fac33f --- /dev/null +++ b/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(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapterSpec.java new file mode 100644 index 0000000..97b83a1 --- /dev/null +++ b/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> + extends MessagingProducerSpec { + + 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 extends + JmsMessageDrivenChannelAdapterSpec> { + + private final JmsListenerContainerSpec spec; + + JmsMessageDrivenChannelAdapterListenerContainerSpec(JmsListenerContainerSpec spec) { + super(spec.get()); + this.spec = spec; + this.spec.get().setAutoStartup(false); + } + + public JmsMessageDrivenChannelAdapterListenerContainerSpec destination(Destination destination) { + spec.destination(destination); + return _this(); + } + + public JmsMessageDrivenChannelAdapterListenerContainerSpec destination(String destinationName) { + spec.destination(destinationName); + return _this(); + } + + public JmsMessageDrivenChannelAdapterListenerContainerSpec configureListenerContainer + (ComponentConfigurer> configurer) { + Assert.notNull(configurer); + configurer.configure(this.spec); + return _this(); + } + + } + +} diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index 0bd0e59..b0a0e7e 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/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 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")) .transform(String::toUpperCase) - .channel(MessageChannels.queue("jmsOutboundInboundReplyChannel")) + .channel(this.jmsOutboundInboundReplyChannel()) + .get(); + } + + @Bean + public IntegrationFlow jmsMessageDriverFlow() { + return IntegrationFlows + .from(Jms.messageDriverAdapter(this.jmsConnectionFactory) + .destination("jmsMessageDriver")) + .transform(String::toLowerCase) + .channel(this.jmsOutboundInboundReplyChannel()) .get(); }