From 3cf85afe1f9f11e56310871da564d3b78e1eee42 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 18 May 2017 14:20:01 -0400 Subject: [PATCH] INT-4278: DSL: AMQP DirectMessageListenerContainer JIRA: https://jira.spring.io/browse/INT-4278 Support configuration of inbound endpoints using `DirectMessageListenerContainer`. **Needs migration guide update** Polishing according PR discussion --- .../AbstractMessageListenerContainerSpec.java | 385 ++++++++++++++++++ .../integration/amqp/dsl/Amqp.java | 112 +++-- .../AmqpInboundChannelAdapterDMLCSpec.java | 42 ++ .../AmqpInboundChannelAdapterSMLCSpec.java | 42 ++ .../dsl/AmqpInboundChannelAdapterSpec.java | 231 +---------- .../amqp/dsl/AmqpInboundGatewayDMLCSpec.java | 47 +++ .../amqp/dsl/AmqpInboundGatewaySMLCSpec.java | 47 +++ .../amqp/dsl/AmqpInboundGatewaySpec.java | 235 +---------- .../DirectMessageListenerContainerSpec.java | 68 ++++ .../SimpleMessageListenerContainerSpec.java | 118 ++++++ .../integration/amqp/dsl/AmqpTests.java | 18 + 11 files changed, 868 insertions(+), 477 deletions(-) create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java create mode 100644 spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java new file mode 100644 index 0000000000..09e2030a9a --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AbstractMessageListenerContainerSpec.java @@ -0,0 +1,385 @@ +/* + * Copyright 2017 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.amqp.dsl; + +import java.util.Map; +import java.util.concurrent.Executor; + +import org.aopalliance.aop.Advice; + +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.core.MessagePostProcessor; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.ListenerContainerIdleEvent; +import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; +import org.springframework.amqp.support.ConditionalExceptionLogger; +import org.springframework.amqp.support.ConsumerTagStrategy; +import org.springframework.integration.dsl.IntegrationComponentSpec; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.TransactionAttribute; +import org.springframework.util.ErrorHandler; +import org.springframework.util.backoff.BackOff; + +/** + * Base class for container specs. + * + * @author Gary Russell + * @author Artem Bilan + * + * @since 5.0 + * + */ +public abstract class AbstractMessageListenerContainerSpec, + C extends AbstractMessageListenerContainer> + extends IntegrationComponentSpec { + + public AbstractMessageListenerContainerSpec(C listenerContainer) { + this.target = listenerContainer; + } + + @Override + public S id(String id) { + return super.id(id); + } + + /** + * @param acknowledgeMode the acknowledgeMode. + * @return the spec. + * @see AbstractMessageListenerContainer#setAcknowledgeMode(AcknowledgeMode) + */ + public S acknowledgeMode(AcknowledgeMode acknowledgeMode) { + this.target.setAcknowledgeMode(acknowledgeMode); + return _this(); + } + + /** + * @param queueName a vararg list of queue names to add. + * @return the spec. + * @see AbstractMessageListenerContainer#addQueueNames(String...) + */ + public S addQueueNames(String... queueName) { + this.target.addQueueNames(queueName); + return _this(); + } + + /** + * @param queues a vararg list of queues to add. + * @return the spec. + * @see AbstractMessageListenerContainer#addQueueNames(String...) + */ + public S addQueues(Queue... queues) { + this.target.addQueues(queues); + return _this(); + } + + /** + * @param errorHandler the errorHandler. + * @return the spec. + * @see AbstractMessageListenerContainer#setErrorHandler(ErrorHandler) + */ + public S errorHandler(ErrorHandler errorHandler) { + this.target.setErrorHandler(errorHandler); + return _this(); + } + + /** + * @param transactional true for transactional channels. + * @return the spec. + * @see AbstractMessageListenerContainer#setChannelTransacted(boolean) + */ + public S channelTransacted(boolean transactional) { + this.target.setChannelTransacted(transactional); + return _this(); + } + + /** + * @param adviceChain the adviceChain. + * @return the spec. + * @see AbstractMessageListenerContainer#setAdviceChain(Advice[]) + */ + public S adviceChain(Advice... adviceChain) { + this.target.setAdviceChain(adviceChain); + return _this(); + } + + /** + * @param recoveryInterval the recoveryInterval + * @return the spec. + * @see AbstractMessageListenerContainer#setRecoveryInterval(long) + */ + public S recoveryInterval(long recoveryInterval) { + this.target.setRecoveryInterval(recoveryInterval); + return _this(); + } + + /** + * @param exclusive true for exclusive. + * @return the spec. + * @see AbstractMessageListenerContainer#setExclusive(boolean) + */ + public S exclusive(boolean exclusive) { + this.target.setExclusive(exclusive); + return _this(); + } + + /** + * @param shutdownTimeout the shutdownTimeout. + * @return the spec. + * @see AbstractMessageListenerContainer#setShutdownTimeout(long) + */ + public S shutdownTimeout(long shutdownTimeout) { + this.target.setShutdownTimeout(shutdownTimeout); + return _this(); + } + + /** + * Configure an {@link Executor} used to invoke the message listener. + * @param taskExecutor the taskExecutor. + * @return the spec. + */ + public S taskExecutor(Executor taskExecutor) { + this.target.setTaskExecutor(taskExecutor); + return _this(); + } + + /** + * @param prefetchCount the prefetchCount. + * @return the spec. + * @see AbstractMessageListenerContainer#setPrefetchCount(int) + */ + public S prefetchCount(int prefetchCount) { + this.target.setPrefetchCount(prefetchCount); + return _this(); + } + + /** + * Configure a {@link PlatformTransactionManager}; used to synchronize the rabbit transaction + * with some other transaction(s). + * @param transactionManager the transactionManager. + * @return the spec. + */ + public S transactionManager(PlatformTransactionManager transactionManager) { + this.target.setTransactionManager(transactionManager); + return _this(); + } + + /** + * @param defaultRequeueRejected the defaultRequeueRejected. + * @return the spec. + * @see AbstractMessageListenerContainer#setDefaultRequeueRejected(boolean) + */ + public S defaultRequeueRejected(boolean defaultRequeueRejected) { + this.target.setDefaultRequeueRejected(defaultRequeueRejected); + return _this(); + } + + /** + * Determine whether or not the container should de-batch batched + * messages (true) or call the listener with the batch (false). Default: true. + * @param deBatchingEnabled the deBatchingEnabled to set. + * @return the spec. + * @see AbstractMessageListenerContainer#setDeBatchingEnabled(boolean) + */ + public S deBatchingEnabled(boolean deBatchingEnabled) { + this.target.setDeBatchingEnabled(deBatchingEnabled); + return _this(); + } + + /** + * Set {@link MessagePostProcessor}s that will be applied after message reception, before + * invoking the {@link MessageListener}. Often used to decompress data. Processors are invoked in order, + * depending on {@code PriorityOrder}, {@code Order} and finally unordered. + * @param afterReceivePostProcessors the post processor. + * @return the spec. + * @see AbstractMessageListenerContainer#setAfterReceivePostProcessors(MessagePostProcessor...) + */ + public S afterReceivePostProcessors(MessagePostProcessor... afterReceivePostProcessors) { + this.target.setAfterReceivePostProcessors(afterReceivePostProcessors); + return _this(); + } + + /** + * Set a qualifier that will prefix the connection factory lookup key; default none. + * @param lookupKeyQualifier the qualifier + * @return the spec. + * @see AbstractMessageListenerContainer#setLookupKeyQualifier(String) + */ + public S lookupKeyQualifier(String lookupKeyQualifier) { + this.target.setLookupKeyQualifier(lookupKeyQualifier); + return _this(); + } + + /** + * Set the implementation of {@link ConsumerTagStrategy} to generate consumer tags. + * By default, the RabbitMQ server generates consumer tags. + * @param consumerTagStrategy the consumerTagStrategy to set. + * @return the spec. + * @see AbstractMessageListenerContainer#setConsumerTagStrategy(ConsumerTagStrategy) + */ + public S consumerTagStrategy(ConsumerTagStrategy consumerTagStrategy) { + this.target.setConsumerTagStrategy(consumerTagStrategy); + return _this(); + } + + /** + * Set consumer arguments. + * @param args the arguments. + * @return the spec. + * @see AbstractMessageListenerContainer#setConsumerArguments(Map) + */ + public S consumerArguments(Map args) { + this.target.setConsumerArguments(args); + return _this(); + } + + /** + * How often to emit {@link ListenerContainerIdleEvent}s in milliseconds. + * @param idleEventInterval the interval. + * @return the spec. + * @see AbstractMessageListenerContainer#setIdleEventInterval(long) + */ + public S idleEventInterval(long idleEventInterval) { + this.target.setIdleEventInterval(idleEventInterval); + return _this(); + } + + /** + * Set the transaction attribute to use when using an external transaction manager. + * @param transactionAttribute the transaction attribute to set + * @return the spec. + * @see AbstractMessageListenerContainer#setTransactionAttribute(TransactionAttribute) + */ + public S transactionAttribute(TransactionAttribute transactionAttribute) { + this.target.setTransactionAttribute(transactionAttribute); + return _this(); + } + + /** + * Specify the {@link BackOff} for interval between recovery attempts. + * The default is 5000 ms, that is, 5 seconds. + * With the {@link BackOff} you can supply the {@code maxAttempts} for recovery before + * the {@code stop()} will be performed. + * @param recoveryBackOff The BackOff to recover. + * @return the spec. + * @see AbstractMessageListenerContainer#setRecoveryBackOff(BackOff) + */ + public S recoveryBackOff(BackOff recoveryBackOff) { + this.target.setRecoveryBackOff(recoveryBackOff); + return _this(); + } + + /** + * Set the {@link MessagePropertiesConverter} for this listener container. + * @param messagePropertiesConverter The properties converter. + * @return the spec. + * @see AbstractMessageListenerContainer#setMessagePropertiesConverter(MessagePropertiesConverter) + */ + public S messagePropertiesConverter(MessagePropertiesConverter messagePropertiesConverter) { + this.target.setMessagePropertiesConverter(messagePropertiesConverter); + return _this(); + } + + /** + * If all of the configured queue(s) are not available on the broker, this setting + * determines whether the condition is fatal. When true, and + * the queues are missing during startup, the context refresh() will fail. + *

When false, the condition is not considered fatal and the container will + * continue to attempt to start the consumers. + * @param missingQueuesFatal the missingQueuesFatal to set. + * @return the spec. + * @see AbstractMessageListenerContainer#setMissingQueuesFatal(boolean) + */ + public S missingQueuesFatal(boolean missingQueuesFatal) { + this.target.setMissingQueuesFatal(missingQueuesFatal); + return _this(); + } + + /** + * Prevent the container from starting if any of the queues defined in the context have + * mismatched arguments (TTL etc). Default false. + * @param mismatchedQueuesFatal true to fail initialization when this condition occurs. + * @return the spec. + * @see AbstractMessageListenerContainer#setMismatchedQueuesFatal(boolean) + */ + public S mismatchedQueuesFatal(boolean mismatchedQueuesFatal) { + this.target.setMismatchedQueuesFatal(mismatchedQueuesFatal); + return _this(); + } + + /** + * Set to true to automatically declare elements (queues, exchanges, bindings) + * in the application context during container start(). + * @param autoDeclare the boolean flag to indicate an declaration operation. + * @return the spec. + * @see AbstractMessageListenerContainer#setAutoDeclare(boolean) + */ + public S autoDeclare(boolean autoDeclare) { + this.target.setAutoDeclare(autoDeclare); + return _this(); + } + + /** + * Set the interval between passive queue declaration attempts in milliseconds. + * @param failedDeclarationRetryInterval the interval, default 5000. + * @return the spec. + * @see AbstractMessageListenerContainer#setFailedDeclarationRetryInterval(long) + */ + public S failedDeclarationRetryInterval(long failedDeclarationRetryInterval) { + this.target.setFailedDeclarationRetryInterval(failedDeclarationRetryInterval); + return _this(); + } + + /** + * Set whether a message with a null messageId is fatal for the consumer + * when using stateful retry. When false, instead of stopping the consumer, + * the message is rejected and not requeued - it will be discarded or routed + * to the dead letter queue, if so configured. Default true. + * @param statefulRetryFatalWithNullMessageId true for fatal. + * @return the spec. + * @see AbstractMessageListenerContainer#setStatefulRetryFatalWithNullMessageId(boolean) + */ + public S statefulRetryFatalWithNullMessageId(boolean statefulRetryFatalWithNullMessageId) { + this.target.setStatefulRetryFatalWithNullMessageId(statefulRetryFatalWithNullMessageId); + return _this(); + } + + /** + * Set a {@link ConditionalExceptionLogger} for logging exclusive consumer failures. The + * default is to log such failures at WARN level. + * @param exclusiveConsumerExceptionLogger the conditional exception logger. + * @return the spec. + * @see AbstractMessageListenerContainer#setExclusiveConsumerExceptionLogger(ConditionalExceptionLogger) + */ + public S exclusiveConsumerExceptionLogger(ConditionalExceptionLogger exclusiveConsumerExceptionLogger) { + this.target.setExclusiveConsumerExceptionLogger(exclusiveConsumerExceptionLogger); + return _this(); + } + + /** + * Set to true to always requeue on transaction rollback with an external TransactionManager. + * @param alwaysRequeueWithTxManagerRollback true to always requeue on rollback. + * @return the spec. + * @see AbstractMessageListenerContainer#setAlwaysRequeueWithTxManagerRollback(boolean) + */ + public S alwaysRequeueWithTxManagerRollback(boolean alwaysRequeueWithTxManagerRollback) { + this.target.setAlwaysRequeueWithTxManagerRollback(alwaysRequeueWithTxManagerRollback); + return _this(); + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java index 9748017c01..cc5f83ceeb 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/Amqp.java @@ -20,12 +20,14 @@ import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.AsyncRabbitTemplate; import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; /** * Factory class for AMQP components. * * @author Artem Bilan + * @author Gary Russell * @since 5.0 */ public final class Amqp { @@ -36,10 +38,10 @@ public final class Amqp { * @param queueNames the queueNames. * @return the AmqpInboundGatewaySpec. */ - public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, String... queueNames) { + public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, String... queueNames) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueueNames(queueNames); - return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer); + return inboundGateway(listenerContainer); } /** @@ -49,11 +51,11 @@ public final class Amqp { * @param queueNames the queueNames. * @return the AmqpInboundGatewaySpec. */ - public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate, + public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate, String... queueNames) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueueNames(queueNames); - return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer, amqpTemplate); + return inboundGateway(listenerContainer, amqpTemplate); } @@ -63,10 +65,10 @@ public final class Amqp { * @param queues the queues. * @return the AmqpInboundGatewaySpec. */ - public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, Queue... queues) { + public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, Queue... queues) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueues(queues); - return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer); + return inboundGateway(listenerContainer); } /** @@ -76,74 +78,126 @@ public final class Amqp { * @param queues the queues. * @return the AmqpInboundGatewaySpec. */ - public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate, + public static AmqpInboundGatewaySMLCSpec inboundGateway(ConnectionFactory connectionFactory, AmqpTemplate amqpTemplate, Queue... queues) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueues(queues); - return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer, amqpTemplate); + return inboundGateway(listenerContainer, amqpTemplate); } /** - * Create an initial {@link AmqpBaseInboundGatewaySpec} + * Create an initial {@link AmqpInboundGatewaySMLCSpec} * with provided {@link SimpleMessageListenerContainer}. * Note: only endpoint options are available from spec. * The {@code listenerContainer} options should be specified - * on the provided {@link SimpleMessageListenerContainer}. + * on the provided {@link SimpleMessageListenerContainer} using + * {@link AmqpInboundGatewaySMLCSpec#configureContainer(java.util.function.Consumer)}. * @param listenerContainer the listenerContainer - * @return the AmqpBaseInboundGatewaySpec. + * @return the AmqpInboundGatewaySMLCSpec. */ - public static AmqpBaseInboundGatewaySpec inboundGateway(SimpleMessageListenerContainer listenerContainer) { - return new AmqpInboundGatewaySpec(listenerContainer); + public static AmqpInboundGatewaySMLCSpec inboundGateway(SimpleMessageListenerContainer listenerContainer) { + return new AmqpInboundGatewaySMLCSpec(listenerContainer); } /** - * Create an initial {@link AmqpBaseInboundGatewaySpec} + * Create an initial {@link AmqpInboundGatewaySMLCSpec} * with provided {@link SimpleMessageListenerContainer}. * Note: only endpoint options are available from spec. * The {@code listenerContainer} options should be specified - * on the provided {@link SimpleMessageListenerContainer}. + * on the provided {@link SimpleMessageListenerContainer}using + * {@link AmqpInboundGatewaySMLCSpec#configureContainer(java.util.function.Consumer)}. * @param listenerContainer the listenerContainer * @param amqpTemplate the {@link AmqpTemplate} to use. - * @return the AmqpBaseInboundGatewaySpec. + * @return the AmqpInboundGatewaySMLCSpec. */ - public static AmqpBaseInboundGatewaySpec inboundGateway(SimpleMessageListenerContainer listenerContainer, + public static AmqpInboundGatewaySMLCSpec inboundGateway(SimpleMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { - return new AmqpInboundGatewaySpec(listenerContainer, amqpTemplate); + return new AmqpInboundGatewaySMLCSpec(listenerContainer, amqpTemplate); } /** - * Create an initial AmqpInboundChannelAdapterSpec. + * Create an initial {@link DirectMessageListenerContainerSpec} + * with provided {@link DirectMessageListenerContainer}. + * Note: only endpoint options are available from spec. + * The {@code listenerContainer} options should be specified + * on the provided {@link DirectMessageListenerContainer} using + * {@link AmqpInboundGatewayDMLCSpec#configureContainer(java.util.function.Consumer)}. + * @param listenerContainer the listenerContainer + * @return the AmqpInboundGatewayDMLCSpec. + */ + public static AmqpInboundGatewayDMLCSpec inboundGateway(DirectMessageListenerContainer listenerContainer) { + return new AmqpInboundGatewayDMLCSpec(listenerContainer); + } + + /** + * Create an initial {@link AmqpInboundGatewayDMLCSpec} + * with provided {@link DirectMessageListenerContainer}. + * Note: only endpoint options are available from spec. + * The {@code listenerContainer} options should be specified + * on the provided {@link DirectMessageListenerContainer} using + * {@link AmqpInboundGatewayDMLCSpec#configureContainer(java.util.function.Consumer)}. + * @param listenerContainer the listenerContainer + * @param amqpTemplate the {@link AmqpTemplate} to use. + * @return the AmqpInboundGatewayDMLCSpec. + */ + public static AmqpInboundGatewayDMLCSpec inboundGateway(DirectMessageListenerContainer listenerContainer, + AmqpTemplate amqpTemplate) { + return new AmqpInboundGatewayDMLCSpec(listenerContainer, amqpTemplate); + } + + /** + * Create an initial AmqpInboundChannelAdapterSpec using a + * {@link SimpleMessageListenerContainer}. * @param connectionFactory the connectionFactory. * @param queueNames the queueNames. * @return the AmqpInboundChannelAdapterSpec. */ - public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, + public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(ConnectionFactory connectionFactory, String... queueNames) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueueNames(queueNames); - return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer); + return new AmqpInboundChannelAdapterSMLCSpec(listenerContainer); } /** - * Create an initial AmqpInboundChannelAdapterSpec. + * Create an initial AmqpInboundChannelAdapterSpec using a + * {@link SimpleMessageListenerContainer}. * @param connectionFactory the connectionFactory. * @param queues the queues. * @return the AmqpInboundChannelAdapterSpec. */ - public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, Queue... queues) { + public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(ConnectionFactory connectionFactory, Queue... queues) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueues(queues); - return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer); + return new AmqpInboundChannelAdapterSMLCSpec(listenerContainer); } /** - * Create an initial AmqpInboundChannelAdapterSpec. + * Create an initial {@link AmqpInboundGatewaySMLCSpec} + * with provided {@link SimpleMessageListenerContainer}. + * Note: only endpoint options are available from spec. + * The {@code listenerContainer} options should be specified + * on the provided {@link SimpleMessageListenerContainer} using + * {@link AmqpInboundGatewaySMLCSpec#configureContainer(java.util.function.Consumer)}. * @param listenerContainer the listenerContainer - * @return the AmqpInboundChannelAdapterSpec. + * @return the AmqpInboundGatewaySMLCSpec. */ - public static AmqpBaseInboundChannelAdapterSpec inboundAdapter( - SimpleMessageListenerContainer listenerContainer) { - return new AmqpInboundChannelAdapterSpec(listenerContainer); + public static AmqpInboundChannelAdapterSMLCSpec inboundAdapter(SimpleMessageListenerContainer listenerContainer) { + return new AmqpInboundChannelAdapterSMLCSpec(listenerContainer); + } + + /** + * Create an initial {@link AmqpInboundGatewayDMLCSpec} + * with provided {@link DirectMessageListenerContainer}. + * Note: only endpoint options are available from spec. + * The {@code listenerContainer} options should be specified + * on the provided {@link DirectMessageListenerContainer} using + * {@link AmqpInboundGatewaySMLCSpec#configureContainer(java.util.function.Consumer)}. + * @param listenerContainer the listenerContainer + * @return the AmqpInboundGatewaySMLCSpec. + */ + public static AmqpInboundChannelAdapterDMLCSpec inboundAdapter(DirectMessageListenerContainer listenerContainer) { + return new AmqpInboundChannelAdapterDMLCSpec(listenerContainer); } /** diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java new file mode 100644 index 0000000000..6004e57068 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterDMLCSpec.java @@ -0,0 +1,42 @@ +/* + * Copyright 2017 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.amqp.dsl; + +import java.util.function.Consumer; + +import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; + +/** + * Spec for an inbound channel adapter with a {@link DirectMessageListenerContainer}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class AmqpInboundChannelAdapterDMLCSpec extends AmqpInboundChannelAdapterSpec { + + AmqpInboundChannelAdapterDMLCSpec(DirectMessageListenerContainer listenerContainer) { + super(listenerContainer); + } + + AmqpInboundChannelAdapterDMLCSpec configureContainer(Consumer configurer) { + configurer.accept(new DirectMessageListenerContainerSpec(this.listenerContainer)); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java new file mode 100644 index 0000000000..ad12069cf5 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSMLCSpec.java @@ -0,0 +1,42 @@ +/* + * Copyright 2017 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.amqp.dsl; + +import java.util.function.Consumer; + +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; + +/** + * Spec for an inbound channel adapter with a {@link SimpleMessageListenerContainer}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class AmqpInboundChannelAdapterSMLCSpec extends AmqpInboundChannelAdapterSpec { + + AmqpInboundChannelAdapterSMLCSpec(SimpleMessageListenerContainer listenerContainer) { + super(listenerContainer); + } + + AmqpInboundChannelAdapterSMLCSpec configureContainer(Consumer configurer) { + configurer.accept(new SimpleMessageListenerContainerSpec(this.listenerContainer)); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java index 4f38cf6d21..dc092be0f2 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java @@ -18,246 +18,33 @@ package org.springframework.integration.amqp.dsl; import java.util.Collection; import java.util.Collections; -import java.util.concurrent.Executor; -import org.aopalliance.aop.Advice; - -import org.springframework.amqp.core.AcknowledgeMode; -import org.springframework.amqp.core.Queue; -import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; import org.springframework.integration.dsl.ComponentsRegistration; import org.springframework.integration.dsl.MessageProducerSpec; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.util.ErrorHandler; /** * A {@link MessageProducerSpec} for {@link AmqpInboundChannelAdapter}s. * + * @param the spec type. + * @param the container type. + * * @author Artem Bilan * @since 5.0 */ -public class AmqpInboundChannelAdapterSpec extends AmqpBaseInboundChannelAdapterSpec +public abstract class AmqpInboundChannelAdapterSpec + , C extends AbstractMessageListenerContainer> + extends AmqpBaseInboundChannelAdapterSpec implements ComponentsRegistration { - private final SimpleMessageListenerContainer listenerContainer; + protected final C listenerContainer; - AmqpInboundChannelAdapterSpec(SimpleMessageListenerContainer listenerContainer) { + AmqpInboundChannelAdapterSpec(C listenerContainer) { super(new AmqpInboundChannelAdapter(listenerContainer)); this.listenerContainer = listenerContainer; } - /** - * @param acknowledgeMode the acknowledgeMode. - * @return the spec. - * @see SimpleMessageListenerContainer#setAcknowledgeMode(AcknowledgeMode) - */ - public AmqpInboundChannelAdapterSpec acknowledgeMode(AcknowledgeMode acknowledgeMode) { - this.listenerContainer.setAcknowledgeMode(acknowledgeMode); - return this; - } - - /** - * @param queueName a vararg list of queue names to add. - * @return the spec. - * @see SimpleMessageListenerContainer#addQueueNames(String...) - */ - public AmqpInboundChannelAdapterSpec addQueueNames(String... queueName) { - this.listenerContainer.addQueueNames(queueName); - return this; - } - - /** - * @param queues a vararg list of queues to add. - * @return the spec. - * @see SimpleMessageListenerContainer#addQueueNames(String...) - */ - public AmqpInboundChannelAdapterSpec addQueues(Queue... queues) { - this.listenerContainer.addQueues(queues); - return this; - } - - /** - * @param errorHandler the errorHandler. - * @return the spec. - * @see SimpleMessageListenerContainer#setErrorHandler(ErrorHandler) - */ - public AmqpInboundChannelAdapterSpec errorHandler(ErrorHandler errorHandler) { - this.listenerContainer.setErrorHandler(errorHandler); - return this; - } - - /** - * @param transactional true for transactional channels. - * @return the spec. - * @see SimpleMessageListenerContainer#setChannelTransacted(boolean) - */ - public AmqpInboundChannelAdapterSpec channelTransacted(boolean transactional) { - this.listenerContainer.setChannelTransacted(transactional); - return this; - } - - /** - * @param adviceChain the adviceChain. - * @return the spec. - * @see SimpleMessageListenerContainer#setAdviceChain(Advice[]) - */ - public AmqpInboundChannelAdapterSpec adviceChain(Advice... adviceChain) { - this.listenerContainer.setAdviceChain(adviceChain); - return this; - } - - /** - * @param recoveryInterval the recoveryInterval - * @return the spec. - * @see SimpleMessageListenerContainer#setRecoveryInterval(long) - */ - public AmqpInboundChannelAdapterSpec recoveryInterval(long recoveryInterval) { - this.listenerContainer.setRecoveryInterval(recoveryInterval); - return this; - } - - /** - * @param concurrentConsumers the concurrentConsumers - * @return the spec. - * @see SimpleMessageListenerContainer#setConcurrentConsumers(int) - */ - public AmqpInboundChannelAdapterSpec concurrentConsumers(int concurrentConsumers) { - this.listenerContainer.setConcurrentConsumers(concurrentConsumers); - return this; - } - - /** - * @param maxConcurrentConsumers the maxConcurrentConsumers. - * @return the spec. - * @see SimpleMessageListenerContainer#setMaxConcurrentConsumers(int) - */ - public AmqpInboundChannelAdapterSpec maxConcurrentConsumers(int maxConcurrentConsumers) { - this.listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers); - return this; - } - - /** - * @param exclusive true for exclusive. - * @return the spec. - * @see SimpleMessageListenerContainer#setExclusive(boolean) - */ - public AmqpInboundChannelAdapterSpec exclusive(boolean exclusive) { - this.listenerContainer.setExclusive(exclusive); - return this; - } - - /** - * @param startConsumerMinInterval the startConsumerMinInterval - * @return the spec. - * @see SimpleMessageListenerContainer#setStartConsumerMinInterval(long) - */ - public AmqpInboundChannelAdapterSpec startConsumerMinInterval(long startConsumerMinInterval) { - this.listenerContainer.setStartConsumerMinInterval(startConsumerMinInterval); - return this; - } - - /** - * @param stopConsumerMinInterval the stopConsumerMinInterval. - * @return the spec. - * @see SimpleMessageListenerContainer#setStopConsumerMinInterval(long) - */ - public AmqpInboundChannelAdapterSpec stopConsumerMinInterval(long stopConsumerMinInterval) { - this.listenerContainer.setStopConsumerMinInterval(stopConsumerMinInterval); - return this; - } - - /** - * @param consecutiveActiveTrigger the consecutiveActiveTrigger. - * @return the spec. - * @see SimpleMessageListenerContainer#setConsecutiveActiveTrigger(int) - */ - public AmqpInboundChannelAdapterSpec consecutiveActiveTrigger(int consecutiveActiveTrigger) { - this.listenerContainer.setConsecutiveActiveTrigger(consecutiveActiveTrigger); - return this; - } - - /** - * @param consecutiveIdleTrigger the consecutiveIdleTrigger. - * @return the spec. - * @see SimpleMessageListenerContainer#setConsecutiveIdleTrigger(int) - */ - public AmqpInboundChannelAdapterSpec consecutiveIdleTrigger(int consecutiveIdleTrigger) { - this.listenerContainer.setConsecutiveIdleTrigger(consecutiveIdleTrigger); - return this; - } - - /** - * @param receiveTimeout the receiveTimeout - * @return the spec. - * @see SimpleMessageListenerContainer#setReceiveTimeout(long) - */ - public AmqpInboundChannelAdapterSpec receiveTimeout(long receiveTimeout) { - this.listenerContainer.setReceiveTimeout(receiveTimeout); - return this; - } - - /** - * @param shutdownTimeout the shutdownTimeout. - * @return the spec. - * @see SimpleMessageListenerContainer#setShutdownTimeout(long) - */ - public AmqpInboundChannelAdapterSpec shutdownTimeout(long shutdownTimeout) { - this.listenerContainer.setShutdownTimeout(shutdownTimeout); - return this; - } - - /** - * Configure an {@link Executor} used to invoke the message listener. - * @param taskExecutor the taskExecutor. - * @return the spec. - */ - public AmqpInboundChannelAdapterSpec taskExecutor(Executor taskExecutor) { - this.listenerContainer.setTaskExecutor(taskExecutor); - return this; - } - - /** - * @param prefetchCount the prefetchCount. - * @return the spec. - * @see SimpleMessageListenerContainer#setPrefetchCount(int) - */ - public AmqpInboundChannelAdapterSpec prefetchCount(int prefetchCount) { - this.listenerContainer.setPrefetchCount(prefetchCount); - return this; - } - - /** - * @param txSize the txSize. - * @return the spec. - * @see SimpleMessageListenerContainer#setTxSize(int) - */ - public AmqpInboundChannelAdapterSpec txSize(int txSize) { - this.listenerContainer.setTxSize(txSize); - return this; - } - - /** - * Configure a {@link PlatformTransactionManager}; used to synchronize the rabbit transaction - * with some other transaction(s). - * @param transactionManager the transactionManager. - * @return the spec. - */ - public AmqpInboundChannelAdapterSpec transactionManager(PlatformTransactionManager transactionManager) { - this.listenerContainer.setTransactionManager(transactionManager); - return this; - } - - /** - * @param defaultRequeueRejected the defaultRequeueRejected. - * @return the spec. - * @see SimpleMessageListenerContainer#setDefaultRequeueRejected(boolean) - */ - public AmqpInboundChannelAdapterSpec defaultRequeueRejected(boolean defaultRequeueRejected) { - this.listenerContainer.setDefaultRequeueRejected(defaultRequeueRejected); - return this; - } - @Override public Collection getComponentsToRegister() { return Collections.singleton(this.listenerContainer); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java new file mode 100644 index 0000000000..5257a6e8ba --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewayDMLCSpec.java @@ -0,0 +1,47 @@ +/* + * Copyright 2017 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.amqp.dsl; + +import java.util.function.Consumer; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; + +/** + * Spec for a gateway with a {@link DirectMessageListenerContainer}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class AmqpInboundGatewayDMLCSpec + extends AmqpInboundGatewaySpec { + + AmqpInboundGatewayDMLCSpec(DirectMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { + super(listenerContainer, amqpTemplate); + } + + AmqpInboundGatewayDMLCSpec(DirectMessageListenerContainer listenerContainer) { + super(listenerContainer); + } + + public AmqpInboundGatewayDMLCSpec configureContainer(Consumer configurer) { + configurer.accept(new DirectMessageListenerContainerSpec(this.listenerContainer)); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java new file mode 100644 index 0000000000..8eef0c103b --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySMLCSpec.java @@ -0,0 +1,47 @@ +/* + * Copyright 2017 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.amqp.dsl; + +import java.util.function.Consumer; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; + +/** + * Spec for a gateway with a {@link SimpleMessageListenerContainer}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class AmqpInboundGatewaySMLCSpec + extends AmqpInboundGatewaySpec { + + AmqpInboundGatewaySMLCSpec(SimpleMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { + super(listenerContainer, amqpTemplate); + } + + AmqpInboundGatewaySMLCSpec(SimpleMessageListenerContainer listenerContainer) { + super(listenerContainer); + } + + public AmqpInboundGatewaySMLCSpec configureContainer(Consumer configurer) { + configurer.accept(new SimpleMessageListenerContainerSpec(this.listenerContainer)); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java index c9dc9d7d84..8fe7e64319 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java @@ -18,262 +18,45 @@ package org.springframework.integration.amqp.dsl; import java.util.Collection; import java.util.Collections; -import java.util.concurrent.Executor; -import org.aopalliance.aop.Advice; - -import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.AmqpTemplate; -import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.integration.amqp.inbound.AmqpInboundGateway; import org.springframework.integration.dsl.ComponentsRegistration; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.util.ErrorHandler; /** * An {@link AmqpBaseInboundGatewaySpec} implementation for a {@link AmqpInboundGateway}. - * Allows to provide {@link SimpleMessageListenerContainer} options. + * Allows to provide {@link AbstractMessageListenerContainer} options. * * @author Artem Bilan * @since 5.0 */ -public class AmqpInboundGatewaySpec extends AmqpBaseInboundGatewaySpec - implements ComponentsRegistration { +public abstract class AmqpInboundGatewaySpec + , C extends AbstractMessageListenerContainer> + extends AmqpBaseInboundGatewaySpec implements ComponentsRegistration { - private final SimpleMessageListenerContainer listenerContainer; + protected final C listenerContainer; - AmqpInboundGatewaySpec(SimpleMessageListenerContainer listenerContainer) { + AmqpInboundGatewaySpec(C listenerContainer) { super(new AmqpInboundGateway(listenerContainer)); this.listenerContainer = listenerContainer; } /** - * Instantiate {@link AmqpInboundGateway} based on the provided {@link SimpleMessageListenerContainer} + * Instantiate {@link AmqpInboundGateway} based on the provided {@link AbstractMessageListenerContainer} * and {@link AmqpTemplate}. * @param listenerContainer the {@link SimpleMessageListenerContainer} to use. * @param amqpTemplate the {@link AmqpTemplate} to use. */ - AmqpInboundGatewaySpec(SimpleMessageListenerContainer listenerContainer, AmqpTemplate amqpTemplate) { + AmqpInboundGatewaySpec(C listenerContainer, AmqpTemplate amqpTemplate) { super(new AmqpInboundGateway(listenerContainer, amqpTemplate)); this.listenerContainer = listenerContainer; } - /** - * @param acknowledgeMode the acknowledgeMode. - * @return the spec. - * @see SimpleMessageListenerContainer#setAcknowledgeMode(AcknowledgeMode) - */ - public AmqpInboundGatewaySpec acknowledgeMode(AcknowledgeMode acknowledgeMode) { - this.listenerContainer.setAcknowledgeMode(acknowledgeMode); - return this; - } - - /** - * @param queueName a vararg list of queue names to add. - * @return the spec. - * @see SimpleMessageListenerContainer#addQueueNames(String...) - */ - public AmqpInboundGatewaySpec addQueueNames(String... queueName) { - this.listenerContainer.addQueueNames(queueName); - return this; - } - - /** - * @param queues a vararg list of queues to add. - * @return the spec. - * @see SimpleMessageListenerContainer#addQueueNames(String...) - */ - public AmqpInboundGatewaySpec addQueues(Queue... queues) { - this.listenerContainer.addQueues(queues); - return this; - } - - /** - * @param errorHandler the errorHandler. - * @return the spec. - * @see SimpleMessageListenerContainer#setErrorHandler(ErrorHandler) - */ - public AmqpInboundGatewaySpec errorHandler(ErrorHandler errorHandler) { - this.listenerContainer.setErrorHandler(errorHandler); - return this; - } - - /** - * @param transactional true for transactional channels. - * @return the spec. - * @see SimpleMessageListenerContainer#setChannelTransacted(boolean) - */ - public AmqpInboundGatewaySpec channelTransacted(boolean transactional) { - this.listenerContainer.setChannelTransacted(transactional); - return this; - } - - /** - * @param adviceChain the adviceChain. - * @return the spec. - * @see SimpleMessageListenerContainer#setAdviceChain(Advice[]) - */ - public AmqpInboundGatewaySpec adviceChain(Advice... adviceChain) { - this.listenerContainer.setAdviceChain(adviceChain); - return this; - } - - /** - * @param recoveryInterval the recoveryInterval - * @return the spec. - * @see SimpleMessageListenerContainer#setRecoveryInterval(long) - */ - public AmqpInboundGatewaySpec recoveryInterval(long recoveryInterval) { - this.listenerContainer.setRecoveryInterval(recoveryInterval); - return this; - } - - /** - * @param concurrentConsumers the concurrentConsumers - * @return the spec. - * @see SimpleMessageListenerContainer#setConcurrentConsumers(int) - */ - public AmqpInboundGatewaySpec concurrentConsumers(int concurrentConsumers) { - this.listenerContainer.setConcurrentConsumers(concurrentConsumers); - return this; - } - - /** - * @param maxConcurrentConsumers the maxConcurrentConsumers. - * @return the spec. - * @see SimpleMessageListenerContainer#setMaxConcurrentConsumers(int) - */ - public AmqpInboundGatewaySpec maxConcurrentConsumers(int maxConcurrentConsumers) { - this.listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers); - return this; - } - - /** - * @param exclusive true for exclusive. - * @return the spec. - * @see SimpleMessageListenerContainer#setExclusive(boolean) - */ - public AmqpInboundGatewaySpec exclusive(boolean exclusive) { - this.listenerContainer.setExclusive(exclusive); - return this; - } - - /** - * @param startConsumerMinInterval the startConsumerMinInterval - * @return the spec. - * @see SimpleMessageListenerContainer#setStartConsumerMinInterval(long) - */ - public AmqpInboundGatewaySpec startConsumerMinInterval(long startConsumerMinInterval) { - this.listenerContainer.setStartConsumerMinInterval(startConsumerMinInterval); - return this; - } - - /** - * @param stopConsumerMinInterval the stopConsumerMinInterval. - * @return the spec. - * @see SimpleMessageListenerContainer#setStopConsumerMinInterval(long) - */ - public AmqpInboundGatewaySpec stopConsumerMinInterval(long stopConsumerMinInterval) { - this.listenerContainer.setStopConsumerMinInterval(stopConsumerMinInterval); - return this; - } - - /** - * @param consecutiveActiveTrigger the consecutiveActiveTrigger. - * @return the spec. - * @see SimpleMessageListenerContainer#setConsecutiveActiveTrigger(int) - */ - public AmqpInboundGatewaySpec consecutiveActiveTrigger(int consecutiveActiveTrigger) { - this.listenerContainer.setConsecutiveActiveTrigger(consecutiveActiveTrigger); - return this; - } - - /** - * @param consecutiveIdleTrigger the consecutiveIdleTrigger. - * @return the spec. - * @see SimpleMessageListenerContainer#setConsecutiveIdleTrigger(int) - */ - public AmqpInboundGatewaySpec consecutiveIdleTrigger(int consecutiveIdleTrigger) { - this.listenerContainer.setConsecutiveIdleTrigger(consecutiveIdleTrigger); - return this; - } - - /** - * @param receiveTimeout the receiveTimeout - * @return the spec. - * @see SimpleMessageListenerContainer#setReceiveTimeout(long) - */ - public AmqpInboundGatewaySpec receiveTimeout(long receiveTimeout) { - this.listenerContainer.setReceiveTimeout(receiveTimeout); - return this; - } - - /** - * @param shutdownTimeout the shutdownTimeout. - * @return the spec. - * @see SimpleMessageListenerContainer#setShutdownTimeout(long) - */ - public AmqpInboundGatewaySpec shutdownTimeout(long shutdownTimeout) { - this.listenerContainer.setShutdownTimeout(shutdownTimeout); - return this; - } - - /** - * Configure an {@link Executor} used to invoke the message listener. - * @param taskExecutor the taskExecutor. - * @return the spec. - */ - public AmqpInboundGatewaySpec taskExecutor(Executor taskExecutor) { - this.listenerContainer.setTaskExecutor(taskExecutor); - return this; - } - - /** - * @param prefetchCount the prefetchCount. - * @return the spec. - * @see SimpleMessageListenerContainer#setPrefetchCount(int) - */ - public AmqpInboundGatewaySpec prefetchCount(int prefetchCount) { - this.listenerContainer.setPrefetchCount(prefetchCount); - return this; - } - - /** - * @param txSize the txSize. - * @return the spec. - * @see SimpleMessageListenerContainer#setTxSize(int) - */ - public AmqpInboundGatewaySpec txSize(int txSize) { - this.listenerContainer.setTxSize(txSize); - return this; - } - - /** - * Configure a {@link PlatformTransactionManager}; used to synchronize the rabbit transaction - * with some other transaction(s). - * @param transactionManager the transactionManager. - * @return the spec. - */ - public AmqpInboundGatewaySpec transactionManager(PlatformTransactionManager transactionManager) { - this.listenerContainer.setTransactionManager(transactionManager); - return this; - } - - /** - * @param defaultRequeueRejected the defaultRequeueRejected. - * @return the spec. - * @see SimpleMessageListenerContainer#setDefaultRequeueRejected(boolean) - */ - public AmqpInboundGatewaySpec defaultRequeueRejected(boolean defaultRequeueRejected) { - this.listenerContainer.setDefaultRequeueRejected(defaultRequeueRejected); - return this; - } - @Override public Collection getComponentsToRegister() { return Collections.singleton(this.listenerContainer); } - } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java new file mode 100644 index 0000000000..1b7263b19d --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/DirectMessageListenerContainerSpec.java @@ -0,0 +1,68 @@ +/* + * Copyright 2017 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.amqp.dsl; + +import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; + +/** + * Spec for a {@link DirectMessageListenerContainer}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class DirectMessageListenerContainerSpec + extends AbstractMessageListenerContainerSpec { + + private final DirectMessageListenerContainer listenerContainer; + + public DirectMessageListenerContainerSpec(DirectMessageListenerContainer listenerContainer) { + super(listenerContainer); + this.listenerContainer = listenerContainer; + } + + /** + * @param consumersPerQueue the consumersPerQueue. + * @return the spec. + * @see DirectMessageListenerContainer#setConsumersPerQueue(int) + */ + public DirectMessageListenerContainerSpec consumersPerQueue(int consumersPerQueue) { + this.listenerContainer.setConsumersPerQueue(consumersPerQueue); + return this; + } + + /** + * @param messagesPerAck the messages per ack. + * @return the spec. + * @see DirectMessageListenerContainer#setMessagesPerAck(int) + */ + public DirectMessageListenerContainerSpec messagesPerAck(int messagesPerAck) { + this.listenerContainer.setMessagesPerAck(messagesPerAck); + return this; + } + + /** + * @param ackTimeout the ack timeout. + * @return the spec. + * @see DirectMessageListenerContainer#setAckTimeout(long) + */ + public DirectMessageListenerContainerSpec ackTimeout(long ackTimeout) { + this.listenerContainer.setAckTimeout(ackTimeout); + return this; + } + +} diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java new file mode 100644 index 0000000000..aed7e8bea2 --- /dev/null +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/SimpleMessageListenerContainerSpec.java @@ -0,0 +1,118 @@ +/* + * Copyright 2017 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.amqp.dsl; + +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; + +/** + * Spec for a {@link SimpleMessageListenerContainer}. + * + * @author Gary Russell + * @since 5.0 + * + */ +public class SimpleMessageListenerContainerSpec extends + AbstractMessageListenerContainerSpec { + + private final SimpleMessageListenerContainer listenerContainer; + + public SimpleMessageListenerContainerSpec(SimpleMessageListenerContainer listenerContainer) { + super(listenerContainer); + this.listenerContainer = listenerContainer; + } + + /** + * @param concurrentConsumers the concurrentConsumers + * @return the spec. + * @see SimpleMessageListenerContainer#setConcurrentConsumers(int) + */ + public SimpleMessageListenerContainerSpec concurrentConsumers(int concurrentConsumers) { + this.listenerContainer.setConcurrentConsumers(concurrentConsumers); + return this; + } + + /** + * @param maxConcurrentConsumers the maxConcurrentConsumers. + * @return the spec. + * @see SimpleMessageListenerContainer#setMaxConcurrentConsumers(int) + */ + public SimpleMessageListenerContainerSpec maxConcurrentConsumers(int maxConcurrentConsumers) { + this.listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers); + return this; + } + + /** + * @param startConsumerMinInterval the startConsumerMinInterval + * @return the spec. + * @see SimpleMessageListenerContainer#setStartConsumerMinInterval(long) + */ + public SimpleMessageListenerContainerSpec startConsumerMinInterval(long startConsumerMinInterval) { + this.listenerContainer.setStartConsumerMinInterval(startConsumerMinInterval); + return this; + } + + /** + * @param stopConsumerMinInterval the stopConsumerMinInterval. + * @return the spec. + * @see SimpleMessageListenerContainer#setStopConsumerMinInterval(long) + */ + public SimpleMessageListenerContainerSpec stopConsumerMinInterval(long stopConsumerMinInterval) { + this.listenerContainer.setStopConsumerMinInterval(stopConsumerMinInterval); + return this; + } + + /** + * @param consecutiveActiveTrigger the consecutiveActiveTrigger. + * @return the spec. + * @see SimpleMessageListenerContainer#setConsecutiveActiveTrigger(int) + */ + public SimpleMessageListenerContainerSpec consecutiveActiveTrigger(int consecutiveActiveTrigger) { + this.listenerContainer.setConsecutiveActiveTrigger(consecutiveActiveTrigger); + return this; + } + + /** + * @param consecutiveIdleTrigger the consecutiveIdleTrigger. + * @return the spec. + * @see SimpleMessageListenerContainer#setConsecutiveIdleTrigger(int) + */ + public SimpleMessageListenerContainerSpec consecutiveIdleTrigger(int consecutiveIdleTrigger) { + this.listenerContainer.setConsecutiveIdleTrigger(consecutiveIdleTrigger); + return this; + } + + /** + * @param receiveTimeout the receiveTimeout + * @return the spec. + * @see SimpleMessageListenerContainer#setReceiveTimeout(long) + */ + public SimpleMessageListenerContainerSpec receiveTimeout(long receiveTimeout) { + this.listenerContainer.setReceiveTimeout(receiveTimeout); + return this; + } + + /** + * @param txSize the txSize. + * @return the spec. + * @see SimpleMessageListenerContainer#setTxSize(int) + */ + public SimpleMessageListenerContainerSpec txSize(int txSize) { + this.listenerContainer.setTxSize(txSize); + return this; + } + +} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java index 07fe64e543..4b3eaeeb92 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java @@ -36,6 +36,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.junit.BrokerRunning; +import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; @@ -211,6 +212,23 @@ public class AmqpTests { return IntegrationFlows .from(Amqp.inboundGateway(rabbitConnectionFactory, amqpTemplate, queue()) .id("amqpInboundGateway") + .configureContainer(c -> c + .recoveryInterval(5000) + .concurrentConsumers(1)) + .defaultReplyTo(defaultReplyTo().getName())) + .transform("hello "::concat) + .transform(String.class, String::toUpperCase) + .get(); + } + + // syntax only + public IntegrationFlow amqpDMLCFlow(ConnectionFactory rabbitConnectionFactory, AmqpTemplate amqpTemplate) { + return IntegrationFlows + .from(Amqp.inboundGateway(new DirectMessageListenerContainer()) + .id("amqpInboundGateway") + .configureContainer(c -> c + .recoveryInterval(5000) + .consumersPerQueue(1)) .defaultReplyTo(defaultReplyTo().getName())) .transform("hello "::concat) .transform(String.class, String::toUpperCase)