SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Changes for intro of AbstractMessageChannelBinder

Browse Source
This commit is contained in:
Marius Bogoevici
2016-06-22 20:06:11 -04:00
committed by Mark Fisher
parent ec207b6f2d
commit e2e02c1693
4 changed files with 273 additions and 394 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
pom.xml
View File

@@ -22,6 +22,11 @@
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
<version>1.1.0.BUILD-SNAPSHOT</version>
</dependency>
</dependencies>
</dependencyManagement>
<modules>
@@ -29,6 +34,19 @@
<module>spring-cloud-starter-stream-rabbit</module>
<module>spring-cloud-stream-binder-rabbit-test-support</module>
</modules>
<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<redirectTestOutputToFile>true</redirectTestOutputToFile>
</configuration>
</plugin>
</plugins>
</pluginManagement>
</build>
<profiles>
<profile>
<id>spring</id>

447
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -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<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>,
ExtendedProducerProperties<RabbitProducerProperties>> implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
public class RabbitMessageChannelBinder
extends AbstractMessageChannelBinder<ExtendedConsumerProperties<RabbitConsumerProperties>,
ExtendedProducerProperties<RabbitProducerProperties>, Queue>
implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties> {
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<MessageChannel, E
}
};
private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
private final RabbitAdmin rabbitAdmin;
private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext();
@@ -142,6 +124,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties();
public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) {
super(true, new String[0]);
Assert.notNull(connectionFactory, "connectionFactory must not be null");
Assert.notNull(rabbitProperties, "rabbitProperties must not be null");
this.connectionFactory = connectionFactory;
@@ -192,89 +175,35 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
&& addresses.length == this.nodes.length,
"'addresses', 'adminAddresses', and 'nodes' properties must have equal length");
this.connectionFactory = new LocalizedQueueConnectionFactory(this.connectionFactory, addresses,
this.adminAddresses, this.nodes, rabbitProperties.getVirtualHost(),
this.adminAddresses, this.nodes, this.rabbitProperties.getVirtualHost(),
this.rabbitProperties.getUsername(), this.rabbitProperties.getPassword(),
this.rabbitProperties.getSsl().isEnabled(), this.rabbitProperties.getSsl().getKeyStore(),
this.rabbitProperties.getSsl().getTrustStore(), this.rabbitProperties.getSsl().getKeyStorePassword(),
this.rabbitProperties.getSsl().getTrustStore(),
this.rabbitProperties.getSsl().getKeyStorePassword(),
this.rabbitProperties.getSsl().getTrustStorePassword());
}
}
@Override
public RabbitConsumerProperties getExtendedConsumerProperties(String channelName) {
return extendedBindingProperties.getExtendedConsumerProperties(channelName);
return this.extendedBindingProperties.getExtendedConsumerProperties(channelName);
}
@Override
public RabbitProducerProperties getExtendedProducerProperties(String channelName) {
return extendedBindingProperties.getExtendedProducerProperties(channelName);
return this.extendedBindingProperties.getExtendedProducerProperties(channelName);
}
@Override
public Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel,
protected MessageProducer createConsumerEndpoint(String name, String group, Queue destination,
ExtendedConsumerProperties<RabbitConsumerProperties> 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<MessageChannel> binding = doRegisterConsumer(baseQueueName, group, inputChannel, queue, properties);
if (durable) {
autoBindDLQ(applyPrefix(prefix, baseQueueName), queueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
}
return binding;
}
private Map<String, Object> queueArgs(String queueName, String prefix, boolean bindDlq) {
Map<String, Object> 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<MessageChannel> doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue,
final ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
DefaultBinding<MessageChannel> consumerBinding;
SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(
this.connectionFactory);
listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
@@ -290,84 +219,149 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
listenerContainer.setTxSize(properties.getExtension().getTxSize());
listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(queue.getName() + "-"));
listenerContainer.setQueues(queue);
int maxAttempts = properties.getMaxAttempts();
if (maxAttempts > 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<MessageChannel>(name, group, moduleInputChannel, adapter) {
@Override
protected void afterUnbind() {
cleanAutoDeclareContext(properties.getExtension().getPrefix(), name);
return adapter;
}
@Override
protected void afterUnbindConsumer(String name, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
cleanAutoDeclareContext(consumerProperties.getExtension().getPrefix(), baseQueueName(name, group));
}
@Override
protected Queue createConsumerDestinationIfNecessary(String name, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> 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<String, Object> queueArgs(String queueName, String prefix, boolean bindDlq) {
Map<String, Object> 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<RabbitProducerProperties> properties,
RabbitTemplate rabbitTemplate) {
String prefix = properties.getExtension().getPrefix();
String exchangeName = applyPrefix(prefix, name);
@Override
protected void createProducerDestinationIfNecessary(String name,
ExtendedProducerProperties<RabbitProducerProperties> 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<RabbitProducerProperties> 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<MessageChannel, E
String partitionSuffix = "-" + i;
String partitionQueueName = baseQueueName + partitionSuffix;
Queue queue = new Queue(partitionQueueName, true, false, false,
queueArgs(partitionQueueName, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq()));
queueArgs(partitionQueueName, properties.getExtension().getPrefix(),
properties.getExtension().isAutoBindDlq()));
declareQueue(queue.getName(), queue);
autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties.getExtension().getPrefix(), properties.getExtension().isAutoBindDlq());
declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(name + partitionSuffix));
autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties.getExtension().getPrefix(),
properties.getExtension().isAutoBindDlq());
declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange)
.with(destination + partitionSuffix));
}
}
}
configureOutboundHandler(endpoint, properties);
DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper();
mapper.setRequestHeaderNames(properties.getExtension().getRequestHeaderPatterns());
mapper.setReplyHeaderNames(properties.getExtension().getReplyHeaderPatterns());
endpoint.setHeaderMapper(mapper);
endpoint.setDefaultDeliveryMode(properties.getExtension().getDeliveryMode());
endpoint.setBeanFactory(this.getBeanFactory());
endpoint.afterPropertiesSet();
return endpoint;
}
private void configureOutboundHandler(AmqpOutboundEndpoint handler,
ExtendedProducerProperties<RabbitProducerProperties> 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<MessageChannel> doBindProducer(String name, MessageChannel outputChannel,
ExtendedProducerProperties<RabbitProducerProperties> 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<MessageChannel, E
return rabbitTemplate;
}
private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel,
AmqpOutboundEndpoint delegate, ExtendedProducerProperties<RabbitProducerProperties> properties) {
return this.doRegisterProducer(name, moduleOutputChannel, delegate, null, properties);
}
private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel,
AmqpOutboundEndpoint delegate, String replyTo,
ExtendedProducerProperties<RabbitProducerProperties> 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<MessageChannel> 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<MessageChannel, E
}
}
public void declareQueue(String beanName, Queue queue) {
private void declareQueue(String beanName, Queue queue) {
try {
this.rabbitAdmin.declareQueue(queue);
}
catch (AmqpConnectException e) {
if (logger.isDebugEnabled()) {
logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
if (this.logger.isDebugEnabled()) {
this.logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
}
}
addToAutoDeclareContext(beanName, queue);
}
public void declareExchange(final String rootName, final Exchange exchange) {
private void declareExchange(final String rootName, final Exchange exchange) {
try {
this.rabbitAdmin.declareExchange(exchange);
}
catch (AmqpConnectException e) {
if (logger.isDebugEnabled()) {
logger.debug("Declaration of exchange: " + exchange.getName() + " deferred - connection not available");
if (this.logger.isDebugEnabled()) {
this.logger.debug(
"Declaration of exchange: " + exchange.getName() + " deferred - connection not available");
}
}
addToAutoDeclareContext(rootName + ".exchange", exchange);
}
public void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) {
private void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) {
try {
this.rabbitAdmin.declareBinding(binding);
}
catch (AmqpConnectException e) {
if (logger.isDebugEnabled()) {
logger.debug("Declaration of binding: " + rootName + ".binding deferred - connection not available");
if (this.logger.isDebugEnabled()) {
this.logger.debug(
"Declaration of binding: " + rootName + ".binding deferred - connection not available");
}
}
addToAutoDeclareContext(rootName + ".binding", binding);
@@ -522,7 +487,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
}
}
public void cleanAutoDeclareContext(String prefix, String name) {
private void cleanAutoDeclareContext(String prefix, String name) {
synchronized (this.autoDeclareContext) {
removeSingleton(applyPrefix(prefix, name) + ".binding");
removeSingleton(applyPrefix(prefix, name));
@@ -540,112 +505,4 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel, E
}
}
}
@Override
public void doManualAck(LinkedList<MessageHeaders> messageHeadersList) {
Iterator<MessageHeaders> iterator = messageHeadersList.iterator();
Map<Object, Long> 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<Object, Long> 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<RabbitProducerProperties> producerProperties;
private final PartitionHandler partitionHandler;
private SendingHandler(MessageHandler delegate, String replyTo,
ExtendedProducerProperties<RabbitProducerProperties> 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;
}
}
}

198
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -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<MessageChannel> producerBinding = binder.bindProducer("bad.0", moduleOutputChannel,
createProducerProperties());
Binding<MessageChannel> consumerBinding = binder.bindConsumer("bad.0", "test", moduleInputChannel,
@@ -137,8 +138,9 @@ public class RabbitBinderTests extends
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setTransacted(true);
Binding<MessageChannel> consumerBinding = binder.bindConsumer("props.0", null, new DirectChannel(), properties);
AbstractEndpoint endpoint = extractEndpoint(consumerBinding);
Binding<MessageChannel> 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<MessageChannel> producerBinding = binder.bindProducer("props.0", new DirectChannel(),
Binding<MessageChannel> 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<RabbitProducerProperties> 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<RabbitProducerProperties> 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<RabbitConsumerProperties> 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<RabbitConsumerProperties> 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<MessageChannel> 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<RabbitConsumerProperties> 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<RabbitConsumerProperties> 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<MessageChannel> 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<RabbitConsumerProperties> 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<RabbitConsumerProperties> 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<MessageChannel> 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<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input0, properties);
Binding<MessageChannel> 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<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.0", "dlqPartGrp", input1, properties);
Binding<MessageChannel> 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<MessageChannel> 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<MessageChannel> 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<MessageChannel> input0Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input0,
consumerProperties);
Binding<MessageChannel> 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<MessageChannel> input1Binding = binder.bindConsumer("partDLQ.1", "dlqPartGrp", input1,
consumerProperties);
@@ -531,14 +540,14 @@ public class RabbitBinderTests extends
admin.declareQueue(queue);
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> 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<RabbitConsumerProperties> 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<MessageChannel> 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<RabbitProducerProperties> 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<RabbitProducerProperties> 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<MessageChannel> producerBinding = binder.bindProducer("batching.0", output, properties);
Binding<MessageChannel> 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<RabbitProducerProperties> properties = createProducerProperties();
properties.getExtension().setPrefix("latebinder.");
properties.getExtension().setAutoBindDlq(true);
ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
producerProperties.getExtension().setPrefix("latebinder.");
producerProperties.getExtension().setAutoBindDlq(true);
MessageChannel moduleOutputChannel = new DirectChannel();
Binding<MessageChannel> late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, properties);
MessageChannel moduleOutputChannel = createBindableChannel("output", createProducerBindingProperties(producerProperties));
Binding<MessageChannel> late0ProducerBinding = binder.bindProducer("late.0", moduleOutputChannel, producerProperties);
QueueChannel moduleInputChannel = new QueueChannel();
ExtendedConsumerProperties<RabbitConsumerProperties> rabbitConsumerProperties = createConsumerProperties();
@@ -650,13 +659,13 @@ public class RabbitBinderTests extends
Binding<MessageChannel> 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<MessageChannel> 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<RabbitProducerProperties> noDlqProducerProperties = createProducerProperties();
noDlqProducerProperties.getExtension().setPrefix("latebinder.");
MessageChannel noDLQOutputChannel = new DirectChannel();
MessageChannel noDLQOutputChannel = createBindableChannel("output",
createProducerBindingProperties(noDlqProducerProperties));
Binding<MessageChannel> noDlqProducerBinding = binder.bindProducer("lateNoDLQ.0", noDLQOutputChannel,
noDlqProducerProperties);
@@ -683,7 +693,7 @@ public class RabbitBinderTests extends
Binding<MessageChannel> noDlqConsumerBinding = binder.bindConsumer("lateNoDLQ.0", "test", noDLQInputChannel,
noDlqConsumerProperties);
MessageChannel outputChannel = new DirectChannel();
MessageChannel outputChannel = createBindableChannel("output", createProducerBindingProperties(noDlqProducerProperties));
Binding<MessageChannel> 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;

4
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java
View File

@@ -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<String, Binding<MessageChannel>> producerBindings = (Map<String, Binding<MessageChannel>>) TestUtils
.getPropertyValue(channelBindingService, "producerBindings");
Binding<MessageChannel> 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