From a35db31e6152abcb79f631491d85a80c5940f7c9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 12 Jun 2014 18:16:43 +0300 Subject: [PATCH] DSL: Add `Jms.inboundGateway()` infrastructure --- .../dsl/IntegrationFlowBuilder.java | 3 +- .../integration/dsl/jms/Jms.java | 23 +++ .../dsl/jms/JmsDestinationAccessorSpec.java | 75 ++++++++++ .../dsl/jms/JmsInboundGateway.java | 128 +++++++++++++++++ .../dsl/jms/JmsInboundGatewaySpec.java | 136 ++++++++++++++++++ .../dsl/jms/JmsListenerContainerSpec.java | 85 +++++++++++ .../integration/dsl/jms/JmsTemplateSpec.java | 45 +----- .../dsl/test/IntegrationFlowTests.java | 48 +++++-- .../src/test/resources/log4j.properties | 2 +- 9 files changed, 493 insertions(+), 52 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGateway.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsListenerContainerSpec.java diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index 3509be7..b525a9d 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/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) { 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 3db3179..01c01e7 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 @@ -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 > JmsInboundGatewaySpec inboundGateway(AbstractMessageListenerContainer listenerContainer) { + return new JmsInboundGatewaySpec(listenerContainer); + } + + public static JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec inboundGateway(ConnectionFactory connectionFactory) { + return inboundGateway(connectionFactory, DefaultMessageListenerContainer.class); + } + + public static + JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec inboundGateway(ConnectionFactory connectionFactory, + Class containerClass) { + try { + JmsListenerContainerSpec spec = new JmsListenerContainerSpec(containerClass) + .connectionFactory(connectionFactory); + return new JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec(spec); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java new file mode 100644 index 0000000..5fea5b9 --- /dev/null +++ b/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, A extends JmsDestinationAccessor> + extends IntegrationComponentSpec { + + 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(); + } + +} 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 new file mode 100644 index 0000000..b357ffc --- /dev/null +++ b/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(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java new file mode 100644 index 0000000..0166f33 --- /dev/null +++ b/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> + extends MessagingGatewaySpec { + + 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 extends + JmsInboundGatewaySpec> { + + private final JmsListenerContainerSpec spec; + + JmsInboundGatewayListenerContainerSpec(JmsListenerContainerSpec spec) { + super(spec.get()); + this.spec = spec; + this.spec.get().setAutoStartup(false); + } + + public JmsInboundGatewayListenerContainerSpec destination(Destination destination) { + spec.destination(destination); + return _this(); + } + + public JmsInboundGatewayListenerContainerSpec destination(String destinationName) { + spec.destination(destinationName); + return _this(); + } + + public JmsInboundGatewayListenerContainerSpec configureListenerContainer + (ComponentConfigurer> configurer) { + Assert.notNull(configurer); + configurer.configure(this.spec); + return _this(); + } + + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsListenerContainerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsListenerContainerSpec.java new file mode 100644 index 0000000..b74f0cb --- /dev/null +++ b/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 + extends JmsDestinationAccessorSpec, C> { + + JmsListenerContainerSpec(Class aClass) throws Exception { + super(aClass.newInstance()); + } + + JmsListenerContainerSpec destination(Destination destination) { + target.setDestination(destination); + return _this(); + } + + JmsListenerContainerSpec destination(String destinationName) { + target.setDestinationName(destinationName); + return _this(); + } + + public JmsListenerContainerSpec messageSelector(String messageSelector) { + target.setMessageSelector(messageSelector); + return _this(); + } + + public JmsListenerContainerSpec subscriptionDurable(boolean subscriptionDurable) { + target.setSubscriptionDurable(subscriptionDurable); + return _this(); + } + + public JmsListenerContainerSpec durableSubscriptionName(String durableSubscriptionName) { + target.setDurableSubscriptionName(durableSubscriptionName); + return _this(); + } + + public JmsListenerContainerSpec exceptionListener(ExceptionListener exceptionListener) { + target.setExceptionListener(exceptionListener); + return _this(); + } + + public JmsListenerContainerSpec errorHandler(ErrorHandler errorHandler) { + target.setErrorHandler(errorHandler); + return _this(); + } + + public JmsListenerContainerSpec exposeListenerSession(boolean exposeListenerSession) { + target.setExposeListenerSession(exposeListenerSession); + return _this(); + } + + public JmsListenerContainerSpec acceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) { + target.setAcceptMessagesWhileStopping(acceptMessagesWhileStopping); + return _this(); + } + + public JmsListenerContainerSpec clientId(String clientId) { + target.setClientId(clientId); + return _this(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsTemplateSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsTemplateSpec.java index 6af83af..f2220c4 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsTemplateSpec.java +++ b/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 { +public class JmsTemplateSpec extends JmsDestinationAccessorSpec { - 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 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 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")) + .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(); } diff --git a/spring-integration-java-dsl/src/test/resources/log4j.properties b/spring-integration-java-dsl/src/test/resources/log4j.properties index 8b5a4f8..80f6cef 100644 --- a/spring-integration-java-dsl/src/test/resources/log4j.properties +++ b/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