diff --git a/pom.xml b/pom.xml index 67d87c78f..3fd2788a6 100644 --- a/pom.xml +++ b/pom.xml @@ -22,6 +22,11 @@ pom import + + org.springframework.cloud + spring-cloud-stream-binder-rabbit + 1.1.0.BUILD-SNAPSHOT + @@ -29,6 +34,19 @@ spring-cloud-starter-stream-rabbit spring-cloud-stream-binder-rabbit-test-support + + + + + org.apache.maven.plugins + maven-surefire-plugin + + true + + + + + spring diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index a996de14d..a88249e89 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -16,15 +16,11 @@ package org.springframework.cloud.stream.binder.rabbit; -import java.io.IOException; import java.util.Arrays; import java.util.HashMap; -import java.util.Iterator; -import java.util.LinkedList; import java.util.Map; import com.rabbitmq.client.AMQP; -import com.rabbitmq.client.Channel; import com.rabbitmq.client.Envelope; import org.aopalliance.aop.Advice; @@ -51,38 +47,25 @@ import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer; import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; -import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; -import org.springframework.cloud.stream.binder.AbstractBinder; -import org.springframework.cloud.stream.binder.Binding; -import org.springframework.cloud.stream.binder.DefaultBinding; +import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; -import org.springframework.cloud.stream.binder.MessageValues; -import org.springframework.cloud.stream.binder.PartitionHandler; -import org.springframework.context.Lifecycle; import org.springframework.context.support.GenericApplicationContext; import org.springframework.core.task.SimpleAsyncTaskExecutor; -import org.springframework.expression.ExpressionParser; -import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.context.IntegrationContextUtils; -import org.springframework.integration.endpoint.EventDrivenConsumer; -import org.springframework.integration.handler.AbstractMessageHandler; -import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.messaging.Message; +import org.springframework.integration.core.MessageProducer; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.SubscribableChannel; import org.springframework.retry.interceptor.RetryOperationsInterceptor; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; @@ -90,7 +73,6 @@ import org.springframework.util.StringUtils; /** * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ. - * * @author Mark Fisher * @author Gary Russell * @author Jennifer Hickey @@ -99,10 +81,12 @@ import org.springframework.util.StringUtils; * @author David Turanski * @author Marius Bogoevici */ -public class RabbitMessageChannelBinder extends AbstractBinder, - ExtendedProducerProperties> implements ExtendedPropertiesBinder { +public class RabbitMessageChannelBinder + extends AbstractMessageChannelBinder, + ExtendedProducerProperties, Queue> + implements ExtendedPropertiesBinder { - public static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR + private static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR = new AnonymousQueue.Base64UrlNamingStrategy("anonymous."); private static final String DEAD_LETTER_EXCHANGE = "DLX"; @@ -119,8 +103,6 @@ public class RabbitMessageChannelBinder extends AbstractBinder doBindConsumer(String name, String group, MessageChannel inputChannel, + protected MessageProducer createConsumerEndpoint(String name, String group, Queue destination, ExtendedConsumerProperties properties) { - boolean anonymousConsumer = !StringUtils.hasText(group); - String baseQueueName = anonymousConsumer ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName()) - : groupedName(name, group); - if (this.logger.isInfoEnabled()) { - this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name); - } - String prefix = properties.getExtension().getPrefix(); - String exchangeName = applyPrefix(prefix, name); - TopicExchange exchange = new TopicExchange(exchangeName); - declareExchange(exchangeName, exchange); - String queueName = applyPrefix(prefix, baseQueueName); - boolean partitioned = !anonymousConsumer && properties.isPartitioned(); - boolean durable = !anonymousConsumer && properties.getExtension().isDurableSubscription(); - Queue queue; + DirectChannel convertingBridgeChannel = new DirectChannel(); + convertingBridgeChannel.setBeanFactory(this.getBeanFactory()); + final String baseQueueName = baseQueueName(name, group); + convertingBridgeChannel.setBeanName(baseQueueName + ".bridge"); - if (anonymousConsumer) { - queue = new Queue(queueName, false, true, true); - } - else { - if (partitioned) { - String partitionSuffix = "-" + properties.getInstanceIndex(); - queueName += partitionSuffix; - } - if (durable) { - queue = new Queue(queueName, true, false, false, - queueArgs(queueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq())); - } - else { - queue = new Queue(queueName, false, false, true); - } - } - - declareQueue(queueName, queue); - - if (partitioned) { - String bindingKey = String.format("%s-%d", name, properties.getInstanceIndex()); - declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(bindingKey)); - } - else { - declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with("#")); - } - Binding binding = doRegisterConsumer(baseQueueName, group, inputChannel, queue, properties); - if (durable) { - autoBindDLQ(applyPrefix(prefix, baseQueueName), queueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()); - } - return binding; - } - - private Map queueArgs(String queueName, String prefix, boolean bindDlq) { - Map args = new HashMap<>(); - if (bindDlq) { - args.put("x-dead-letter-exchange", applyPrefix(prefix, "DLX")); - args.put("x-dead-letter-routing-key", queueName); - } - return args; - } - - private Binding doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue, - final ExtendedConsumerProperties properties) { - DefaultBinding consumerBinding; SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer( this.connectionFactory); listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode()); @@ -290,84 +219,149 @@ public class RabbitMessageChannelBinder extends AbstractBinder 1 || properties.getExtension().isRepublishToDlq()) { + listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(destination.getName() + "-")); + listenerContainer.setQueues((Queue) destination); + if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) { RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless() - .maxAttempts(maxAttempts) - .backOffOptions(properties.getBackOffInitialInterval(), - properties.getBackOffMultiplier(), - properties.getBackOffMaxInterval()) - .recoverer(determineRecoverer(name, properties.getExtension().getPrefix(), properties.getExtension().isRepublishToDlq())) + .retryOperations(buildRetryTemplate(properties)) + .recoverer(determineRecoverer(baseQueueName, properties.getExtension().getPrefix(), + properties.getExtension().isRepublishToDlq())) .build(); - listenerContainer.setAdviceChain(new Advice[] { retryInterceptor }); + listenerContainer.setAdviceChain(new Advice[] {retryInterceptor}); } listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor); listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter); listenerContainer.afterPropertiesSet(); + AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer); adapter.setBeanFactory(this.getBeanFactory()); - DirectChannel bridgeToModuleChannel = new DirectChannel(); - bridgeToModuleChannel.setBeanFactory(this.getBeanFactory()); - bridgeToModuleChannel.setBeanName(name + ".bridge"); - adapter.setOutputChannel(bridgeToModuleChannel); - adapter.setBeanName("inbound." + name); - DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper(); + adapter.setBeanName("inbound." + baseQueueName); + DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper(); mapper.setRequestHeaderNames(properties.getExtension().getRequestHeaderPatterns()); mapper.setReplyHeaderNames(properties.getExtension().getReplyHeaderPatterns()); adapter.setHeaderMapper(mapper); adapter.afterPropertiesSet(); - consumerBinding = new DefaultBinding(name, group, moduleInputChannel, adapter) { - @Override - protected void afterUnbind() { - cleanAutoDeclareContext(properties.getExtension().getPrefix(), name); + return adapter; + } + + @Override + protected void afterUnbindConsumer(String name, String group, + ExtendedConsumerProperties consumerProperties) { + cleanAutoDeclareContext(consumerProperties.getExtension().getPrefix(), baseQueueName(name, group)); + } + + @Override + protected Queue createConsumerDestinationIfNecessary(String name, String group, + ExtendedConsumerProperties properties) { + boolean anonymous = !StringUtils.hasText(group); + String baseQueueName = anonymous ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName()) + : groupedName(name, group); + if (this.logger.isInfoEnabled()) { + this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name); + } + String prefix = properties.getExtension().getPrefix(); + String exchangeName = applyPrefix(prefix, name); + TopicExchange exchange = new TopicExchange(exchangeName); + declareExchange(exchangeName, exchange); + + String queueName = applyPrefix(prefix, baseQueueName); + boolean partitioned = !anonymous && properties.isPartitioned(); + boolean durable = !anonymous && properties.getExtension().isDurableSubscription(); + Queue queue; + + if (anonymous) { + queue = new Queue(queueName, false, true, true); + } + else { + if (partitioned) { + String partitionSuffix = "-" + properties.getInstanceIndex(); + queueName += partitionSuffix; } - }; - ReceivingHandler convertingBridge = new ReceivingHandler(); - convertingBridge.setOutputChannel(moduleInputChannel); - convertingBridge.setBeanName(name + ".convert.bridge"); - convertingBridge.afterPropertiesSet(); - bridgeToModuleChannel.subscribe(convertingBridge); - adapter.start(); - return consumerBinding; + if (durable) { + queue = new Queue(queueName, true, false, false, + queueArgs(queueName, properties.getExtension().getPrefix(), + properties.getExtension().isAutoBindDlq())); + } + else { + queue = new Queue(queueName, false, false, true); + } + } + declareQueue(queueName, queue); + if (partitioned) { + String bindingKey = String.format("%s-%d", name, properties.getInstanceIndex()); + declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(bindingKey)); + } + else { + declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with("#")); + } + if (durable) { + autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName, + properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()); + } + return queue; + } + + private String baseQueueName(String name, String group) { + return !StringUtils.hasText(group) ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName()) + : groupedName(name, group); + } + + private Map queueArgs(String queueName, String prefix, boolean bindDlq) { + Map args = new HashMap<>(); + if (bindDlq) { + args.put("x-dead-letter-exchange", applyPrefix(prefix, "DLX")); + args.put("x-dead-letter-routing-key", queueName); + } + return args; } private MessageRecoverer determineRecoverer(String name, String prefix, boolean republish) { if (republish) { RabbitTemplate errorTemplate = new RabbitTemplate(this.connectionFactory); - RepublishMessageRecoverer republishMessageRecoverer = new RepublishMessageRecoverer(errorTemplate, + return new RepublishMessageRecoverer(errorTemplate, deadLetterExchangeName(prefix), applyPrefix(prefix, name)); - return republishMessageRecoverer; } else { return new RejectAndDontRequeueRecoverer(); } } - private AmqpOutboundEndpoint buildOutboundEndpoint(final String name, - ExtendedProducerProperties properties, - RabbitTemplate rabbitTemplate) { - String prefix = properties.getExtension().getPrefix(); - String exchangeName = applyPrefix(prefix, name); + @Override + protected void createProducerDestinationIfNecessary(String name, + ExtendedProducerProperties producerProperties) { + String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name); TopicExchange exchange = new TopicExchange(exchangeName); declareExchange(exchangeName, exchange); - AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(rabbitTemplate); + } + + @Override + protected MessageHandler createProducerMessageHandler(final String destination, + ExtendedProducerProperties properties) + throws Exception { + String prefix = properties.getExtension().getPrefix(); + String exchangeName = applyPrefix(prefix, destination); + TopicExchange exchange = new TopicExchange(exchangeName); + declareExchange(exchangeName, exchange); + final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(buildRabbitTemplate(properties.getExtension())); endpoint.setExchangeName(exchange.getName()); if (!properties.isPartitioned()) { - endpoint.setRoutingKey(name); + endpoint.setRoutingKey(destination); } else { - endpoint.setExpressionRoutingKey(EXPRESSION_PARSER.parseExpression(buildPartitionRoutingExpression(name))); + endpoint.setRoutingKeyExpression(EXPRESSION_PARSER.parseExpression(buildPartitionRoutingExpression( + destination))); } for (String requiredGroupName : properties.getRequiredGroups()) { String baseQueueName = exchangeName + "." + requiredGroupName; if (!properties.isPartitioned()) { - Queue queue = new Queue(baseQueueName, true, false, false, queueArgs(baseQueueName, prefix, properties.getExtension().isAutoBindDlq())); + Queue queue = new Queue(baseQueueName, true, false, false, + queueArgs(baseQueueName, prefix, properties.getExtension().isAutoBindDlq())); declareQueue(baseQueueName, queue); - autoBindDLQ(baseQueueName, baseQueueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()); - org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with(name); + autoBindDLQ(baseQueueName, baseQueueName, properties.getExtension().getPrefix(), + properties.getExtension().isAutoBindDlq()); + org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with( + destination); declareBinding(baseQueueName, binding); } else { @@ -376,41 +370,29 @@ public class RabbitMessageChannelBinder extends AbstractBinder producerProperties) { - DefaultAmqpHeaderMapper mapper = new DefaultAmqpHeaderMapper(); - mapper.setRequestHeaderNames(producerProperties.getExtension().getRequestHeaderPatterns()); - mapper.setReplyHeaderNames(producerProperties.getExtension().getReplyHeaderPatterns()); - handler.setHeaderMapper(mapper); - handler.setDefaultDeliveryMode(producerProperties.getExtension().getDeliveryMode()); - handler.setBeanFactory(this.getBeanFactory()); - handler.afterPropertiesSet(); - } - - @Override - public Binding doBindProducer(String name, MessageChannel outputChannel, - ExtendedProducerProperties producerProperties) { - String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name); - TopicExchange exchange = new TopicExchange(exchangeName); - declareExchange(exchangeName, exchange); - AmqpOutboundEndpoint endpoint = this.buildOutboundEndpoint(name, producerProperties, - buildRabbitTemplate(producerProperties.getExtension())); - return doRegisterProducer(name, outputChannel, endpoint, producerProperties); - } private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) { - RabbitTemplate rabbitTemplate = null; + RabbitTemplate rabbitTemplate; if (properties.isBatchingEnabled()) { BatchingStrategy batchingStrategy = new SimpleBatchingStrategy( properties.getBatchSize(), @@ -432,30 +414,11 @@ public class RabbitMessageChannelBinder extends AbstractBinder doRegisterProducer(final String name, MessageChannel moduleOutputChannel, - AmqpOutboundEndpoint delegate, ExtendedProducerProperties properties) { - return this.doRegisterProducer(name, moduleOutputChannel, delegate, null, properties); - } - - private Binding doRegisterProducer(final String name, MessageChannel moduleOutputChannel, - AmqpOutboundEndpoint delegate, String replyTo, - ExtendedProducerProperties properties) { - Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel); - MessageHandler handler = new SendingHandler(delegate, replyTo, properties); - EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel, handler); - consumer.setBeanFactory(getBeanFactory()); - consumer.setBeanName("outbound." + name); - consumer.afterPropertiesSet(); - DefaultBinding producerBinding = new DefaultBinding<>(name, null, moduleOutputChannel, consumer); - consumer.start(); - return producerBinding; - } - /** * If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original * queue name because we use default exchange routing by queue name for the original message. - * @param queueName The base name for the queue (including the binder prefix, if any). - * @param routingKey The routing key for the queue. + * @param queueName The base name for the queue (including the binder prefix, if any). + * @param routingKey The routing key for the queue. * @param autoBindDlq true if the DLQ should be bound. */ private void autoBindDLQ(final String queueName, String routingKey, String prefix, boolean autoBindDlq) { @@ -474,37 +437,39 @@ public class RabbitMessageChannelBinder extends AbstractBinder messageHeadersList) { - Iterator iterator = messageHeadersList.iterator(); - Map channelsToAck = new HashMap<>(); - while (iterator.hasNext()) { - MessageHeaders messageHeaders = iterator.next(); - if (messageHeaders.containsKey(AmqpHeaders.CHANNEL)) { - Channel channel = (com.rabbitmq.client.Channel) messageHeaders.get(AmqpHeaders.CHANNEL); - Long deliveryTag = (Long) messageHeaders.get(AmqpHeaders.DELIVERY_TAG); - channelsToAck.put(channel, deliveryTag); - } - } - for (Map.Entry entry : channelsToAck.entrySet()) { - try { - ((Channel) entry.getKey()).basicAck(entry.getValue(), true); - } - catch (IOException e) { - this.logger.error("Exception while manually acknowledging " + e); - } - } - } - - private final class SendingHandler extends AbstractMessageHandler implements Lifecycle { - - private final MessageHandler delegate; - - private final String replyTo; - - private final ExtendedProducerProperties producerProperties; - - private final PartitionHandler partitionHandler; - - private SendingHandler(MessageHandler delegate, String replyTo, - ExtendedProducerProperties properties) { - this.delegate = delegate; - this.replyTo = replyTo; - producerProperties = properties; - ConfigurableListableBeanFactory beanFactory = RabbitMessageChannelBinder.this.getBeanFactory(); - this.setBeanFactory(beanFactory); - this.partitionHandler = new PartitionHandler(beanFactory, evaluationContext, partitionSelector, - properties); - } - - @Override - protected void handleMessageInternal(Message message) throws Exception { - MessageValues messageToSend = serializePayloadIfNecessary(message); - - if (this.replyTo != null) { - messageToSend.put(AmqpHeaders.REPLY_TO, this.replyTo); - } - if (producerProperties.isPartitioned()) { - messageToSend.put(PARTITION_HEADER, - this.partitionHandler.determinePartition(message)); - } - - this.delegate.handleMessage(messageToSend.toMessage(getMessageBuilderFactory())); - } - - @Override - public void start() { - if (this.delegate instanceof Lifecycle) { - ((Lifecycle) this.delegate).start(); - } - } - - @Override - public void stop() { - if (this.delegate instanceof Lifecycle) { - ((Lifecycle) this.delegate).stop(); - } - } - - @Override - public boolean isRunning() { - if (this.delegate instanceof Lifecycle) { - return ((Lifecycle) this.delegate).isRunning(); - } - else { - return true; - } - } - - } - - private final class ReceivingHandler extends AbstractReplyProducingMessageHandler { - - private ReceivingHandler() { - super(); - this.setBeanFactory(RabbitMessageChannelBinder.this.getBeanFactory()); - } - - @Override - protected Object handleRequestMessage(Message requestMessage) { - return deserializePayloadIfNecessary(requestMessage).toMessage(getMessageBuilderFactory()); - } - - @Override - protected boolean shouldCopyRequestHeaders() { - /* - * we've already copied the headers so no need for the ARPMH to do it, and we don't want the content-type - * restored if absent. - */ - return false; - } - - } - } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 81eac8856..d4a828e44 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -50,11 +50,12 @@ import org.springframework.cloud.stream.binder.PartitionSelectorStrategy; import org.springframework.cloud.stream.binder.PartitionTestSupport; import org.springframework.cloud.stream.binder.Spy; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; +import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.context.ApplicationContext; +import org.springframework.context.Lifecycle; import org.springframework.expression.spel.standard.SpelExpression; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -109,8 +110,8 @@ public class RabbitBinderTests extends @Test public void testSendAndReceiveBad() throws Exception { RabbitTestBinder binder = getBinder(); - DirectChannel moduleOutputChannel = new DirectChannel(); - DirectChannel moduleInputChannel = new DirectChannel(); + DirectChannel moduleOutputChannel = createBindableChannel("output", new BindingProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", new BindingProperties()); Binding producerBinding = binder.bindProducer("bad.0", moduleOutputChannel, createProducerProperties()); Binding consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel, @@ -137,8 +138,9 @@ public class RabbitBinderTests extends RabbitTestBinder binder = getBinder(); ExtendedConsumerProperties properties = createConsumerProperties(); properties.getExtension().setTransacted(true); - Binding consumerBinding = binder.bindConsumer("props.0", null, new DirectChannel(), properties); - AbstractEndpoint endpoint = extractEndpoint(consumerBinding); + Binding consumerBinding = binder.bindConsumer("props.0", null, + createBindableChannel("input", new BindingProperties()), properties); + Lifecycle endpoint = extractEndpoint(consumerBinding); SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", SimpleMessageListenerContainer.class); assertThat(container.getAcknowledgeMode()).isEqualTo(AcknowledgeMode.AUTO); @@ -171,7 +173,8 @@ public class RabbitBinderTests extends properties.getExtension().setRequeueRejected(false); properties.getExtension().setTxSize(10); properties.setInstanceIndex(0); - consumerBinding = binder.bindConsumer("props.0", "test", new DirectChannel(), properties); + consumerBinding = binder.bindConsumer("props.0", "test", createBindableChannel("input", new BindingProperties()), + properties); endpoint = extractEndpoint(consumerBinding); container = verifyContainer(endpoint); @@ -185,39 +188,42 @@ public class RabbitBinderTests extends @Test public void testProducerProperties() throws Exception { RabbitTestBinder binder = getBinder(); - Binding producerBinding = binder.bindProducer("props.0", new DirectChannel(), + Binding producerBinding = binder.bindProducer("props.0", + createBindableChannel("input", new BindingProperties()), createProducerProperties()); @SuppressWarnings("unchecked") - AbstractEndpoint endpoint = extractEndpoint(producerBinding); - MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode", + Lifecycle endpoint = extractEndpoint(producerBinding); + MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode", MessageDeliveryMode.class); assertThat(mode).isEqualTo(MessageDeliveryMode.PERSISTENT); List requestHeaders = TestUtils.getPropertyValue(endpoint, - "handler.delegate.headerMapper.requestHeaderMatcher.matchers", List.class); + "headerMapper.requestHeaderMatcher.matchers", List.class); assertThat(requestHeaders).hasSize(2); producerBinding.unbind(); assertThat(endpoint.isRunning()).isFalse(); - assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class)) + assertThat(TestUtils.getPropertyValue(endpoint, "amqpTemplate.transactional", Boolean.class)) .isFalse(); - ExtendedProducerProperties properties = createProducerProperties(); - properties.getExtension().setPrefix("foo."); - properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); - properties.getExtension().setRequestHeaderPatterns(new String[] { "foo" }); - properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'")); - properties.setPartitionKeyExtractorClass(TestPartitionKeyExtractorClass.class); - properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0")); - properties.setPartitionSelectorClass(TestPartitionSelectorClass.class); - properties.setPartitionCount(1); - properties.getExtension().setTransacted(true); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.getExtension().setPrefix("foo."); + producerProperties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); + producerProperties.getExtension().setRequestHeaderPatterns(new String[] { "foo" }); + producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("'foo'")); + producerProperties.setPartitionKeyExtractorClass(TestPartitionKeyExtractorClass.class); + producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("0")); + producerProperties.setPartitionSelectorClass(TestPartitionSelectorClass.class); + producerProperties.setPartitionCount(1); + producerProperties.getExtension().setTransacted(true); - producerBinding = binder.bindProducer("props.0", new DirectChannel(), properties); + BindingProperties producerBindingProperties = createProducerBindingProperties(producerProperties); + producerBinding = binder.bindProducer("props.0", createBindableChannel("output", producerBindingProperties), + producerProperties); endpoint = extractEndpoint(producerBinding); - assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression", SpelExpression.class) + assertThat(TestUtils.getPropertyValue(endpoint, "routingKeyExpression", SpelExpression.class) .getExpressionString()).isEqualTo("'props.0-' + headers['partition']"); - mode = TestUtils.getPropertyValue(endpoint, "handler.delegate.defaultDeliveryMode", MessageDeliveryMode.class); + mode = TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode", MessageDeliveryMode.class); assertThat(mode).isEqualTo(MessageDeliveryMode.NON_PERSISTENT); - assertThat(TestUtils.getPropertyValue(endpoint, "handler.delegate.amqpTemplate.transactional", Boolean.class)) + assertThat(TestUtils.getPropertyValue(endpoint, "amqpTemplate.transactional", Boolean.class)) .isTrue(); verifyFooRequestProducer(endpoint); @@ -231,13 +237,13 @@ public class RabbitBinderTests extends RabbitTestBinder binder = getBinder(); - ExtendedConsumerProperties properties = createConsumerProperties(); - properties.getExtension().setPrefix(TEST_PREFIX); - properties.getExtension().setAutoBindDlq(true); - properties.getExtension().setDurableSubscription(true); - properties.setMaxAttempts(1); // disable retry - properties.getExtension().setRequeueRejected(false); - DirectChannel moduleInputChannel = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix(TEST_PREFIX); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.getExtension().setDurableSubscription(true); + consumerProperties.setMaxAttempts(1); // disable retry + consumerProperties.getExtension().setRequeueRejected(false); + DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); moduleInputChannel.setBeanName("durableTest"); moduleInputChannel.subscribe(new MessageHandler() { @@ -248,7 +254,7 @@ public class RabbitBinderTests extends }); Binding consumerBinding = binder.bindConsumer("durabletest.0", "tgroup", moduleInputChannel, - properties); + consumerProperties); RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); template.convertAndSend(TEST_PREFIX + "durabletest.0", "", "foo"); @@ -273,13 +279,14 @@ public class RabbitBinderTests extends RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); RabbitTestBinder binder = getBinder(); - ExtendedConsumerProperties properties = createConsumerProperties(); - properties.getExtension().setPrefix(TEST_PREFIX); - properties.getExtension().setAutoBindDlq(true); - properties.getExtension().setDurableSubscription(false); - properties.setMaxAttempts(1); // disable retry - properties.getExtension().setRequeueRejected(false); - DirectChannel moduleInputChannel = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix(TEST_PREFIX); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.getExtension().setDurableSubscription(false); + consumerProperties.setMaxAttempts(1); // disable retry + consumerProperties.getExtension().setRequeueRejected(false); + BindingProperties bindingProperties = createConsumerBindingProperties(consumerProperties); + DirectChannel moduleInputChannel = createBindableChannel("input", bindingProperties); moduleInputChannel.setBeanName("nondurabletest"); moduleInputChannel.subscribe(new MessageHandler() { @@ -290,7 +297,7 @@ public class RabbitBinderTests extends }); Binding consumerBinding = binder.bindConsumer("nondurabletest.0", "tgroup", moduleInputChannel, - properties); + consumerProperties); consumerBinding.unbind(); assertThat(admin.getQueueProperties(TEST_PREFIX + "nondurabletest.0.dlq")).isNull(); @@ -299,13 +306,14 @@ public class RabbitBinderTests extends @Test public void testAutoBindDLQ() throws Exception { RabbitTestBinder binder = getBinder(); - ExtendedConsumerProperties properties = createConsumerProperties(); - properties.getExtension().setPrefix(TEST_PREFIX); - properties.getExtension().setAutoBindDlq(true); - properties.setMaxAttempts(1); // disable retry - properties.getExtension().setRequeueRejected(false); - properties.getExtension().setDurableSubscription(true); - DirectChannel moduleInputChannel = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix(TEST_PREFIX); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.setMaxAttempts(1); // disable retry + consumerProperties.getExtension().setRequeueRejected(false); + consumerProperties.getExtension().setDurableSubscription(true); + BindingProperties bindingProperties = createConsumerBindingProperties(consumerProperties); + DirectChannel moduleInputChannel = createBindableChannel("input", bindingProperties); moduleInputChannel.setBeanName("dlqTest"); moduleInputChannel.subscribe(new MessageHandler() { @@ -316,7 +324,7 @@ public class RabbitBinderTests extends }); Binding consumerBinding = binder.bindConsumer("dlqtest", "default", moduleInputChannel, - properties); + consumerProperties); RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo"); @@ -352,13 +360,13 @@ public class RabbitBinderTests extends properties.getExtension().setRequeueRejected(false); properties.setPartitioned(true); properties.setInstanceIndex(0); - DirectChannel input0 = new DirectChannel(); + DirectChannel input0 = createBindableChannel("input", createConsumerBindingProperties(properties)); input0.setBeanName("test.input0DLQ"); Binding input0Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input0, properties); Binding defaultConsumerBinding1 = binder.bindConsumer("partDLQ.0", "default", new QueueChannel(), properties); properties.setInstanceIndex(1); - DirectChannel input1 = new DirectChannel(); + DirectChannel input1 = createBindableChannel("input1", createConsumerBindingProperties(properties)); input1.setBeanName("test.input1DLQ"); Binding input1Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input1, properties); Binding defaultConsumerBinding2 = binder.bindConsumer("partDLQ.0", "default", @@ -370,7 +378,8 @@ public class RabbitBinderTests extends producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class); producerProperties.setPartitionSelectorClass(PartitionTestSupport.class); producerProperties.setPartitionCount(2); - DirectChannel output = new DirectChannel(); + BindingProperties bindingProperties = createProducerBindingProperties(producerProperties); + DirectChannel output = createBindableChannel("output", bindingProperties); output.setBeanName("test.output"); Binding outputBinding = binder.bindProducer("partDLQ.0", output, producerProperties); @@ -440,7 +449,7 @@ public class RabbitBinderTests extends properties.setPartitionKeyExtractorClass(PartitionTestSupport.class); properties.setPartitionSelectorClass(PartitionTestSupport.class); properties.setPartitionCount(2); - DirectChannel output = new DirectChannel(); + DirectChannel output = createBindableChannel("output", createProducerBindingProperties(properties)); output.setBeanName("test.output"); Binding outputBinding = binder.bindProducer("partDLQ.1", output, properties); @@ -451,14 +460,14 @@ public class RabbitBinderTests extends consumerProperties.getExtension().setRequeueRejected(false); consumerProperties.setPartitioned(true); consumerProperties.setInstanceIndex(0); - DirectChannel input0 = new DirectChannel(); + DirectChannel input0 = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); input0.setBeanName("test.input0DLQ"); Binding input0Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input0, consumerProperties); Binding defaultConsumerBinding1 = binder.bindConsumer("partDLQ.1", "defaultConsumer", new QueueChannel(), consumerProperties); consumerProperties.setInstanceIndex(1); - DirectChannel input1 = new DirectChannel(); + DirectChannel input1 = createBindableChannel("input1", createConsumerBindingProperties(consumerProperties)); input1.setBeanName("test.input1DLQ"); Binding input1Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input1, consumerProperties); @@ -531,14 +540,14 @@ public class RabbitBinderTests extends admin.declareQueue(queue); RabbitTestBinder binder = getBinder(); - ExtendedConsumerProperties properties = createConsumerProperties(); - properties.getExtension().setPrefix(TEST_PREFIX); - properties.getExtension().setAutoBindDlq(true); - properties.getExtension().setRepublishToDlq(true); - properties.setMaxAttempts(1); // disable retry - properties.getExtension().setRequeueRejected(false); - properties.getExtension().setDurableSubscription(true); - DirectChannel moduleInputChannel = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.getExtension().setPrefix(TEST_PREFIX); + consumerProperties.getExtension().setAutoBindDlq(true); + consumerProperties.getExtension().setRepublishToDlq(true); + consumerProperties.setMaxAttempts(1); // disable retry + consumerProperties.getExtension().setRequeueRejected(false); + consumerProperties.getExtension().setDurableSubscription(true); + DirectChannel moduleInputChannel = createBindableChannel("input", createConsumerBindingProperties(consumerProperties)); moduleInputChannel.setBeanName("dlqPubTest"); moduleInputChannel.subscribe(new MessageHandler() { @@ -549,7 +558,7 @@ public class RabbitBinderTests extends }); Binding consumerBinding = binder.bindConsumer("dlqpubtest", "default", moduleInputChannel, - properties); + consumerProperties); RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); template.convertAndSend("", TEST_PREFIX + "dlqpubtest.default", "foo"); @@ -574,18 +583,18 @@ public class RabbitBinderTests extends public void testBatchingAndCompression() throws Exception { RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); RabbitTestBinder binder = getBinder(); - ExtendedProducerProperties properties = createProducerProperties(); - properties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); - properties.getExtension().setBatchingEnabled(true); - properties.getExtension().setBatchSize(2); - properties.getExtension().setBatchBufferLimit(100000); - properties.getExtension().setBatchTimeout(30000); - properties.getExtension().setCompress(true); - properties.setRequiredGroups("default"); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); + producerProperties.getExtension().setBatchingEnabled(true); + producerProperties.getExtension().setBatchSize(2); + producerProperties.getExtension().setBatchBufferLimit(100000); + producerProperties.getExtension().setBatchTimeout(30000); + producerProperties.getExtension().setCompress(true); + producerProperties.setRequiredGroups("default"); - DirectChannel output = new DirectChannel(); + DirectChannel output = createBindableChannel("input", createProducerBindingProperties(producerProperties)); output.setBeanName("batchingProducer"); - Binding producerBinding = binder.bindProducer("batching.0", output, properties); + Binding producerBinding = binder.bindProducer("batching.0", output, producerProperties); Log logger = spy(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor.logger", Log.class)); new DirectFieldAccessor(TestUtils.getPropertyValue(binder, "binder.compressingPostProcessor")) @@ -637,12 +646,12 @@ public class RabbitBinderTests extends RabbitMessageChannelBinder rabbitBinder = new RabbitMessageChannelBinder(cf, new RabbitProperties()); RabbitTestBinder binder = new RabbitTestBinder(cf, rabbitBinder); - ExtendedProducerProperties properties = createProducerProperties(); - properties.getExtension().setPrefix("latebinder."); - properties.getExtension().setAutoBindDlq(true); + ExtendedProducerProperties producerProperties = createProducerProperties(); + producerProperties.getExtension().setPrefix("latebinder."); + producerProperties.getExtension().setAutoBindDlq(true); - MessageChannel moduleOutputChannel = new DirectChannel(); - Binding late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, properties); + MessageChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties)); + Binding late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, producerProperties); QueueChannel moduleInputChannel = new QueueChannel(); ExtendedConsumerProperties rabbitConsumerProperties = createConsumerProperties(); @@ -650,13 +659,13 @@ public class RabbitBinderTests extends Binding late0ConsumerBinding = binder.bindConsumer("late.0", "test", moduleInputChannel, rabbitConsumerProperties); - properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload.equals('0') ? 0 : 1")); - properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); - properties.setPartitionCount(2); + producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload.equals('0') ? 0 : 1")); + producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); + producerProperties.setPartitionCount(2); - MessageChannel partOutputChannel = new DirectChannel(); + MessageChannel partOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties)); Binding partlate0ProducerBinding = binder.bindProducer("partlate.0", partOutputChannel, - properties); + producerProperties); QueueChannel partInputChannel0 = new QueueChannel(); QueueChannel partInputChannel1 = new QueueChannel(); @@ -673,7 +682,8 @@ public class RabbitBinderTests extends ExtendedProducerProperties noDlqProducerProperties = createProducerProperties(); noDlqProducerProperties.getExtension().setPrefix("latebinder."); - MessageChannel noDLQOutputChannel = new DirectChannel(); + MessageChannel noDLQOutputChannel = createBindableChannel("output", + createProducerBindingProperties(noDlqProducerProperties)); Binding noDlqProducerBinding = binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel, noDlqProducerProperties); @@ -683,7 +693,7 @@ public class RabbitBinderTests extends Binding noDlqConsumerBinding = binder.bindConsumer("lateNoDLQ.0", "test", noDLQInputChannel, noDlqConsumerProperties); - MessageChannel outputChannel = new DirectChannel(); + MessageChannel outputChannel = createBindableChannel("output", createProducerBindingProperties(noDlqProducerProperties)); Binding pubSubProducerBinding = binder.bindProducer("latePubSub", outputChannel, noDlqProducerProperties); QueueChannel pubSubInputChannel = new QueueChannel(); @@ -743,7 +753,7 @@ public class RabbitBinderTests extends this.rabbitAvailableRule.getResource().destroy(); } - private SimpleMessageListenerContainer verifyContainer(AbstractEndpoint endpoint) { + private SimpleMessageListenerContainer verifyContainer(Lifecycle endpoint) { SimpleMessageListenerContainer container; Advice retry; container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", @@ -770,16 +780,16 @@ public class RabbitBinderTests extends return container; } - private void verifyFooRequestProducer(AbstractEndpoint endpoint) { + private void verifyFooRequestProducer(Lifecycle endpoint) { List requestMatchers = TestUtils.getPropertyValue(endpoint, - "handler.delegate.headerMapper.requestHeaderMatcher.matchers", List.class); + "headerMapper.requestHeaderMatcher.matchers", List.class); assertThat(requestMatchers).hasSize(1); assertThat(TestUtils.getPropertyValue(requestMatchers.get(0), "pattern")).isEqualTo("foo"); } @Override - protected String getEndpointRouting(AbstractEndpoint endpoint) { - return TestUtils.getPropertyValue(endpoint, "handler.delegate.routingKeyExpression", SpelExpression.class) + protected String getEndpointRouting(Object endpoint) { + return TestUtils.getPropertyValue(endpoint, "routingKeyExpression", SpelExpression.class) .getExpressionString(); } @@ -788,12 +798,6 @@ public class RabbitBinderTests extends return name; } - @Override - protected String getPubSubEndpointRouting(AbstractEndpoint endpoint) { - return TestUtils.getPropertyValue(endpoint, "handler.delegate.exchangeNameExpression", SpelExpression.class) - .getExpressionString(); - } - @Override protected String getClassUnderTestName() { return CLASS_UNDER_TEST_NAME; diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index e12e35c07..eca55df22 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -41,9 +41,9 @@ import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; +import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; import org.springframework.cloud.stream.binding.ChannelBindingService; import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.messaging.MessageChannel; @@ -120,7 +120,7 @@ public class RabbitBinderModuleTests { Map> producerBindings = (Map>) TestUtils .getPropertyValue(channelBindingService, "producerBindings"); Binding outputBinding = producerBindings.get("output"); - assertThat(TestUtils.getPropertyValue(outputBinding, "endpoint.handler.delegate.amqpTemplate.transactional", + assertThat(TestUtils.getPropertyValue(outputBinding, "endpoint.amqpTemplate.transactional", Boolean.class)).isTrue(); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor