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

Merged 2.2.x to master

Browse Source
This commit is contained in:
Oleg Zhurakousky
2019-02-04 18:14:12 +01:00
parent ebe3b30904
commit 351f7f5b4c
21 changed files with 1221 additions and 804 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
pom.xml
View File

@@ -79,10 +79,24 @@
</plugins>
</pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>${maven-compiler-plugin.version}</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
<compilerArgument>-parameters</compilerArgument>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
</plugin>
<plugin>
<groupId>io.spring.javaformat</groupId>
<artifactId>spring-javaformat-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
<profiles>

2
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitAdminException.java
View File

@@ -16,9 +16,9 @@
package org.springframework.cloud.stream.binder.rabbit.admin;
/**
* Exceptions thrown while interfacing with the RabbitMQ admin plugin.
*
* @author Gary Russell
* @since 1.2
*/

88
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java
View File

@@ -30,9 +30,10 @@ import org.springframework.cloud.stream.binder.BindingCleaner;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
/**
* Implementation of {@link org.springframework.cloud.stream.binder.BindingCleaner} for the {@code RabbitBinder}.
* Implementation of {@link org.springframework.cloud.stream.binder.BindingCleaner} for
* the {@code RabbitBinder}.
*
* @author Gary Russell
* @author David Turanski
* @since 1.2
@@ -47,28 +48,28 @@ public class RabbitBindingCleaner implements BindingCleaner {
@Override
public Map<String, List<String>> clean(String entity, boolean isJob) {
return clean("http://localhost:15672", "guest", "guest", "/", BINDER_PREFIX, entity, isJob);
}
public Map<String, List<String>> clean(String adminUri, String user, String pw, String vhost,
String binderPrefix, String entity, boolean isJob) {
return doClean(
adminUri == null ? "http://localhost:15672" : adminUri,
user == null ? "guest" : user,
pw == null ? "guest" : pw,
vhost == null ? "/" : vhost,
binderPrefix == null ? BINDER_PREFIX : binderPrefix,
return clean("http://localhost:15672", "guest", "guest", "/", BINDER_PREFIX,
entity, isJob);
}
private Map<String, List<String>> doClean(String adminUri, String user, String pw, String vhost,
String binderPrefix, String entity, boolean isJob) {
RestTemplate restTemplate = RabbitManagementUtils.buildRestTemplate(adminUri, user, pw);
List<String> removedQueues = isJob
? null
public Map<String, List<String>> clean(String adminUri, String user, String pw,
String vhost, String binderPrefix, String entity, boolean isJob) {
return doClean(adminUri == null ? "http://localhost:15672" : adminUri,
user == null ? "guest" : user, pw == null ? "guest" : pw,
vhost == null ? "/" : vhost,
binderPrefix == null ? BINDER_PREFIX : binderPrefix, entity, isJob);
}
private Map<String, List<String>> doClean(String adminUri, String user, String pw,
String vhost, String binderPrefix, String entity, boolean isJob) {
RestTemplate restTemplate = RabbitManagementUtils.buildRestTemplate(adminUri,
user, pw);
List<String> removedQueues = isJob ? null
: findStreamQueues(adminUri, vhost, binderPrefix, entity, restTemplate);
List<String> removedExchanges = findExchanges(adminUri, vhost, binderPrefix, entity, restTemplate);
// Delete the queues in reverse order to enable re-running after a partial success.
List<String> removedExchanges = findExchanges(adminUri, vhost, binderPrefix,
entity, restTemplate);
// Delete the queues in reverse order to enable re-running after a partial
// success.
// The queue search above starts with 0 and terminates on a not found.
for (int i = removedQueues.size() - 1; i >= 0; i--) {
String queueName = removedQueues.get(i);
@@ -100,9 +101,10 @@ public class RabbitBindingCleaner implements BindingCleaner {
return results;
}
private List<String> findStreamQueues(String adminUri, String vhost, String binderPrefix, String stream,
RestTemplate restTemplate) {
String queueNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, stream));
private List<String> findStreamQueues(String adminUri, String vhost,
String binderPrefix, String stream, RestTemplate restTemplate) {
String queueNamePrefix = adjustPrefix(
AbstractBinder.applyPrefix(binderPrefix, stream));
List<Map<String, Object>> queues = listAllQueues(adminUri, vhost, restTemplate);
List<String> removedQueues = new ArrayList<>();
for (Map<String, Object> queue : queues) {
@@ -115,10 +117,10 @@ public class RabbitBindingCleaner implements BindingCleaner {
return removedQueues;
}
private List<Map<String, Object>> listAllQueues(String adminUri, String vhost, RestTemplate restTemplate) {
private List<Map<String, Object>> listAllQueues(String adminUri, String vhost,
RestTemplate restTemplate) {
URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("queues", "{vhost}")
.buildAndExpand(vhost).encode().toUri();
.pathSegment("queues", "{vhost}").buildAndExpand(vhost).encode().toUri();
List<Map<String, Object>> queues = restTemplate.getForObject(uri, List.class);
return queues;
}
@@ -138,51 +140,57 @@ public class RabbitBindingCleaner implements BindingCleaner {
}
}
private List<String> findExchanges(String adminUri, String vhost, String binderPrefix, String entity,
RestTemplate restTemplate) {
private List<String> findExchanges(String adminUri, String vhost, String binderPrefix,
String entity, RestTemplate restTemplate) {
List<String> removedExchanges = new ArrayList<>();
URI uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("exchanges", "{vhost}")
.buildAndExpand(vhost).encode().toUri();
.pathSegment("exchanges", "{vhost}").buildAndExpand(vhost).encode()
.toUri();
List<Map<String, Object>> exchanges = restTemplate.getForObject(uri, List.class);
String exchangeNamePrefix = adjustPrefix(AbstractBinder.applyPrefix(binderPrefix, entity));
String exchangeNamePrefix = adjustPrefix(
AbstractBinder.applyPrefix(binderPrefix, entity));
for (Map<String, Object> exchange : exchanges) {
String exchangeName = (String) exchange.get("name");
if (exchangeName.startsWith(exchangeNamePrefix)) {
uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("exchanges", "{vhost}", "{name}", "bindings", "source")
.pathSegment("exchanges", "{vhost}", "{name}", "bindings",
"source")
.buildAndExpand(vhost, exchangeName).encode().toUri();
List<Map<String, Object>> bindings = restTemplate.getForObject(uri, List.class);
List<Map<String, Object>> bindings = restTemplate.getForObject(uri,
List.class);
if (hasNoForeignBindings(bindings, exchangeNamePrefix)) {
uri = UriComponentsBuilder.fromUriString(adminUri + "/api")
.pathSegment("exchanges", "{vhost}", "{name}", "bindings", "destination")
.pathSegment("exchanges", "{vhost}", "{name}", "bindings",
"destination")
.buildAndExpand(vhost, exchangeName).encode().toUri();
bindings = restTemplate.getForObject(uri, List.class);
if (bindings.size() == 0) {
removedExchanges.add((String) exchange.get("name"));
}
else {
throw new RabbitAdminException("Cannot delete exchange " + exchangeName
+ "; it is a destination: " + bindings);
throw new RabbitAdminException("Cannot delete exchange "
+ exchangeName + "; it is a destination: " + bindings);
}
}
else {
throw new RabbitAdminException("Cannot delete exchange " + exchangeName + "; it has bindings: "
+ bindings);
throw new RabbitAdminException("Cannot delete exchange "
+ exchangeName + "; it has bindings: " + bindings);
}
}
}
return removedExchanges;
}
private boolean hasNoForeignBindings(List<Map<String, Object>> bindings, String exchangeNamePrefix) {
private boolean hasNoForeignBindings(List<Map<String, Object>> bindings,
String exchangeNamePrefix) {
if (bindings.size() == 0) {
return true;
}
boolean noForeign = true;
for (Map<String, Object> binding : bindings) {
if (!("queue".equals(binding.get("destination_type")))
|| !((String) binding.get("destination")).startsWith(exchangeNamePrefix)) {
|| !((String) binding.get("destination"))
.startsWith(exchangeNamePrefix)) {
noForeign = false;
break;
}

31
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java
View File

@@ -44,13 +44,16 @@ import org.springframework.web.client.RestTemplate;
*/
public abstract class RabbitManagementUtils {
public static RestTemplate buildRestTemplate(String adminUri, String user, String password) {
public static RestTemplate buildRestTemplate(String adminUri, String user,
String password) {
BasicCredentialsProvider credsProvider = new BasicCredentialsProvider();
credsProvider.setCredentials(
new AuthScope(AuthScope.ANY_HOST, AuthScope.ANY_PORT),
new UsernamePasswordCredentials(user, password));
HttpClient httpClient = HttpClients.custom().setDefaultCredentialsProvider(credsProvider).build();
// Set up pre-emptive basic Auth because the rabbit plugin doesn't currently support challenge/response for PUT
HttpClient httpClient = HttpClients.custom()
.setDefaultCredentialsProvider(credsProvider).build();
// Set up pre-emptive basic Auth because the rabbit plugin doesn't currently
// support challenge/response for PUT
// Create AuthCache instance
AuthCache authCache = new BasicAuthCache();
// Generate BASIC scheme object and add it to the local; from the apache docs...
@@ -63,20 +66,24 @@ public abstract class RabbitManagementUtils {
catch (URISyntaxException e) {
throw new RabbitAdminException("Invalid URI", e);
}
authCache.put(new HttpHost(uri.getHost(), uri.getPort(), uri.getScheme()), basicAuth);
authCache.put(new HttpHost(uri.getHost(), uri.getPort(), uri.getScheme()),
basicAuth);
// Add AuthCache to the execution context
final HttpClientContext localContext = HttpClientContext.create();
localContext.setAuthCache(authCache);
RestTemplate restTemplate = new RestTemplate(new HttpComponentsClientHttpRequestFactory(httpClient) {
RestTemplate restTemplate = new RestTemplate(
new HttpComponentsClientHttpRequestFactory(httpClient) {
@Override
protected HttpContext createHttpContext(HttpMethod httpMethod, URI uri) {
return localContext;
}
@Override
protected HttpContext createHttpContext(HttpMethod httpMethod,
URI uri) {
return localContext;
}
});
restTemplate.setMessageConverters(Collections.<HttpMessageConverter<?>>singletonList(
new MappingJackson2HttpMessageConverter()));
});
restTemplate
.setMessageConverters(Collections.<HttpMessageConverter<?>>singletonList(
new MappingJackson2HttpMessageConverter()));
return restTemplate;
}

3
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBindingProperties.java
View File

@@ -22,7 +22,7 @@ import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
* @author Marius Bogoevici
* @author Oleg Zhurakousky
*/
public class RabbitBindingProperties implements BinderSpecificPropertiesProvider{
public class RabbitBindingProperties implements BinderSpecificPropertiesProvider {
private RabbitConsumerProperties consumer = new RabbitConsumerProperties();
@@ -43,4 +43,5 @@ public class RabbitBindingProperties implements BinderSpecificPropertiesProvider
public void setProducer(RabbitProducerProperties producer) {
this.producer = producer;
}
}

11
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java
View File

@@ -26,7 +26,7 @@ import org.springframework.amqp.core.ExchangeTypes;
* @since 1.2
*
*/
public abstract class RabbitCommonProperties {
public abstract class RabbitCommonProperties {
public static final String DEAD_LETTER_EXCHANGE = "DLX";
@@ -66,7 +66,8 @@ public abstract class RabbitCommonProperties {
private boolean bindQueue = true;
/**
* routing key to bind (default # for non-partitioned, destination-instanceIndex for partitioned)
* routing key to bind (default # for non-partitioned, destination-instanceIndex for
* partitioned)
*/
private String bindingRoutingKey;
@@ -116,7 +117,8 @@ public abstract class RabbitCommonProperties {
private boolean declareDlx = true;
/**
* a dead letter routing key to assign to that queue; if autoBindDlq is true, defaults to destination
* a dead letter routing key to assign to that queue; if autoBindDlq is true, defaults
* to destination
*/
private String deadLetterRoutingKey;
@@ -151,7 +153,8 @@ public abstract class RabbitCommonProperties {
private String dlqDeadLetterExchange;
/**
* if a DLQ is declared, a dead letter routing key to assign to that queue; default none
* if a DLQ is declared, a dead letter routing key to assign to that queue; default
* none
*/
private String dlqDeadLetterRoutingKey;

4
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java
View File

@@ -77,7 +77,7 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
/**
* patterns to match which headers are mapped (inbound)
*/
private String[] headerPatterns = new String[] {"*"};
private String[] headerPatterns = new String[] { "*" };
/**
* interval between reconnection attempts
@@ -102,7 +102,7 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
/**
* interval between attempts to passively declare missing queues
*/
private Long failedDeclarationRetryInterval;
private Long failedDeclarationRetryInterval;
/**
* Used to create the consumer tags; will be appended by '#n' where 'n' increments for

5
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java
View File

@@ -27,8 +27,8 @@ import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
* @author Soby Chacko
*/
@ConfigurationProperties("spring.cloud.stream.rabbit")
public class RabbitExtendedBindingProperties
extends AbstractExtendedBindingProperties<RabbitConsumerProperties, RabbitProducerProperties, RabbitBindingProperties> {
public class RabbitExtendedBindingProperties extends
AbstractExtendedBindingProperties<RabbitConsumerProperties, RabbitProducerProperties, RabbitBindingProperties> {
private static final String DEFAULTS_PREFIX = "spring.cloud.stream.rabbit.default";
@@ -41,4 +41,5 @@ public class RabbitExtendedBindingProperties
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return RabbitBindingProperties.class;
}
}

11
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java
View File

@@ -65,21 +65,24 @@ public class RabbitProducerProperties extends RabbitCommonProperties {
/**
* patterns to match which headers are mapped (inbound)
*/
private String[] headerPatterns = new String[] {"*"};
private String[] headerPatterns = new String[] { "*" };
/**
* when using a delayed message exchange, a SpEL expression to determine the delay to apply to messages
* when using a delayed message exchange, a SpEL expression to determine the delay to
* apply to messages
*/
private Expression delayExpression;
/**
* a custom routing key when publishing messages; default is the destination name; suffixed by "-partition" when partitioned
* a custom routing key when publishing messages; default is the destination name;
* suffixed by "-partition" when partitioned
*/
private Expression routingKeyExpression;
/**
* the channel name to which to send publisher confirms (acks) if the connection
* factory is so configured; default 'nullChannel'; requires 'errorChannelEnabled=true'
* factory is so configured; default 'nullChannel'; requires
* 'errorChannelEnabled=true'
*/
private String confirmAckChannel;

256
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
View File

@@ -60,12 +60,12 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
* @author Oleg Zhurakousky
*/
public class RabbitExchangeQueueProvisioner implements ApplicationListener<DeclarationExceptionEvent>,
ProvisioningProvider<ExtendedConsumerProperties<RabbitConsumerProperties>,
ExtendedProducerProperties<RabbitProducerProperties>> {
public class RabbitExchangeQueueProvisioner
implements ApplicationListener<DeclarationExceptionEvent>,
ProvisioningProvider<ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
private static final Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR
= new Base64UrlNamingStrategy("anonymous.");
private static final Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR = new Base64UrlNamingStrategy(
"anonymous.");
/**
* The delimiter between a group and index when constructing a binder
@@ -90,38 +90,48 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
@Override
public ProducerDestination provisionProducerDestination(String name,
ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
final String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name);
Exchange exchange = buildExchange(producerProperties.getExtension(), exchangeName);
ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
final String exchangeName = applyPrefix(
producerProperties.getExtension().getPrefix(), name);
Exchange exchange = buildExchange(producerProperties.getExtension(),
exchangeName);
if (producerProperties.getExtension().isDeclareExchange()) {
declareExchange(exchangeName, exchange);
}
Binding binding = null;
for (String requiredGroupName : producerProperties.getRequiredGroups()) {
String baseQueueName = producerProperties.getExtension().isQueueNameGroupOnly()
? requiredGroupName : (exchangeName + "." + requiredGroupName);
String baseQueueName = producerProperties.getExtension()
.isQueueNameGroupOnly() ? requiredGroupName
: (exchangeName + "." + requiredGroupName);
if (!producerProperties.isPartitioned()) {
autoBindDLQ(baseQueueName, baseQueueName, producerProperties.getExtension());
autoBindDLQ(baseQueueName, baseQueueName,
producerProperties.getExtension());
if (producerProperties.getExtension().isBindQueue()) {
Queue queue = new Queue(baseQueueName, true, false, false,
queueArgs(baseQueueName, producerProperties.getExtension(), false));
Queue queue = new Queue(baseQueueName, true, false, false, queueArgs(
baseQueueName, producerProperties.getExtension(), false));
declareQueue(baseQueueName, queue);
binding = notPartitionedBinding(exchange, queue, producerProperties.getExtension());
binding = notPartitionedBinding(exchange, queue,
producerProperties.getExtension());
}
}
else {
// if the stream is partitioned, create one queue for each target partition for the default group
// if the stream is partitioned, create one queue for each target
// partition for the default group
for (int i = 0; i < producerProperties.getPartitionCount(); i++) {
String partitionSuffix = "-" + i;
String partitionQueueName = baseQueueName + partitionSuffix;
autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, producerProperties.getExtension());
autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix,
producerProperties.getExtension());
if (producerProperties.getExtension().isBindQueue()) {
Queue queue = new Queue(partitionQueueName, true, false, false,
queueArgs(partitionQueueName, producerProperties.getExtension(), false));
queueArgs(partitionQueueName,
producerProperties.getExtension(), false));
declareQueue(queue.getName(), queue);
String prefix = producerProperties.getExtension().getPrefix();
String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length());
binding = partitionedBinding(destination, exchange, queue, producerProperties.getExtension(), i);
String destination = StringUtils.isEmpty(prefix) ? exchangeName
: exchangeName.substring(prefix.length());
binding = partitionedBinding(destination, exchange, queue,
producerProperties.getExtension(), i);
}
}
}
@@ -137,10 +147,14 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
consumerDestination = doProvisionConsumerDestination(name, group, properties);
}
else {
String[] provisionedDestinations = Stream.of(StringUtils.tokenizeToStringArray(name, ",", true, true))
.map(destination -> doProvisionConsumerDestination(destination, group, properties).getName())
.toArray(String[]::new);
consumerDestination = new RabbitConsumerDestination(StringUtils.arrayToCommaDelimitedString(provisionedDestinations), null);
String[] provisionedDestinations = Stream
.of(StringUtils.tokenizeToStringArray(name, ",", true, true))
.map(destination -> doProvisionConsumerDestination(destination, group,
properties).getName())
.toArray(String[]::new);
consumerDestination = new RabbitConsumerDestination(
StringUtils.arrayToCommaDelimitedString(provisionedDestinations),
null);
}
return consumerDestination;
}
@@ -148,15 +162,18 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
private ConsumerDestination doProvisionConsumerDestination(String name, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
boolean anonymous = !StringUtils.hasText(group);
String baseQueueName;
String baseQueueName;
if (properties.getExtension().isQueueNameGroupOnly()) {
baseQueueName = anonymous ? ANONYMOUS_GROUP_NAME_GENERATOR.generateName() : group;
baseQueueName = anonymous ? ANONYMOUS_GROUP_NAME_GENERATOR.generateName()
: group;
}
else {
baseQueueName = groupedName(name, anonymous ? ANONYMOUS_GROUP_NAME_GENERATOR.generateName() : group);
baseQueueName = groupedName(name,
anonymous ? ANONYMOUS_GROUP_NAME_GENERATOR.generateName() : group);
}
if (this.logger.isInfoEnabled()) {
this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name);
this.logger.info("declaring queue for inbound: " + baseQueueName
+ ", bound to: " + name);
}
String prefix = properties.getExtension().getPrefix();
final String exchangeName = applyPrefix(prefix, name);
@@ -169,7 +186,8 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
boolean durable = !anonymous && properties.getExtension().isDurableSubscription();
Queue queue;
if (anonymous) {
queue = new Queue(queueName, false, true, true, queueArgs(queueName, properties.getExtension(), false));
queue = new Queue(queueName, false, true, true,
queueArgs(queueName, properties.getExtension(), false));
}
else {
if (partitioned) {
@@ -188,107 +206,111 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
Binding binding = null;
if (properties.getExtension().isBindQueue()) {
declareQueue(queueName, queue);
binding = declareConsumerBindings(name, properties, exchange, partitioned, queue);
binding = declareConsumerBindings(name, properties, exchange, partitioned,
queue);
}
if (durable) {
autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName,
properties.getExtension());
autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName),
queueName, properties.getExtension());
}
return new RabbitConsumerDestination(queue.getName(), binding);
}
/**
* Construct a name comprised of the name and group.
*
* @param name the name.
* @param name the name.
* @param group the group.
* @return the constructed name.
*/
protected final String groupedName(String name, String group) {
return name + GROUP_INDEX_DELIMITER + (StringUtils.hasText(group) ? group : "default");
return name + GROUP_INDEX_DELIMITER
+ (StringUtils.hasText(group) ? group : "default");
}
private Binding partitionedBinding(String destination, Exchange exchange, Queue queue,
RabbitCommonProperties extendedProperties, int index) {
RabbitCommonProperties extendedProperties, int index) {
String bindingKey = extendedProperties.getBindingRoutingKey();
if (bindingKey == null) {
bindingKey = destination;
}
bindingKey += "-" + index;
if (exchange instanceof TopicExchange) {
Binding binding = BindingBuilder.bind(queue)
.to((TopicExchange) exchange)
Binding binding = BindingBuilder.bind(queue).to((TopicExchange) exchange)
.with(bindingKey);
declareBinding(queue.getName(), binding);
return binding;
}
else if (exchange instanceof DirectExchange) {
Binding binding = BindingBuilder.bind(queue)
.to((DirectExchange) exchange)
Binding binding = BindingBuilder.bind(queue).to((DirectExchange) exchange)
.with(bindingKey);
declareBinding(queue.getName(), binding);
return binding;
}
else if (exchange instanceof FanoutExchange) {
throw new ProvisioningException("A fanout exchange is not appropriate for partitioned apps");
throw new ProvisioningException(
"A fanout exchange is not appropriate for partitioned apps");
}
else {
throw new ProvisioningException("Cannot bind to a " + exchange.getType() + " exchange");
throw new ProvisioningException(
"Cannot bind to a " + exchange.getType() + " exchange");
}
}
private Binding declareConsumerBindings(String name, ExtendedConsumerProperties<RabbitConsumerProperties> properties,
Exchange exchange, boolean partitioned, Queue queue) {
private Binding declareConsumerBindings(String name,
ExtendedConsumerProperties<RabbitConsumerProperties> properties,
Exchange exchange, boolean partitioned, Queue queue) {
if (partitioned) {
return partitionedBinding(name, exchange, queue, properties.getExtension(), properties.getInstanceIndex());
return partitionedBinding(name, exchange, queue, properties.getExtension(),
properties.getInstanceIndex());
}
else {
return notPartitionedBinding(exchange, queue, properties.getExtension());
}
}
private Binding notPartitionedBinding(Exchange exchange, Queue queue, RabbitCommonProperties extendedProperties) {
private Binding notPartitionedBinding(Exchange exchange, Queue queue,
RabbitCommonProperties extendedProperties) {
String routingKey = extendedProperties.getBindingRoutingKey();
if (routingKey == null) {
routingKey = "#";
}
if (exchange instanceof TopicExchange) {
Binding binding = BindingBuilder.bind(queue)
.to((TopicExchange) exchange)
Binding binding = BindingBuilder.bind(queue).to((TopicExchange) exchange)
.with(routingKey);
declareBinding(queue.getName(), binding);
return binding;
}
else if (exchange instanceof DirectExchange) {
Binding binding = BindingBuilder.bind(queue)
.to((DirectExchange) exchange)
Binding binding = BindingBuilder.bind(queue).to((DirectExchange) exchange)
.with(routingKey);
declareBinding(queue.getName(), binding);
return binding;
}
else if (exchange instanceof FanoutExchange) {
Binding binding = BindingBuilder.bind(queue)
.to((FanoutExchange) exchange);
Binding binding = BindingBuilder.bind(queue).to((FanoutExchange) exchange);
declareBinding(queue.getName(), binding);
return binding;
}
else {
throw new ProvisioningException("Cannot bind to a " + exchange.getType() + " exchange");
throw new ProvisioningException(
"Cannot bind to a " + exchange.getType() + " exchange");
}
}
/**
* 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 baseQueueName The base name for the queue (including the binder prefix, if any).
* @param routingKey The routing key for the queue.
* 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 baseQueueName The base name for the queue (including the binder prefix, if
* any).
* @param routingKey The routing key for the queue.
* @param properties the properties.
*/
private void autoBindDLQ(final String baseQueueName, String routingKey, RabbitCommonProperties properties) {
private void autoBindDLQ(final String baseQueueName, String routingKey,
RabbitCommonProperties properties) {
boolean autoBindDlq = properties.isAutoBindDlq();
if (this.logger.isDebugEnabled()) {
this.logger.debug("autoBindDLQ=" + autoBindDlq
+ " for: " + baseQueueName);
this.logger.debug("autoBindDLQ=" + autoBindDlq + " for: " + baseQueueName);
}
if (autoBindDlq) {
String dlqName;
@@ -298,27 +320,29 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
else {
dlqName = properties.getDeadLetterQueueName();
}
Queue dlq = new Queue(dlqName, true, false, false, queueArgs(dlqName, properties, true));
Queue dlq = new Queue(dlqName, true, false, false,
queueArgs(dlqName, properties, true));
declareQueue(dlqName, dlq);
String dlxName = deadLetterExchangeName(properties);
if (properties.isDeclareDlx()) {
declareExchange(dlxName,
new ExchangeBuilder(dlxName, properties.getDeadLetterExchangeType())
.durable(true)
.build());
new ExchangeBuilder(dlxName,
properties.getDeadLetterExchangeType()).durable(true)
.build());
}
Binding dlqBinding = new Binding(dlq.getName(), DestinationType.QUEUE, dlxName,
properties.getDlqDeadLetterRoutingKey() == null ? routingKey : properties.getDeadLetterRoutingKey(),
Binding dlqBinding = new Binding(dlq.getName(), DestinationType.QUEUE,
dlxName, properties.getDlqDeadLetterRoutingKey() == null ? routingKey
: properties.getDeadLetterRoutingKey(),
null);
declareBinding(dlqName, dlqBinding);
if (properties instanceof RabbitConsumerProperties &&
((RabbitConsumerProperties) properties).isRepublishToDlq()) {
if (properties instanceof RabbitConsumerProperties
&& ((RabbitConsumerProperties) properties).isRepublishToDlq()) {
/*
* Also bind with the base queue name when republishToDlq is used, which does not know about
* partitioning
* Also bind with the base queue name when republishToDlq is used, which
* does not know about partitioning
*/
declareBinding(dlqName,
new Binding(dlq.getName(), DestinationType.QUEUE, dlxName, baseQueueName, null));
declareBinding(dlqName, new Binding(dlq.getName(), DestinationType.QUEUE,
dlxName, baseQueueName, null));
}
}
}
@@ -326,14 +350,12 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
/**
* For binder implementations that support dead lettering, construct the name of the
* dead letter entity for the underlying pipe name.
*
* @param name the name.
*/
public static String constructDLQName(String name) {
return name + ".dlq";
}
private String deadLetterExchangeName(RabbitCommonProperties properties) {
if (properties.getDeadLetterExchange() == null) {
return properties.getPrefix() + RabbitCommonProperties.DEAD_LETTER_EXCHANGE;
@@ -343,14 +365,14 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
}
}
private void declareQueue(String beanName, Queue queue) {
try {
this.rabbitAdmin.declareQueue(queue);
}
catch (AmqpConnectException e) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
this.logger.debug("Declaration of queue: " + queue.getName()
+ " deferred - connection not available");
}
}
catch (RuntimeException e) {
@@ -359,13 +381,15 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
throw e;
}
if (this.logger.isDebugEnabled()) {
this.logger.debug("Declaration of queue: " + queue.getName() + " deferred", e);
this.logger.debug(
"Declaration of queue: " + queue.getName() + " deferred", e);
}
}
addToAutoDeclareContext(beanName, queue);
}
private Map<String, Object> queueArgs(String queueName, RabbitCommonProperties properties, boolean isDlq) {
private Map<String, Object> queueArgs(String queueName,
RabbitCommonProperties properties, boolean isDlq) {
Map<String, Object> args = new HashMap<>();
if (!isDlq) {
if (properties.isAutoBindDlq()) {
@@ -392,21 +416,27 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
args.put("x-dead-letter-exchange", properties.getDlqDeadLetterExchange());
}
if (properties.getDlqDeadLetterRoutingKey() != null) {
args.put("x-dead-letter-routing-key", properties.getDlqDeadLetterRoutingKey());
args.put("x-dead-letter-routing-key",
properties.getDlqDeadLetterRoutingKey());
}
}
additionalArgs(args, properties, isDlq);
return args;
}
private void additionalArgs(Map<String, Object> args, RabbitCommonProperties properties, boolean isDlq) {
private void additionalArgs(Map<String, Object> args,
RabbitCommonProperties properties, boolean isDlq) {
Integer expires = isDlq ? properties.getDlqExpires() : properties.getExpires();
Integer maxLength = isDlq ? properties.getDlqMaxLength() : properties.getMaxLength();
Integer maxLengthBytes = isDlq ? properties.getDlqMaxLengthBytes() : properties.getMaxLengthBytes();
Integer maxPriority = isDlq ? properties.getDlqMaxPriority() : properties.getMaxPriority();
Integer maxLength = isDlq ? properties.getDlqMaxLength()
: properties.getMaxLength();
Integer maxLengthBytes = isDlq ? properties.getDlqMaxLengthBytes()
: properties.getMaxLengthBytes();
Integer maxPriority = isDlq ? properties.getDlqMaxPriority()
: properties.getMaxPriority();
Integer ttl = isDlq ? properties.getDlqTtl() : properties.getTtl();
boolean lazy = isDlq ? properties.isDlqLazy() : properties.isLazy();
String overflow = isDlq ? properties.getDlqOverflowBehavior() : properties.getOverflowBehavior();
String overflow = isDlq ? properties.getDlqOverflowBehavior()
: properties.getOverflowBehavior();
if (expires != null) {
args.put("x-expires", expires);
}
@@ -430,14 +460,15 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
}
}
public static String applyPrefix(String prefix, String name) {
return prefix + name;
}
private Exchange buildExchange(RabbitCommonProperties properties, String exchangeName) {
private Exchange buildExchange(RabbitCommonProperties properties,
String exchangeName) {
try {
ExchangeBuilder builder = new ExchangeBuilder(exchangeName, properties.getExchangeType());
ExchangeBuilder builder = new ExchangeBuilder(exchangeName,
properties.getExchangeType());
builder.durable(properties.isExchangeDurable());
if (properties.isExchangeAutoDelete()) {
builder.autoDelete();
@@ -458,8 +489,8 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
}
catch (AmqpConnectException e) {
if (this.logger.isDebugEnabled()) {
this.logger.debug(
"Declaration of exchange: " + exchange.getName() + " deferred - connection not available");
this.logger.debug("Declaration of exchange: " + exchange.getName()
+ " deferred - connection not available");
}
}
catch (RuntimeException e) {
@@ -468,7 +499,9 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
throw e;
}
if (this.logger.isDebugEnabled()) {
this.logger.debug("Declaration of exchange: " + exchange.getName() + " deferred", e);
this.logger.debug(
"Declaration of exchange: " + exchange.getName() + " deferred",
e);
}
}
addToAutoDeclareContext(rootName + ".exchange", exchange);
@@ -482,14 +515,15 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
}
}
private 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 (this.logger.isDebugEnabled()) {
this.logger.debug(
"Declaration of binding: " + rootName + ".binding deferred - connection not available");
this.logger.debug("Declaration of binding: " + rootName
+ ".binding deferred - connection not available");
}
}
catch (RuntimeException e) {
@@ -498,7 +532,8 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
throw e;
}
if (this.logger.isDebugEnabled()) {
this.logger.debug("Declaration of binding: " + rootName + ".binding deferred", e);
this.logger.debug(
"Declaration of binding: " + rootName + ".binding deferred", e);
}
}
addToAutoDeclareContext(rootName + ".binding", binding);
@@ -507,20 +542,22 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
public void cleanAutoDeclareContext(ConsumerDestination destination,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
synchronized (this.autoDeclareContext) {
Stream.of(StringUtils.tokenizeToStringArray(destination.getName(), ",", true, true)).forEach(name -> {
name = name.trim();
removeSingleton(name + ".binding");
removeSingleton(name);
String dlq = name + ".dlq";
removeSingleton(dlq + ".binding");
removeSingleton(dlq);
});
Stream.of(StringUtils.tokenizeToStringArray(destination.getName(), ",", true,
true)).forEach(name -> {
name = name.trim();
removeSingleton(name + ".binding");
removeSingleton(name);
String dlq = name + ".dlq";
removeSingleton(dlq + ".binding");
removeSingleton(dlq);
});
}
}
private void removeSingleton(String name) {
if (this.autoDeclareContext.containsBean(name)) {
ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory();
ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext
.getBeanFactory();
if (beanFactory instanceof DefaultListableBeanFactory) {
((DefaultListableBeanFactory) beanFactory).destroySingleton(name);
}
@@ -556,16 +593,16 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
@Override
public String toString() {
return "RabbitProducerDestination{" +
"exchange=" + exchange +
", binding=" + binding +
'}';
return "RabbitProducerDestination{" + "exchange=" + exchange + ", binding="
+ binding + '}';
}
}
private static final class RabbitConsumerDestination implements ConsumerDestination {
private final String queue;
private final Binding binding;
RabbitConsumerDestination(String queue, Binding binding) {
@@ -576,16 +613,15 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
@Override
public String toString() {
return "RabbitConsumerDestination{" +
"queue=" + queue +
", binding=" + binding +
'}';
return "RabbitConsumerDestination{" + "queue=" + queue + ", binding="
+ binding + '}';
}
@Override
public String getName() {
return this.queue;
}
}
}

23
spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java
View File

@@ -31,13 +31,15 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport;
/**
* JUnit {@link org.junit.Rule} that detects the fact that RabbitMQ is available on localhost.
* JUnit {@link org.junit.Rule} that detects the fact that RabbitMQ is available on
* localhost.
*
* @author Mark Fisher
* @author Gary Russell
* @author Eric Bottard
*/
public class RabbitTestSupport extends AbstractExternalResourceTestSupport<CachingConnectionFactory> {
public class RabbitTestSupport
extends AbstractExternalResourceTestSupport<CachingConnectionFactory> {
private final boolean management;
@@ -79,7 +81,8 @@ public class RabbitTestSupport extends AbstractExternalResourceTestSupport<Cachi
private volatile ServerSocket serverSocket;
public RabbitProxy() throws IOException {
ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket(0);
ServerSocket serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(0);
this.port = serverSocket.getLocalPort();
serverSocket.close();
}
@@ -89,7 +92,8 @@ public class RabbitTestSupport extends AbstractExternalResourceTestSupport<Cachi
}
public void start() throws IOException {
this.serverSocket = ServerSocketFactory.getDefault().createServerSocket(this.port);
this.serverSocket = ServerSocketFactory.getDefault()
.createServerSocket(this.port);
this.serverExec.execute(new Runnable() {
@Override
@@ -102,15 +106,18 @@ public class RabbitTestSupport extends AbstractExternalResourceTestSupport<Cachi
@Override
public void run() {
try {
final Socket rabbitSocket = SocketFactory.getDefault().createSocket("localhost",
5672);
final Socket rabbitSocket = SocketFactory
.getDefault()
.createSocket("localhost", 5672);
socketExec.execute(new Runnable() {
@Override
public void run() {
try {
InputStream is = rabbitSocket.getInputStream();
OutputStream os = socket.getOutputStream();
InputStream is = rabbitSocket
.getInputStream();
OutputStream os = socket
.getOutputStream();
int c;
while ((c = is.read()) >= 0) {
os.write(c);

11
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExpressionEvaluatingInterceptor.java
View File

@@ -49,14 +49,14 @@ public class RabbitExpressionEvaluatingInterceptor implements ChannelInterceptor
private final EvaluationContext evaluationContext;
/**
* Construct an instance with the provided expressions and evaluation context.
* At least one expression muse be non-null.
* Construct an instance with the provided expressions and evaluation context. At
* least one expression muse be non-null.
* @param routingKeyExpression the routing key expresssion.
* @param delayExpression the delay expression.
* @param evaluationContext the evaluation context.
*/
public RabbitExpressionEvaluatingInterceptor(Expression routingKeyExpression, Expression delayExpression,
EvaluationContext evaluationContext) {
public RabbitExpressionEvaluatingInterceptor(Expression routingKeyExpression,
Expression delayExpression, EvaluationContext evaluationContext) {
Assert.isTrue(routingKeyExpression != null || delayExpression != null,
"At least one expression is required");
Assert.notNull(evaluationContext, "the 'evaluationContext' cannot be null");
@@ -83,7 +83,8 @@ public class RabbitExpressionEvaluatingInterceptor implements ChannelInterceptor
this.routingKeyExpression.getValue(this.evaluationContext, message));
}
if (this.delayExpression != null) {
builder.setHeader(DELAY_HEADER, this.delayExpression.getValue(this.evaluationContext, message));
builder.setHeader(DELAY_HEADER,
this.delayExpression.getValue(this.evaluationContext, message));
}
return builder.build();
}

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

@@ -118,30 +118,28 @@ import com.rabbitmq.client.Envelope;
* @author Soby Chacko
* @author Oleg Zhurakousky
*/
public class RabbitMessageChannelBinder
extends AbstractMessageChannelBinder<ExtendedConsumerProperties<RabbitConsumerProperties>,
ExtendedProducerProperties<RabbitProducerProperties>, RabbitExchangeQueueProvisioner>
implements ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties>,
DisposableBean {
public class RabbitMessageChannelBinder extends
AbstractMessageChannelBinder<ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>, RabbitExchangeQueueProvisioner>
implements
ExtendedPropertiesBinder<MessageChannel, RabbitConsumerProperties, RabbitProducerProperties>,
DisposableBean {
private static final SimplePassthroughMessageConverter passThoughConverter =
new SimplePassthroughMessageConverter();
private static final SimplePassthroughMessageConverter passThoughConverter = new SimplePassthroughMessageConverter();
private static final AmqpMessageHeaderErrorMessageStrategy errorMessageStrategy =
new AmqpMessageHeaderErrorMessageStrategy();
private static final AmqpMessageHeaderErrorMessageStrategy errorMessageStrategy = new AmqpMessageHeaderErrorMessageStrategy();
private static final MessagePropertiesConverter inboundMessagePropertiesConverter =
new DefaultMessagePropertiesConverter() {
private static final MessagePropertiesConverter inboundMessagePropertiesConverter = new DefaultMessagePropertiesConverter() {
@Override
public MessageProperties toMessageProperties(AMQP.BasicProperties source, Envelope envelope,
String charset) {
MessageProperties properties = super.toMessageProperties(source, envelope, charset);
properties.setDeliveryMode(null);
return properties;
}
@Override
public MessageProperties toMessageProperties(AMQP.BasicProperties source,
Envelope envelope, String charset) {
MessageProperties properties = super.toMessageProperties(source, envelope,
charset);
properties.setDeliveryMode(null);
return properties;
}
};
};
private final RabbitProperties rabbitProperties;
@@ -161,12 +159,14 @@ public class RabbitMessageChannelBinder
private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties();
public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties,
public RabbitMessageChannelBinder(ConnectionFactory connectionFactory,
RabbitProperties rabbitProperties,
RabbitExchangeQueueProvisioner provisioningProvider) {
this(connectionFactory, rabbitProperties, provisioningProvider, null);
}
public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties,
public RabbitMessageChannelBinder(ConnectionFactory connectionFactory,
RabbitProperties rabbitProperties,
RabbitExchangeQueueProvisioner provisioningProvider,
ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer) {
super(new String[0], provisioningProvider, containerCustomizer);
@@ -181,16 +181,19 @@ public class RabbitMessageChannelBinder
* {@link DelegatingDecompressingPostProcessor} with its default delegates.
* @param decompressingPostProcessor the post processor.
*/
public void setDecompressingPostProcessor(MessagePostProcessor decompressingPostProcessor) {
public void setDecompressingPostProcessor(
MessagePostProcessor decompressingPostProcessor) {
this.decompressingPostProcessor = decompressingPostProcessor;
}
/**
* Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress messages.
* Defaults to a {@link org.springframework.amqp.support.postprocessor.GZipPostProcessor}.
* Set a {@link org.springframework.amqp.core.MessagePostProcessor} to compress
* messages. Defaults to a
* {@link org.springframework.amqp.support.postprocessor.GZipPostProcessor}.
* @param compressingPostProcessor the post processor.
*/
public void setCompressingPostProcessor(MessagePostProcessor compressingPostProcessor) {
public void setCompressingPostProcessor(
MessagePostProcessor compressingPostProcessor) {
this.compressingPostProcessor = compressingPostProcessor;
}
@@ -203,7 +206,8 @@ public class RabbitMessageChannelBinder
this.clustered = nodes.length > 1;
}
public void setExtendedBindingProperties(RabbitExtendedBindingProperties extendedBindingProperties) {
public void setExtendedBindingProperties(
RabbitExtendedBindingProperties extendedBindingProperties) {
this.extendedBindingProperties = extendedBindingProperties;
}
@@ -211,16 +215,21 @@ public class RabbitMessageChannelBinder
public void onInit() throws Exception {
super.onInit();
if (this.clustered) {
String[] addresses = StringUtils.commaDelimitedListToStringArray(this.rabbitProperties.getAddresses());
String[] addresses = StringUtils.commaDelimitedListToStringArray(
this.rabbitProperties.getAddresses());
Assert.state(addresses.length == this.adminAddresses.length
&& addresses.length == this.nodes.length,
Assert.state(
addresses.length == this.adminAddresses.length
&& 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, this.rabbitProperties.getVirtualHost(),
this.rabbitProperties.getUsername(), this.rabbitProperties.getPassword(),
this.rabbitProperties.getSsl().isEnabled(), this.rabbitProperties.getSsl().getKeyStore(),
this.connectionFactory = new LocalizedQueueConnectionFactory(
this.connectionFactory, addresses, 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().getTrustStorePassword());
@@ -258,18 +267,24 @@ public class RabbitMessageChannelBinder
}
@Override
protected MessageHandler createProducerMessageHandler(final ProducerDestination producerDestination,
ExtendedProducerProperties<RabbitProducerProperties> producerProperties, MessageChannel errorChannel) {
Assert.state(!HeaderMode.embeddedHeaders.equals(producerProperties.getHeaderMode()),
protected MessageHandler createProducerMessageHandler(
final ProducerDestination producerDestination,
ExtendedProducerProperties<RabbitProducerProperties> producerProperties,
MessageChannel errorChannel) {
Assert.state(
!HeaderMode.embeddedHeaders.equals(producerProperties.getHeaderMode()),
"the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively");
String prefix = producerProperties.getExtension().getPrefix();
String exchangeName = producerDestination.getName();
String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length());
String destination = StringUtils.isEmpty(prefix) ? exchangeName
: exchangeName.substring(prefix.length());
final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(
buildRabbitTemplate(producerProperties.getExtension(), errorChannel != null));
buildRabbitTemplate(producerProperties.getExtension(),
errorChannel != null));
endpoint.setExchangeName(producerDestination.getName());
RabbitProducerProperties extendedProperties = producerProperties.getExtension();
boolean expressionInterceptorNeeded = expressionInterceptorNeeded(extendedProperties);
boolean expressionInterceptorNeeded = expressionInterceptorNeeded(
extendedProperties);
Expression routingKeyExpression = extendedProperties.getRoutingKeyExpression();
if (!producerProperties.isPartitioned()) {
if (routingKeyExpression == null) {
@@ -278,7 +293,8 @@ public class RabbitMessageChannelBinder
else {
if (expressionInterceptorNeeded) {
endpoint.setRoutingKeyExpressionString("headers['"
+ RabbitExpressionEvaluatingInterceptor.ROUTING_KEY_HEADER + "']");
+ RabbitExpressionEvaluatingInterceptor.ROUTING_KEY_HEADER
+ "']");
}
else {
endpoint.setRoutingKeyExpression(routingKeyExpression);
@@ -287,16 +303,19 @@ public class RabbitMessageChannelBinder
}
else {
if (routingKeyExpression == null) {
endpoint.setRoutingKeyExpression(buildPartitionRoutingExpression(destination, false));
endpoint.setRoutingKeyExpression(
buildPartitionRoutingExpression(destination, false));
}
else {
if (expressionInterceptorNeeded) {
endpoint.setRoutingKeyExpression(buildPartitionRoutingExpression("headers['"
+ RabbitExpressionEvaluatingInterceptor.ROUTING_KEY_HEADER + "']", true));
endpoint.setRoutingKeyExpression(
buildPartitionRoutingExpression("headers['"
+ RabbitExpressionEvaluatingInterceptor.ROUTING_KEY_HEADER
+ "']", true));
}
else {
endpoint.setRoutingKeyExpression(buildPartitionRoutingExpression(routingKeyExpression.getExpressionString(),
true));
endpoint.setRoutingKeyExpression(buildPartitionRoutingExpression(
routingKeyExpression.getExpressionString(), true));
}
}
}
@@ -310,10 +329,12 @@ public class RabbitMessageChannelBinder
}
}
DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper();
List<String> headerPatterns = new ArrayList<>(extendedProperties.getHeaderPatterns().length + 1);
List<String> headerPatterns = new ArrayList<>(
extendedProperties.getHeaderPatterns().length + 1);
headerPatterns.add("!" + BinderHeaders.PARTITION_HEADER);
headerPatterns.addAll(Arrays.asList(extendedProperties.getHeaderPatterns()));
mapper.setRequestHeaderNames(headerPatterns.toArray(new String[headerPatterns.size()]));
mapper.setRequestHeaderNames(
headerPatterns.toArray(new String[headerPatterns.size()]));
endpoint.setHeaderMapper(mapper);
endpoint.setDefaultDeliveryMode(extendedProperties.getDeliveryMode());
endpoint.setBeanFactory(this.getBeanFactory());
@@ -321,13 +342,15 @@ public class RabbitMessageChannelBinder
checkConnectionFactoryIsErrorCapable();
endpoint.setReturnChannel(errorChannel);
endpoint.setConfirmNackChannel(errorChannel);
String ackChannelBeanName = StringUtils.hasText(extendedProperties.getConfirmAckChannel())
? extendedProperties.getConfirmAckChannel()
: IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME;
String ackChannelBeanName = StringUtils
.hasText(extendedProperties.getConfirmAckChannel())
? extendedProperties.getConfirmAckChannel()
: IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME;
if (!ackChannelBeanName.equals(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME)
&& !getApplicationContext().containsBean(ackChannelBeanName)) {
GenericApplicationContext context = (GenericApplicationContext) getApplicationContext();
context.registerBean(ackChannelBeanName, DirectChannel.class, () -> new DirectChannel());
context.registerBean(ackChannelBeanName, DirectChannel.class,
() -> new DirectChannel());
}
endpoint.setConfirmAckChannelName(ackChannelBeanName);
endpoint.setConfirmCorrelationExpressionString("#root");
@@ -337,110 +360,134 @@ public class RabbitMessageChannelBinder
return endpoint;
}
@Override
protected void postProcessOutputChannel(MessageChannel outputChannel,
ExtendedProducerProperties<RabbitProducerProperties> producerProperties) {
RabbitProducerProperties extendedProperties = producerProperties.getExtension();
if (expressionInterceptorNeeded(extendedProperties)) {
((AbstractMessageChannel) outputChannel).addInterceptor(0,
new RabbitExpressionEvaluatingInterceptor(extendedProperties.getRoutingKeyExpression(),
extendedProperties.getDelayExpression(), getEvaluationContext()));
new RabbitExpressionEvaluatingInterceptor(
extendedProperties.getRoutingKeyExpression(),
extendedProperties.getDelayExpression(),
getEvaluationContext()));
}
}
private boolean expressionInterceptorNeeded(RabbitProducerProperties extendedProperties) {
private boolean expressionInterceptorNeeded(
RabbitProducerProperties extendedProperties) {
return extendedProperties.getRoutingKeyExpression() != null
&& extendedProperties.getRoutingKeyExpression().getExpressionString().contains("payload")
|| (extendedProperties.getDelayExpression() != null
&& extendedProperties.getDelayExpression().getExpressionString().contains("payload"));
&& extendedProperties.getRoutingKeyExpression().getExpressionString()
.contains("payload")
|| (extendedProperties.getDelayExpression() != null && extendedProperties
.getDelayExpression().getExpressionString().contains("payload"));
}
private void checkConnectionFactoryIsErrorCapable() {
if (!(this.connectionFactory instanceof CachingConnectionFactory)) {
logger.warn("Unknown connection factory type, cannot determine error capabilities: "
+ this.connectionFactory.getClass());
logger.warn(
"Unknown connection factory type, cannot determine error capabilities: "
+ this.connectionFactory.getClass());
}
else {
CachingConnectionFactory ccf = (CachingConnectionFactory) this.connectionFactory;
if (!ccf.isPublisherConfirms() && !ccf.isPublisherReturns()) {
logger.warn("Producer error channel is enabled, but the connection factory is not configured for "
+ "returns or confirms; the error channel will receive no messages");
logger.warn(
"Producer error channel is enabled, but the connection factory is not configured for "
+ "returns or confirms; the error channel will receive no messages");
}
else if (!ccf.isPublisherConfirms()) {
logger.info("Producer error channel is enabled, but the connection factory is only configured to "
+ "handle returned messages; negative acks will not be reported");
logger.info(
"Producer error channel is enabled, but the connection factory is only configured to "
+ "handle returned messages; negative acks will not be reported");
}
else if (!ccf.isPublisherReturns()) {
logger.info("Producer error channel is enabled, but the connection factory is only configured to "
+ "handle negatively acked messages; returned messages will not be reported");
logger.info(
"Producer error channel is enabled, but the connection factory is only configured to "
+ "handle negatively acked messages; returned messages will not be reported");
}
}
}
private Expression buildPartitionRoutingExpression(String expressionRoot, boolean rootIsExpression) {
private Expression buildPartitionRoutingExpression(String expressionRoot,
boolean rootIsExpression) {
String partitionRoutingExpression = rootIsExpression
? expressionRoot + " + '-' + headers['" + BinderHeaders.PARTITION_HEADER + "']"
: "'" + expressionRoot + "-' + headers['" + BinderHeaders.PARTITION_HEADER + "']";
? expressionRoot + " + '-' + headers['" + BinderHeaders.PARTITION_HEADER
+ "']"
: "'" + expressionRoot + "-' + headers['" + BinderHeaders.PARTITION_HEADER
+ "']";
return new SpelExpressionParser().parseExpression(partitionRoutingExpression);
}
@Override
protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group,
protected MessageProducer createConsumerEndpoint(
ConsumerDestination consumerDestination, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
Assert.state(!HeaderMode.embeddedHeaders.equals(properties.getHeaderMode()),
"the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively");
String destination = consumerDestination.getName();
boolean directContainer = properties.getExtension().getContainerType().equals(ContainerType.DIRECT);
boolean directContainer = properties.getExtension().getContainerType()
.equals(ContainerType.DIRECT);
AbstractMessageListenerContainer listenerContainer = directContainer
? new DirectMessageListenerContainer(this.connectionFactory)
: new SimpleMessageListenerContainer(this.connectionFactory);
listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
listenerContainer
.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode());
listenerContainer.setChannelTransacted(properties.getExtension().isTransacted());
listenerContainer.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected());
listenerContainer
.setDefaultRequeueRejected(properties.getExtension().isRequeueRejected());
int concurrency = properties.getConcurrency();
concurrency = concurrency > 0 ? concurrency : 1;
if (directContainer) {
setDMLCProperties(properties, (DirectMessageListenerContainer) listenerContainer, concurrency);
setDMLCProperties(properties,
(DirectMessageListenerContainer) listenerContainer, concurrency);
}
else {
setSMLCProperties(properties, (SimpleMessageListenerContainer) listenerContainer, concurrency);
setSMLCProperties(properties,
(SimpleMessageListenerContainer) listenerContainer, concurrency);
}
listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch());
listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
listenerContainer
.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
listenerContainer.setTaskExecutor(
new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-"));
String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true);
listenerContainer.setQueueNames(queues);
listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor);
listenerContainer.setMessagePropertiesConverter(
RabbitMessageChannelBinder.inboundMessagePropertiesConverter);
listenerContainer.setExclusive(properties.getExtension().isExclusive());
listenerContainer.setMissingQueuesFatal(properties.getExtension().getMissingQueuesFatal());
listenerContainer
.setMissingQueuesFatal(properties.getExtension().getMissingQueuesFatal());
if (properties.getExtension().getFailedDeclarationRetryInterval() != null) {
listenerContainer.setFailedDeclarationRetryInterval(
properties.getExtension().getFailedDeclarationRetryInterval());
}
if (getApplicationEventPublisher() != null) {
listenerContainer.setApplicationEventPublisher(getApplicationEventPublisher());
listenerContainer
.setApplicationEventPublisher(getApplicationEventPublisher());
}
else if (getApplicationContext() != null) {
listenerContainer.setApplicationEventPublisher(getApplicationContext());
}
getContainerCustomizer().configure(listenerContainer, consumerDestination.getName(), group);
getContainerCustomizer().configure(listenerContainer,
consumerDestination.getName(), group);
if (StringUtils.hasText(properties.getExtension().getConsumerTagPrefix())) {
final AtomicInteger index = new AtomicInteger();
listenerContainer.setConsumerTagStrategy(q ->
properties.getExtension().getConsumerTagPrefix() + "#" + index.getAndIncrement());
listenerContainer.setConsumerTagStrategy(
q -> properties.getExtension().getConsumerTagPrefix() + "#"
+ index.getAndIncrement());
}
listenerContainer.afterPropertiesSet();
AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer);
AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(
listenerContainer);
adapter.setBeanFactory(this.getBeanFactory());
adapter.setBeanName("inbound." + destination);
DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper();
mapper.setRequestHeaderNames(properties.getExtension().getHeaderPatterns());
adapter.setHeaderMapper(mapper);
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(consumerDestination, group, properties);
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(
consumerDestination, group, properties);
if (properties.getMaxAttempts() > 1) {
adapter.setRetryTemplate(buildRetryTemplate(properties));
adapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
@@ -453,7 +500,8 @@ public class RabbitMessageChannelBinder
return adapter;
}
private void setSMLCProperties(ExtendedConsumerProperties<RabbitConsumerProperties> properties,
private void setSMLCProperties(
ExtendedConsumerProperties<RabbitConsumerProperties> properties,
SimpleMessageListenerContainer listenerContainer, int concurrency) {
listenerContainer.setConcurrentConsumers(concurrency);
@@ -463,42 +511,51 @@ public class RabbitMessageChannelBinder
}
listenerContainer.setTxSize(properties.getExtension().getTxSize());
if (properties.getExtension().getQueueDeclarationRetries() != null) {
listenerContainer.setDeclarationRetries(properties.getExtension().getQueueDeclarationRetries());
listenerContainer.setDeclarationRetries(
properties.getExtension().getQueueDeclarationRetries());
}
}
private void setDMLCProperties(ExtendedConsumerProperties<RabbitConsumerProperties> properties,
private void setDMLCProperties(
ExtendedConsumerProperties<RabbitConsumerProperties> properties,
DirectMessageListenerContainer listenerContainer, int concurrency) {
listenerContainer.setConsumersPerQueue(concurrency);
if (properties.getExtension().getMaxConcurrency() > concurrency) {
this.logger.warn("maxConcurrency is not supported with a direct container type");
this.logger
.warn("maxConcurrency is not supported with a direct container type");
}
if (properties.getExtension().getTxSize() > 1) {
this.logger.warn("txSize is not supported with a direct container type");
}
if (properties.getExtension().getQueueDeclarationRetries() != null) {
this.logger.warn("queueDeclarationRetries is not supported with a direct container type");
this.logger.warn(
"queueDeclarationRetries is not supported with a direct container type");
}
}
@Override
protected PolledConsumerResources createPolledConsumerResources(String name, String group, ConsumerDestination destination,
protected PolledConsumerResources createPolledConsumerResources(String name,
String group, ConsumerDestination destination,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
Assert.isTrue(!consumerProperties.isMultiplex(),
"The Spring Integration polled MessageSource does not currently support muiltiple queues");
AmqpMessageSource source = new AmqpMessageSource(this.connectionFactory, destination.getName());
AmqpMessageSource source = new AmqpMessageSource(this.connectionFactory,
destination.getName());
source.setRawMessageHeader(true);
return new PolledConsumerResources(source,
registerErrorInfrastructure(destination, group, consumerProperties, true));
return new PolledConsumerResources(source, registerErrorInfrastructure(
destination, group, consumerProperties, true));
}
@Override
protected void postProcessPollableSource(DefaultPollableMessageSource bindingTarget) {
bindingTarget.setAttributesProvider((accessor, message) -> {
Object rawMessage = message.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
Object rawMessage = message.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
if (rawMessage != null) {
accessor.setAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, rawMessage);
accessor.setAttribute(
AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE,
rawMessage);
}
});
}
@@ -509,7 +566,8 @@ public class RabbitMessageChannelBinder
}
@Override
protected MessageHandler getErrorMessageHandler(ConsumerDestination destination, String group,
protected MessageHandler getErrorMessageHandler(ConsumerDestination destination,
String group,
final ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
if (properties.getExtension().isRepublishToDlq()) {
return new MessageHandler() {
@@ -521,21 +579,26 @@ public class RabbitMessageChannelBinder
this.template.setUsePublisherConnection(true);
}
private final String exchange = deadLetterExchangeName(properties.getExtension());
private final String exchange = deadLetterExchangeName(
properties.getExtension());
private final String routingKey = properties.getExtension().getDeadLetterRoutingKey();
private final String routingKey = properties.getExtension()
.getDeadLetterRoutingKey();
private final int frameMaxHeadroom = properties.getExtension().getFrameMaxHeadroom();
private final int frameMaxHeadroom = properties.getExtension()
.getFrameMaxHeadroom();
private int maxStackTraceLength = -1;
@Override
public void handleMessage(org.springframework.messaging.Message<?> message) throws MessagingException {
public void handleMessage(
org.springframework.messaging.Message<?> message)
throws MessagingException {
Message amqpMessage = (Message) message.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
if (!(message instanceof ErrorMessage)) {
logger.error("Expected an ErrorMessage, not a " + message.getClass().toString() + " for: "
+ message);
logger.error("Expected an ErrorMessage, not a "
+ message.getClass().toString() + " for: " + message);
}
else if (amqpMessage == null) {
logger.error("No raw message header in " + message);
@@ -548,34 +611,44 @@ public class RabbitMessageChannelBinder
}
return;
}
MessageProperties messageProperties = amqpMessage.getMessageProperties();
MessageProperties messageProperties = amqpMessage
.getMessageProperties();
Map<String, Object> headers = messageProperties.getHeaders();
String stackTraceAsString = getStackTraceAsString(cause);
if (this.maxStackTraceLength < 0) {
int rabbitMaxStackTraceLength = RabbitUtils
.getMaxFrame(this.template.getConnectionFactory());
if (rabbitMaxStackTraceLength > 0) {
//maxStackTraceLength -= this.frameMaxHeadroom;
this.maxStackTraceLength = rabbitMaxStackTraceLength - this.frameMaxHeadroom;
// maxStackTraceLength -= this.frameMaxHeadroom;
this.maxStackTraceLength = rabbitMaxStackTraceLength
- this.frameMaxHeadroom;
}
}
if (this.maxStackTraceLength > 0 && stackTraceAsString.length() > this.maxStackTraceLength) {
stackTraceAsString = stackTraceAsString.substring(0, this.maxStackTraceLength);
logger.warn("Stack trace in republished message header truncated due to frame_max limitations; "
+ "consider increasing frame_max on the broker or reduce the stack trace depth", cause);
if (this.maxStackTraceLength > 0 && stackTraceAsString
.length() > this.maxStackTraceLength) {
stackTraceAsString = stackTraceAsString.substring(0,
this.maxStackTraceLength);
logger.warn(
"Stack trace in republished message header truncated due to frame_max limitations; "
+ "consider increasing frame_max on the broker or reduce the stack trace depth",
cause);
}
headers.put(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE, stackTraceAsString);
headers.put(RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE,
stackTraceAsString);
headers.put(RepublishMessageRecoverer.X_EXCEPTION_MESSAGE,
cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage());
cause.getCause() != null ? cause.getCause().getMessage()
: cause.getMessage());
headers.put(RepublishMessageRecoverer.X_ORIGINAL_EXCHANGE,
messageProperties.getReceivedExchange());
headers.put(RepublishMessageRecoverer.X_ORIGINAL_ROUTING_KEY,
messageProperties.getReceivedRoutingKey());
if (properties.getExtension().getRepublishDeliveyMode() != null) {
messageProperties.setDeliveryMode(properties.getExtension().getRepublishDeliveyMode());
messageProperties.setDeliveryMode(
properties.getExtension().getRepublishDeliveyMode());
}
this.template.send(this.exchange,
this.routingKey != null ? this.routingKey : messageProperties.getConsumerQueue(),
this.routingKey != null ? this.routingKey
: messageProperties.getConsumerQueue(),
amqpMessage);
}
}
@@ -589,8 +662,9 @@ public class RabbitMessageChannelBinder
*/
private boolean shouldRepublish(Throwable throwable) {
Throwable cause = throwable;
while (cause != null && !(cause instanceof AmqpRejectAndDontRequeueException)
&& !(cause instanceof ImmediateAcknowledgeAmqpException)) {
while (cause != null
&& !(cause instanceof AmqpRejectAndDontRequeueException)
&& !(cause instanceof ImmediateAcknowledgeAmqpException)) {
cause = cause.getCause();
}
return !(cause instanceof ImmediateAcknowledgeAmqpException);
@@ -603,29 +677,35 @@ public class RabbitMessageChannelBinder
private final RejectAndDontRequeueRecoverer recoverer = new RejectAndDontRequeueRecoverer();
@Override
public void handleMessage(org.springframework.messaging.Message<?> message) throws MessagingException {
public void handleMessage(
org.springframework.messaging.Message<?> message)
throws MessagingException {
Message amqpMessage = (Message) message.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
/*
* NOTE: The following IF and subsequent ELSE IF should never happen under normal interaction and
* it should always go to the last ELSE
* However, given that this is a handler subscribing to the public channel and that we can't control what
* type of Message may be sent to that channel (user decides to send a Message manually) the 'IF/ELSE IF' provides
* a safety net to handle any message properly.
* NOTE: The following IF and subsequent ELSE IF should never happen
* under normal interaction and it should always go to the last ELSE
* However, given that this is a handler subscribing to the public
* channel and that we can't control what type of Message may be sent
* to that channel (user decides to send a Message manually) the
* 'IF/ELSE IF' provides a safety net to handle any message properly.
*/
if (!(message instanceof ErrorMessage)) {
logger.error("Expected an ErrorMessage, not a " + message.getClass().toString() + " for: "
+ message);
throw new ListenerExecutionFailedException("Unexpected error message " + message,
logger.error("Expected an ErrorMessage, not a "
+ message.getClass().toString() + " for: " + message);
throw new ListenerExecutionFailedException(
"Unexpected error message " + message,
new AmqpRejectAndDontRequeueException(""), null);
}
else if (amqpMessage == null) {
logger.error("No raw message header in " + message);
throw new ListenerExecutionFailedException("Unexpected error message " + message,
throw new ListenerExecutionFailedException(
"Unexpected error message " + message,
new AmqpRejectAndDontRequeueException(""), amqpMessage);
}
else {
this.recoverer.recover(amqpMessage, (Throwable) message.getPayload());
this.recoverer.recover(amqpMessage,
(Throwable) message.getPayload());
}
}
@@ -636,21 +716,22 @@ public class RabbitMessageChannelBinder
}
}
@Override
protected MessageHandler getPolledConsumerErrorMessageHandler(ConsumerDestination destination, String group,
protected MessageHandler getPolledConsumerErrorMessageHandler(
ConsumerDestination destination, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
MessageHandler handler = getErrorMessageHandler(destination, group, properties);
if (handler != null) {
return handler;
}
final MessageHandler superHandler = super.getErrorMessageHandler(destination, group, properties);
final MessageHandler superHandler = super.getErrorMessageHandler(destination,
group, properties);
return message -> {
Message amqpMessage = (Message) message.getHeaders()
.get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
if (!(message instanceof ErrorMessage)) {
logger.error("Expected an ErrorMessage, not a " + message.getClass().toString() + " for: "
+ message);
logger.error("Expected an ErrorMessage, not a "
+ message.getClass().toString() + " for: " + message);
}
else if (amqpMessage == null) {
if (superHandler != null) {
@@ -659,8 +740,10 @@ public class RabbitMessageChannelBinder
}
else {
if (message.getPayload() instanceof MessagingException) {
AcknowledgmentCallback ack = StaticMessageHeaderAccessor.getAcknowledgmentCallback(
((MessagingException) message.getPayload()).getFailedMessage());
AcknowledgmentCallback ack = StaticMessageHeaderAccessor
.getAcknowledgmentCallback(
((MessagingException) message.getPayload())
.getFailedMessage());
if (ack != null) {
if (properties.getExtension().isRequeueRejected()) {
ack.acknowledge(Status.REQUEUE);
@@ -682,7 +765,8 @@ public class RabbitMessageChannelBinder
private String deadLetterExchangeName(RabbitCommonProperties properties) {
if (properties.getDeadLetterExchange() == null) {
return applyPrefix(properties.getPrefix(), RabbitCommonProperties.DEAD_LETTER_EXCHANGE);
return applyPrefix(properties.getPrefix(),
RabbitCommonProperties.DEAD_LETTER_EXCHANGE);
}
else {
return properties.getDeadLetterExchange();
@@ -690,20 +774,23 @@ public class RabbitMessageChannelBinder
}
@Override
protected void afterUnbindConsumer(ConsumerDestination consumerDestination, String group,
protected void afterUnbindConsumer(ConsumerDestination consumerDestination,
String group,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
provisioningProvider.cleanAutoDeclareContext(consumerDestination, consumerProperties);
provisioningProvider.cleanAutoDeclareContext(consumerDestination,
consumerProperties);
}
private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties, boolean mandatory) {
private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties,
boolean mandatory) {
RabbitTemplate rabbitTemplate;
if (properties.isBatchingEnabled()) {
BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(
properties.getBatchSize(),
properties.getBatchBufferLimit(),
properties.getBatchSize(), properties.getBatchBufferLimit(),
properties.getBatchTimeout());
rabbitTemplate = new BatchingRabbitTemplate(batchingStrategy,
getApplicationContext().getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME,
getApplicationContext().getBean(
IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME,
TaskScheduler.class));
}
else {
@@ -717,7 +804,8 @@ public class RabbitMessageChannelBinder
rabbitTemplate.setBeforePublishPostProcessors(this.compressingPostProcessor);
}
rabbitTemplate.setMandatory(mandatory); // returned messages
if (rabbitProperties != null && rabbitProperties.getTemplate().getRetry().isEnabled()) {
if (rabbitProperties != null
&& rabbitProperties.getTemplate().getRetry().isEnabled()) {
Retry retry = rabbitProperties.getTemplate().getRetry();
RetryPolicy retryPolicy = new SimpleRetryPolicy(retry.getMaxAttempts());
ExponentialBackOffPolicy backOff = new ExponentialBackOffPolicy();
@@ -740,7 +828,8 @@ public class RabbitMessageChannelBinder
return stringWriter.getBuffer().toString();
}
private static final class SimplePassthroughMessageConverter extends AbstractMessageConverter {
private static final class SimplePassthroughMessageConverter
extends AbstractMessageConverter {
private static final SimpleMessageConverter converter = new SimpleMessageConverter();
@@ -749,7 +838,8 @@ public class RabbitMessageChannelBinder
}
@Override
protected Message createMessage(Object object, MessageProperties messageProperties) {
protected Message createMessage(Object object,
MessageProperties messageProperties) {
if (object instanceof byte[]) {
return new Message((byte[]) object, messageProperties);
}

7
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ExtendedBindingHandlerMappingsProviderConfiguration.java
View File

@@ -25,20 +25,21 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
*
* @author Oleg Zhurakousky
*
*/
@Configuration
public class ExtendedBindingHandlerMappingsProviderConfiguration {
public class ExtendedBindingHandlerMappingsProviderConfiguration {
@Bean
public MappingsProvider rabbitExtendedPropertiesDefaultMappingsProvider() {
return () -> {
Map<ConfigurationPropertyName, ConfigurationPropertyName> mappings = new HashMap<>();
mappings.put(ConfigurationPropertyName.of("spring.cloud.stream.rabbit.bindings"),
mappings.put(
ConfigurationPropertyName.of("spring.cloud.stream.rabbit.bindings"),
ConfigurationPropertyName.of("spring.cloud.stream.rabbit.default"));
return mappings;
};
}
}

22
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java
View File

@@ -52,7 +52,8 @@ import org.springframework.lang.Nullable;
*/
@Configuration
@Import({ PropertyPlaceholderAutoConfiguration.class })
@EnableConfigurationProperties({ RabbitBinderConfigurationProperties.class, RabbitExtendedBindingProperties.class })
@EnableConfigurationProperties({ RabbitBinderConfigurationProperties.class,
RabbitExtendedBindingProperties.class })
public class RabbitMessageChannelBinderConfiguration {
@Autowired
@@ -68,10 +69,14 @@ public class RabbitMessageChannelBinderConfiguration {
private RabbitExtendedBindingProperties rabbitExtendedBindingProperties;
@Bean
RabbitMessageChannelBinder rabbitMessageChannelBinder(@Nullable ListenerContainerCustomizer<AbstractMessageListenerContainer> listenerContainerCustomizer) throws Exception {
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(this.rabbitConnectionFactory,
this.rabbitProperties, provisioningProvider(), listenerContainerCustomizer);
binder.setAdminAddresses(this.rabbitBinderConfigurationProperties.getAdminAddresses());
RabbitMessageChannelBinder rabbitMessageChannelBinder(
@Nullable ListenerContainerCustomizer<AbstractMessageListenerContainer> listenerContainerCustomizer)
throws Exception {
RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(
this.rabbitConnectionFactory, this.rabbitProperties,
provisioningProvider(), listenerContainerCustomizer);
binder.setAdminAddresses(
this.rabbitBinderConfigurationProperties.getAdminAddresses());
binder.setCompressingPostProcessor(gZipPostProcessor());
binder.setDecompressingPostProcessor(deCompressingPostProcessor());
binder.setNodes(this.rabbitBinderConfigurationProperties.getNodes());
@@ -87,7 +92,8 @@ public class RabbitMessageChannelBinderConfiguration {
@Bean
MessagePostProcessor gZipPostProcessor() {
GZipPostProcessor gZipPostProcessor = new GZipPostProcessor();
gZipPostProcessor.setLevel(this.rabbitBinderConfigurationProperties.getCompressionLevel());
gZipPostProcessor
.setLevel(this.rabbitBinderConfigurationProperties.getCompressionLevel());
return gZipPostProcessor;
}
@@ -101,8 +107,8 @@ public class RabbitMessageChannelBinderConfiguration {
@ConditionalOnProperty("spring.cloud.stream.rabbit.binder.connection-name-prefix")
public ConnectionNameStrategy connectionNamer(CachingConnectionFactory cf) {
final AtomicInteger nameIncrementer = new AtomicInteger();
ConnectionNameStrategy namer = f -> this.rabbitBinderConfigurationProperties.getConnectionNamePrefix()
+ "#" + nameIncrementer.getAndIncrement();
ConnectionNameStrategy namer = f -> this.rabbitBinderConfigurationProperties
.getConnectionNamePrefix() + "#" + nameIncrementer.getAndIncrement();
// TODO: this can be removed when Boot 2.0.1 wires it in
cf.setConnectionNameStrategy(namer);
return namer;

42
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitServiceAutoConfiguration.java
View File

@@ -68,8 +68,8 @@ public class RabbitServiceAutoConfiguration {
protected static class CloudProfile {
/**
* Configuration to be used when the cloud profile is set, and Cloud Connectors are found
* on the classpath.
* Configuration to be used when the cloud profile is set, and Cloud Connectors
* are found on the classpath.
*/
@Configuration
@ConditionalOnClass(Cloud.class)
@@ -82,8 +82,8 @@ public class RabbitServiceAutoConfiguration {
}
/**
* Active only if {@code spring.cloud.stream.override-cloud-connectors} is not set to
* {@code true}.
* Active only if {@code spring.cloud.stream.override-cloud-connectors} is not
* set to {@code true}.
*/
@Configuration
@ConditionalOnProperty(value = "spring.cloud.stream.override-cloud-connectors", havingValue = "false", matchIfMissing = true)
@@ -94,7 +94,8 @@ public class RabbitServiceAutoConfiguration {
protected static class UseCloudConnectors {
/**
* Creates a {@link ConnectionFactory} using the singleton service connector.
* Creates a {@link ConnectionFactory} using the singleton service
* connector.
* @param cloud {@link Cloud} instance to be used for accessing services.
* @param connectorConfigObjectProvider the {@link ObjectProvider} for the
* {@link RabbitConnectionFactoryConfig}.
@@ -107,10 +108,12 @@ public class RabbitServiceAutoConfiguration {
ConfigurableApplicationContext applicationContext,
RabbitProperties rabbitProperties) throws Exception {
ConnectionFactory connectionFactory = cloud.getSingletonServiceConnector(ConnectionFactory.class,
connectorConfigObjectProvider.getIfUnique());
ConnectionFactory connectionFactory = cloud
.getSingletonServiceConnector(ConnectionFactory.class,
connectorConfigObjectProvider.getIfUnique());
configureCachingConnectionFactory((CachingConnectionFactory) connectionFactory,
configureCachingConnectionFactory(
(CachingConnectionFactory) connectionFactory,
applicationContext, rabbitProperties);
return connectionFactory;
@@ -121,11 +124,13 @@ public class RabbitServiceAutoConfiguration {
RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
return new RabbitTemplate(connectionFactory);
}
}
/**
* Configuration to be used if {@code spring.cloud.stream.override-cloud-connectors} is set
* to {@code true}. Defers to Spring Boot auto-configuration.
* Configuration to be used if
* {@code spring.cloud.stream.override-cloud-connectors} is set to
* {@code true}. Defers to Spring Boot auto-configuration.
*/
@Configuration
@ConditionalOnProperty("spring.cloud.stream.override-cloud-connectors")
@@ -153,6 +158,7 @@ public class RabbitServiceAutoConfiguration {
@Profile("!cloud")
@Import(RabbitAutoConfiguration.class)
protected static class NoCloudProfile {
}
@Configuration
@@ -166,8 +172,10 @@ public class RabbitServiceAutoConfiguration {
}
static void configureCachingConnectionFactory(CachingConnectionFactory connectionFactory,
ConfigurableApplicationContext applicationContext, RabbitProperties rabbitProperties) throws Exception {
static void configureCachingConnectionFactory(
CachingConnectionFactory connectionFactory,
ConfigurableApplicationContext applicationContext,
RabbitProperties rabbitProperties) throws Exception {
if (StringUtils.hasText(rabbitProperties.getAddresses())) {
connectionFactory.setAddresses(rabbitProperties.determineAddresses());
@@ -176,18 +184,20 @@ public class RabbitServiceAutoConfiguration {
connectionFactory.setPublisherConfirms(rabbitProperties.isPublisherConfirms());
connectionFactory.setPublisherReturns(rabbitProperties.isPublisherReturns());
if (rabbitProperties.getCache().getChannel().getSize() != null) {
connectionFactory.setChannelCacheSize(rabbitProperties.getCache().getChannel().getSize());
connectionFactory.setChannelCacheSize(
rabbitProperties.getCache().getChannel().getSize());
}
if (rabbitProperties.getCache().getConnection().getMode() != null) {
connectionFactory.setCacheMode(rabbitProperties.getCache().getConnection().getMode());
connectionFactory
.setCacheMode(rabbitProperties.getCache().getConnection().getMode());
}
if (rabbitProperties.getCache().getConnection().getSize() != null) {
connectionFactory.setConnectionCacheSize(
rabbitProperties.getCache().getConnection().getSize());
}
if (rabbitProperties.getCache().getChannel().getCheckoutTimeout() != null) {
connectionFactory.setChannelCheckoutTimeout(
rabbitProperties.getCache().getChannel().getCheckoutTimeout().toMillis());
connectionFactory.setChannelCheckoutTimeout(rabbitProperties.getCache()
.getChannel().getCheckoutTimeout().toMillis());
}
connectionFactory.setApplicationContext(applicationContext);
applicationContext.addApplicationListener(connectionFactory);

13
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java
View File

@@ -32,9 +32,7 @@ import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSuppo
import static org.assertj.core.api.Assertions.assertThat;
/**
*
* @author Gary Russell
*/
public class LocalizedQueueConnectionFactoryIntegrationTests {
@@ -48,13 +46,15 @@ public class LocalizedQueueConnectionFactoryIntegrationTests {
public void setup() {
ConnectionFactory defaultConnectionFactory = rabbitAvailableRule.getResource();
String[] addresses = new String[] { "localhost:9999", "localhost:5672" };
String[] adminAddresses = new String[] { "http://localhost:15672", "http://localhost:15672" };
String[] adminAddresses = new String[] { "http://localhost:15672",
"http://localhost:15672" };
String[] nodes = new String[] { "foo@bar", "rabbit@localhost" };
String vhost = "/";
String username = "guest";
String password = "guest";
this.lqcf = new LocalizedQueueConnectionFactory(defaultConnectionFactory, addresses,
adminAddresses, nodes, vhost, username, password, false, null, null, null, null);
this.lqcf = new LocalizedQueueConnectionFactory(defaultConnectionFactory,
addresses, adminAddresses, nodes, vhost, username, password, false, null,
null, null, null);
}
@Test
@@ -62,7 +62,8 @@ public class LocalizedQueueConnectionFactoryIntegrationTests {
RabbitAdmin admin = new RabbitAdmin(this.lqcf);
Queue queue = new Queue(UUID.randomUUID().toString(), false, false, true);
admin.declareQueue(queue);
ConnectionFactory targetConnectionFactory = this.lqcf.getTargetConnectionFactory("[" + queue.getName() + "]");
ConnectionFactory targetConnectionFactory = this.lqcf
.getTargetConnectionFactory("[" + queue.getName() + "]");
RabbitTemplate template = new RabbitTemplate(targetConnectionFactory);
template.convertAndSend("", queue.getName(), "foo");
assertThat(template.receiveAndConvert(queue.getName())).isEqualTo("foo");

79
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java
View File

@@ -59,64 +59,74 @@ public class RabbitBinderCleanerTests {
@Test
public void testCleanStream() {
final RabbitBindingCleaner cleaner = new RabbitBindingCleaner();
final RestTemplate template = RabbitManagementUtils.buildRestTemplate("http://localhost:15672", "guest",
"guest");
final RestTemplate template = RabbitManagementUtils
.buildRestTemplate("http://localhost:15672", "guest", "guest");
final String stream1 = UUID.randomUUID().toString();
String stream2 = stream1 + "-1";
String firstQueue = null;
CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource();
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
for (int i = 0; i < 5; i++) {
String queue1Name = AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".default." + i);
String queue2Name = AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".default." + i);
String queue1Name = AbstractBinder.applyPrefix(BINDER_PREFIX,
stream1 + ".default." + i);
String queue2Name = AbstractBinder.applyPrefix(BINDER_PREFIX,
stream2 + ".default." + i);
if (firstQueue == null) {
firstQueue = queue1Name;
}
URI uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}")
.buildAndExpand("/", queue1Name)
URI uri = UriComponentsBuilder
.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}").buildAndExpand("/", queue1Name)
.encode().toUri();
template.put(uri, new AmqpQueue(false, true));
uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}").buildAndExpand("/", queue2Name)
.encode().toUri();
template.put(uri, new AmqpQueue(false, true));
uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}")
.buildAndExpand("/", queue2Name)
.buildAndExpand("/", AbstractBinder.constructDLQName(queue1Name))
.encode().toUri();
template.put(uri, new AmqpQueue(false, true));
uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}")
.buildAndExpand("/", AbstractBinder.constructDLQName(queue1Name)).encode().toUri();
template.put(uri, new AmqpQueue(false, true));
TopicExchange exchange = new TopicExchange(queue1Name);
rabbitAdmin.declareExchange(exchange);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue1Name)).to(exchange).with(queue1Name));
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue1Name))
.to(exchange).with(queue1Name));
exchange = new TopicExchange(queue2Name);
rabbitAdmin.declareExchange(exchange);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue2Name)).to(exchange).with(queue2Name));
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(queue2Name))
.to(exchange).with(queue2Name));
}
final TopicExchange topic1 = new TopicExchange(
AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".foo.bar"));
rabbitAdmin.declareExchange(topic1);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(firstQueue)).to(topic1).with("#"));
rabbitAdmin.declareBinding(
BindingBuilder.bind(new Queue(firstQueue)).to(topic1).with("#"));
String foreignQueue = UUID.randomUUID().toString();
rabbitAdmin.declareQueue(new Queue(foreignQueue));
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(foreignQueue)).to(topic1).with("#"));
rabbitAdmin.declareBinding(
BindingBuilder.bind(new Queue(foreignQueue)).to(topic1).with("#"));
final TopicExchange topic2 = new TopicExchange(
AbstractBinder.applyPrefix(BINDER_PREFIX, stream2 + ".foo.bar"));
rabbitAdmin.declareExchange(topic2);
rabbitAdmin.declareBinding(BindingBuilder.bind(new Queue(firstQueue)).to(topic2).with("#"));
rabbitAdmin.declareBinding(
BindingBuilder.bind(new Queue(firstQueue)).to(topic2).with("#"));
new RabbitTemplate(connectionFactory).execute(new ChannelCallback<Void>() {
@Override
public Void doInRabbit(Channel channel) throws Exception {
String queueName = AbstractBinder.applyPrefix(BINDER_PREFIX, stream1 + ".default." + 4);
String consumerTag = channel.basicConsume(queueName, new DefaultConsumer(channel));
String queueName = AbstractBinder.applyPrefix(BINDER_PREFIX,
stream1 + ".default." + 4);
String consumerTag = channel.basicConsume(queueName,
new DefaultConsumer(channel));
try {
waitForConsumerStateNot(queueName, 0);
cleaner.clean(stream1, false);
fail("Expected exception");
}
catch (RabbitAdminException e) {
assertThat(e).hasMessageContaining("Queue " + queueName + " is in use");
assertThat(e)
.hasMessageContaining("Queue " + queueName + " is in use");
}
channel.basicCancel(consumerTag);
waitForConsumerStateNot(queueName, 1);
@@ -131,14 +141,17 @@ public class RabbitBinderCleanerTests {
return null;
}
private void waitForConsumerStateNot(String queueName, int state) throws InterruptedException {
private void waitForConsumerStateNot(String queueName, int state)
throws InterruptedException {
int n = 0;
URI uri = UriComponentsBuilder.fromUriString("http://localhost:15672/api/queues").pathSegment(
"{vhost}", "{queue}")
.buildAndExpand("/", queueName).encode().toUri();
URI uri = UriComponentsBuilder
.fromUriString("http://localhost:15672/api/queues")
.pathSegment("{vhost}", "{queue}").buildAndExpand("/", queueName)
.encode().toUri();
Object consumers = null;
while (n++ < 100 && (consumers == null || consumers.equals(Integer.valueOf(state)))) {
while (n++ < 100 && (consumers == null
|| consumers.equals(Integer.valueOf(state)))) {
Map<String, Object> queueInfo = template.getForObject(uri, Map.class);
consumers = queueInfo.get("consumers");
if (consumers == null || consumers.equals(Integer.valueOf(state))) {
@@ -147,7 +160,8 @@ public class RabbitBinderCleanerTests {
}
assertThat(consumers).isNotNull();
assertThat(n).withFailMessage("Consumer state remained at " + state + " after 10 seconds")
assertThat(n).withFailMessage(
"Consumer state remained at " + state + " after 10 seconds")
.isLessThan(100);
}
@@ -162,8 +176,10 @@ public class RabbitBinderCleanerTests {
// should *not* clean stream2
assertThat(cleanedQueues).hasSize(10);
for (int i = 0; i < 5; i++) {
assertThat(cleanedQueues.get(i * 2)).isEqualTo(BINDER_PREFIX + stream1 + ".default." + i);
assertThat(cleanedQueues.get(i * 2 + 1)).isEqualTo(BINDER_PREFIX + stream1 + ".default." + i + ".dlq");
assertThat(cleanedQueues.get(i * 2))
.isEqualTo(BINDER_PREFIX + stream1 + ".default." + i);
assertThat(cleanedQueues.get(i * 2 + 1))
.isEqualTo(BINDER_PREFIX + stream1 + ".default." + i + ".dlq");
}
List<String> cleanedExchanges = cleanedMap.get("exchanges");
assertThat(cleanedExchanges).hasSize(6);
@@ -174,7 +190,8 @@ public class RabbitBinderCleanerTests {
cleanedQueues = cleanedMap.get("queues");
assertThat(cleanedQueues).hasSize(5);
for (int i = 0; i < 5; i++) {
assertThat(cleanedQueues.get(i)).isEqualTo(BINDER_PREFIX + stream2 + ".default." + i);
assertThat(cleanedQueues.get(i))
.isEqualTo(BINDER_PREFIX + stream2 + ".default." + i);
}
cleanedExchanges = cleanedMap.get("exchanges");
assertThat(cleanedExchanges).hasSize(6);
@@ -191,23 +208,19 @@ public class RabbitBinderCleanerTests {
this.durable = durable;
}
@JsonProperty("auto_delete")
protected boolean isAutoDelete() {
return autoDelete;
}
protected void setAutoDelete(boolean autoDelete) {
this.autoDelete = autoDelete;
}
protected boolean isDurable() {
return durable;
}
protected void setDurable(boolean durable) {
this.durable = durable;
}

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

File diff suppressed because it is too large Load Diff

32
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java
View File

@@ -47,8 +47,7 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
*/
public class RabbitTestBinder extends
AbstractPollableConsumerTestBinder<RabbitMessageChannelBinder,
ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
AbstractPollableConsumerTestBinder<RabbitMessageChannelBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
private final RabbitAdmin rabbitAdmin;
@@ -60,12 +59,14 @@ public class RabbitTestBinder extends
private final AnnotationConfigApplicationContext applicationContext;
public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) {
this(connectionFactory, new RabbitMessageChannelBinder(connectionFactory, rabbitProperties,
new RabbitExchangeQueueProvisioner(connectionFactory)));
public RabbitTestBinder(ConnectionFactory connectionFactory,
RabbitProperties rabbitProperties) {
this(connectionFactory, new RabbitMessageChannelBinder(connectionFactory,
rabbitProperties, new RabbitExchangeQueueProvisioner(connectionFactory)));
}
public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitMessageChannelBinder binder) {
public RabbitTestBinder(ConnectionFactory connectionFactory,
RabbitMessageChannelBinder binder) {
this.applicationContext = new AnnotationConfigApplicationContext(Config.class);
binder.setApplicationContext(this.applicationContext);
this.setPollableConsumerBinder(binder);
@@ -77,15 +78,16 @@ public class RabbitTestBinder extends
}
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel,
public Binding<MessageChannel> bindConsumer(String name, String group,
MessageChannel moduleInputChannel,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
captureConsumerResources(name, group, properties);
return super.bindConsumer(name, group, moduleInputChannel, properties);
}
@Override
public Binding<PollableSource<MessageHandler>> bindPollableConsumer(String name, String group,
PollableSource<MessageHandler> inboundBindTarget,
public Binding<PollableSource<MessageHandler>> bindPollableConsumer(String name,
String group, PollableSource<MessageHandler> inboundBindTarget,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
captureConsumerResources(name, group, properties);
return super.bindPollableConsumer(name, group, inboundBindTarget, properties);
@@ -102,11 +104,13 @@ public class RabbitTestBinder extends
if (properties.isMultiplex()) {
names = StringUtils.commaDelimitedListToStringArray(name);
for (String nayme : names) {
this.queues.add(properties.getExtension().getPrefix() + nayme.trim() + "." + group);
this.queues.add(properties.getExtension().getPrefix()
+ nayme.trim() + "." + group);
}
}
else {
this.queues.add(properties.getExtension().getPrefix() + name + "." + group);
this.queues.add(
properties.getExtension().getPrefix() + name + "." + group);
}
}
}
@@ -123,7 +127,8 @@ public class RabbitTestBinder extends
}
@Override
public Binding<MessageChannel> bindProducer(String name, MessageChannel moduleOutputChannel,
public Binding<MessageChannel> bindProducer(String name,
MessageChannel moduleOutputChannel,
ExtendedProducerProperties<RabbitProducerProperties> properties) {
this.queues.add(properties.getExtension().getPrefix() + name + ".default");
this.exchanges.add(properties.getExtension().getPrefix() + name);
@@ -133,7 +138,8 @@ public class RabbitTestBinder extends
this.queues.add(properties.getExtension().getPrefix() + group);
}
else {
this.queues.add(properties.getExtension().getPrefix() + name + "." + group);
this.queues.add(
properties.getExtension().getPrefix() + name + "." + group);
}
}
}

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

@@ -78,8 +78,8 @@ public class RabbitBinderModuleTests {
private ConfigurableApplicationContext context;
public static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock(ConnectionFactory.class,
Mockito.RETURNS_MOCKS);
public static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock(
ConnectionFactory.class, Mockito.RETURNS_MOCKS);
@After
public void tearDown() {
@@ -95,8 +95,8 @@ public class RabbitBinderModuleTests {
@Test
public void testParentConnectionFactoryInheritedByDefault() {
context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
.run("--server.port=0", "--spring.cloud.stream.rabbit.binder.connection-name-prefix=foo");
.web(WebApplicationType.NONE).run("--server.port=0",
"--spring.cloud.stream.rabbit.binder.connection-name-prefix=foo");
BinderFactory binderFactory = context.getBean(BinderFactory.class);
Binder<?, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
@@ -107,29 +107,35 @@ public class RabbitBinderModuleTests {
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
CompositeHealthIndicator bindersHealthIndicator = context
.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(
bindersHealthIndicator);
assertThat(bindersHealthIndicator).isNotNull();
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("registry.healthIndicators");
assertThat(healthIndicators).containsKey(("rabbit"));
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo((Status.UP));
assertThat(healthIndicators.get("rabbit").health().getStatus())
.isEqualTo((Status.UP));
ConnectionFactory publisherConnectionFactory = binderConnectionFactory.getPublisherConnectionFactory();
assertThat(TestUtils.getPropertyValue(publisherConnectionFactory, "connection.target")).isNull();
ConnectionFactory publisherConnectionFactory = binderConnectionFactory
.getPublisherConnectionFactory();
assertThat(TestUtils.getPropertyValue(publisherConnectionFactory,
"connection.target")).isNull();
DirectChannel checkPf = new DirectChannel();
Binding<MessageChannel> binding = ((RabbitMessageChannelBinder) binder).bindProducer("checkPF", checkPf,
new ExtendedProducerProperties<>(
new RabbitProducerProperties()));
Binding<MessageChannel> binding = ((RabbitMessageChannelBinder) binder)
.bindProducer("checkPF", checkPf,
new ExtendedProducerProperties<>(new RabbitProducerProperties()));
checkPf.send(new GenericMessage<>("foo".getBytes()));
binding.unbind();
assertThat(TestUtils.getPropertyValue(publisherConnectionFactory, "connection.target")).isNotNull();
assertThat(TestUtils.getPropertyValue(publisherConnectionFactory,
"connection.target")).isNotNull();
CachingConnectionFactory cf = this.context.getBean(CachingConnectionFactory.class);
ConnectionNameStrategy cns = TestUtils.getPropertyValue(cf, "connectionNameStrategy",
ConnectionNameStrategy.class);
CachingConnectionFactory cf = this.context
.getBean(CachingConnectionFactory.class);
ConnectionNameStrategy cns = TestUtils.getPropertyValue(cf,
"connectionNameStrategy", ConnectionNameStrategy.class);
assertThat(cns.obtainNewConnectionName(cf)).isEqualTo("foo#2");
new RabbitAdmin(rabbitTestSupport.getResource()).deleteExchange("checkPF");
}
@@ -138,8 +144,7 @@ public class RabbitBinderModuleTests {
@SuppressWarnings("unchecked")
public void testParentConnectionFactoryInheritedByDefaultAndRabbitSettingsPropagated() {
context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
.run("--server.port=0",
.web(WebApplicationType.NONE).run("--server.port=0",
"--spring.cloud.stream.bindings.input.group=someGroup",
"--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true",
"--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true");
@@ -147,46 +152,52 @@ public class RabbitBinderModuleTests {
Binder<?, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
BindingService bindingService = context.getBean(BindingService.class);
DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor(bindingService);
DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor(
bindingService);
Map<String, List<Binding<MessageChannel>>> consumerBindings = (Map<String, List<Binding<MessageChannel>>>) channelBindingServiceAccessor
.getPropertyValue("consumerBindings");
Binding<MessageChannel> inputBinding = consumerBindings.get("input").get(0);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(inputBinding,
"lifecycle.messageListenerContainer", SimpleMessageListenerContainer.class);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(
inputBinding, "lifecycle.messageListenerContainer",
SimpleMessageListenerContainer.class);
assertThat(TestUtils.getPropertyValue(container, "beanName"))
.isEqualTo("setByCustomizerForQueue:input.someGroup,andGroup:someGroup");
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class)).isTrue();
.isEqualTo("setByCustomizerForQueue:input.someGroup,andGroup:someGroup");
assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class))
.isTrue();
Map<String, Binding<MessageChannel>> producerBindings = (Map<String, Binding<MessageChannel>>) TestUtils
.getPropertyValue(bindingService, "producerBindings");
Binding<MessageChannel> outputBinding = producerBindings.get("output");
assertThat(TestUtils.getPropertyValue(outputBinding, "lifecycle.amqpTemplate.transactional",
Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(outputBinding,
"lifecycle.amqpTemplate.transactional", Boolean.class)).isTrue();
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
CompositeHealthIndicator bindersHealthIndicator = context
.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(
bindersHealthIndicator);
assertThat(bindersHealthIndicator).isNotNull();
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("registry.healthIndicators");
assertThat(healthIndicators).containsKey("rabbit");
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo(Status.UP);
assertThat(healthIndicators.get("rabbit").health().getStatus())
.isEqualTo(Status.UP);
CachingConnectionFactory cf = this.context.getBean(CachingConnectionFactory.class);
ConnectionNameStrategy cns = TestUtils.getPropertyValue(cf, "connectionNameStrategy",
ConnectionNameStrategy.class);
CachingConnectionFactory cf = this.context
.getBean(CachingConnectionFactory.class);
ConnectionNameStrategy cns = TestUtils.getPropertyValue(cf,
"connectionNameStrategy", ConnectionNameStrategy.class);
assertThat(cns.obtainNewConnectionName(cf)).startsWith("rabbitConnectionFactory");
}
@Test
public void testParentConnectionFactoryInheritedIfOverridden() {
context = new SpringApplicationBuilder(SimpleProcessor.class, ConnectionFactoryConfiguration.class)
.web(WebApplicationType.NONE)
.run("--server.port=0");
context = new SpringApplicationBuilder(SimpleProcessor.class,
ConnectionFactoryConfiguration.class).web(WebApplicationType.NONE)
.run("--server.port=0");
BinderFactory binderFactory = context.getBean(BinderFactory.class);
Binder<?, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
@@ -196,16 +207,18 @@ public class RabbitBinderModuleTests {
assertThat(binderConnectionFactory).isSameAs(MOCK_CONNECTION_FACTORY);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
CompositeHealthIndicator bindersHealthIndicator = context
.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
assertThat(bindersHealthIndicator).isNotNull();
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(
bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("registry.healthIndicators");
assertThat(healthIndicators).containsKey("rabbit");
// mock connection factory behaves as if down
assertThat(healthIndicators.get("rabbit").health().getStatus()).isEqualTo(Status.DOWN);
assertThat(healthIndicators.get("rabbit").health().getStatus())
.isEqualTo(Status.DOWN);
}
@Test
@@ -234,24 +247,27 @@ public class RabbitBinderModuleTests {
.getPropertyValue("connectionFactory");
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isNotSameAs(connectionFactory);
CompositeHealthIndicator bindersHealthIndicator = context.getBean("bindersHealthIndicator",
CompositeHealthIndicator.class);
CompositeHealthIndicator bindersHealthIndicator = context
.getBean("bindersHealthIndicator", CompositeHealthIndicator.class);
assertThat(bindersHealthIndicator);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(bindersHealthIndicator);
DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(
bindersHealthIndicator);
@SuppressWarnings("unchecked")
Map<String, HealthIndicator> healthIndicators = (Map<String, HealthIndicator>) directFieldAccessor
.getPropertyValue("registry.healthIndicators");
assertThat(healthIndicators).containsKey("custom");
assertThat(healthIndicators.get("custom").health().getStatus()).isEqualTo(Status.UP);
assertThat(healthIndicators.get("custom").health().getStatus())
.isEqualTo(Status.UP);
String name = UUID.randomUUID().toString();
Binding<MessageChannel> binding = binder.bindProducer(name, new DirectChannel(),
new ExtendedProducerProperties<>(new RabbitProducerProperties()));
RetryTemplate template = TestUtils.getPropertyValue(binding, "lifecycle.amqpTemplate.retryTemplate",
RetryTemplate.class);
RetryTemplate template = TestUtils.getPropertyValue(binding,
"lifecycle.amqpTemplate.retryTemplate", RetryTemplate.class);
assertThat(template).isNotNull();
SimpleRetryPolicy retryPolicy = TestUtils.getPropertyValue(template, "retryPolicy", SimpleRetryPolicy.class);
ExponentialBackOffPolicy backOff = TestUtils.getPropertyValue(template, "backOffPolicy",
ExponentialBackOffPolicy.class);
SimpleRetryPolicy retryPolicy = TestUtils.getPropertyValue(template,
"retryPolicy", SimpleRetryPolicy.class);
ExponentialBackOffPolicy backOff = TestUtils.getPropertyValue(template,
"backOffPolicy", ExponentialBackOffPolicy.class);
assertThat(retryPolicy.getMaxAttempts()).isEqualTo(2);
assertThat(backOff.getInitialInterval()).isEqualTo(1000L);
assertThat(backOff.getMultiplier()).isEqualTo(1.1);
@@ -263,22 +279,24 @@ public class RabbitBinderModuleTests {
@Test
public void testCloudProfile() {
this.context = new SpringApplicationBuilder(SimpleProcessor.class, MockCloudConfiguration.class)
.web(WebApplicationType.NONE)
.profiles("cloud")
.run();
this.context = new SpringApplicationBuilder(SimpleProcessor.class,
MockCloudConfiguration.class).web(WebApplicationType.NONE)
.profiles("cloud").run();
BinderFactory binderFactory = this.context.getBean(BinderFactory.class);
Binder<?, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
.getPropertyValue("connectionFactory");
ConnectionFactory connectionFactory = this.context.getBean(ConnectionFactory.class);
ConnectionFactory connectionFactory = this.context
.getBean(ConnectionFactory.class);
assertThat(binderConnectionFactory).isNotSameAs(connectionFactory);
assertThat(TestUtils.getPropertyValue(connectionFactory, "addresses")).isNotNull();
assertThat(TestUtils.getPropertyValue(binderConnectionFactory, "addresses")).isNull();
assertThat(TestUtils.getPropertyValue(connectionFactory, "addresses"))
.isNotNull();
assertThat(TestUtils.getPropertyValue(binderConnectionFactory, "addresses"))
.isNull();
Cloud cloud = this.context.getBean(Cloud.class);
@@ -288,24 +306,28 @@ public class RabbitBinderModuleTests {
@Test
public void testExtendedProperties() {
context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
.run("--server.port=0", "--spring.cloud.stream.rabbit.default.producer.routing-key-expression=fooRoutingKey",
.web(WebApplicationType.NONE).run("--server.port=0",
"--spring.cloud.stream.rabbit.default.producer.routing-key-expression=fooRoutingKey",
"--spring.cloud.stream.rabbit.bindings.output.producer.batch-size=512",
"--spring.cloud.stream.rabbit.default.consumer.max-concurrency=4",
"--spring.cloud.stream.rabbit.bindings.input.consumer.exchange-type=fanout");
BinderFactory binderFactory = context.getBean(BinderFactory.class);
Binder<?, ?, ?> rabbitBinder = binderFactory.getBinder(null, MessageChannel.class);
Binder<?, ?, ?> rabbitBinder = binderFactory.getBinder(null,
MessageChannel.class);
RabbitProducerProperties rabbitProducerProperties =
(RabbitProducerProperties)((ExtendedPropertiesBinder) rabbitBinder).getExtendedProducerProperties("output");
RabbitProducerProperties rabbitProducerProperties = (RabbitProducerProperties) ((ExtendedPropertiesBinder) rabbitBinder)
.getExtendedProducerProperties("output");
assertThat(rabbitProducerProperties.getRoutingKeyExpression().getExpressionString()).isEqualTo("fooRoutingKey");
assertThat(
rabbitProducerProperties.getRoutingKeyExpression().getExpressionString())
.isEqualTo("fooRoutingKey");
assertThat(rabbitProducerProperties.getBatchSize()).isEqualTo(512);
RabbitConsumerProperties rabbitConsumerProperties =
(RabbitConsumerProperties)((ExtendedPropertiesBinder) rabbitBinder).getExtendedConsumerProperties("input");
RabbitConsumerProperties rabbitConsumerProperties = (RabbitConsumerProperties) ((ExtendedPropertiesBinder) rabbitBinder)
.getExtendedConsumerProperties("input");
assertThat(rabbitConsumerProperties.getExchangeType()).isEqualTo(ExchangeTypes.FANOUT);
assertThat(rabbitConsumerProperties.getExchangeType())
.isEqualTo(ExchangeTypes.FANOUT);
assertThat(rabbitConsumerProperties.getMaxConcurrency()).isEqualTo(4);
}
@@ -315,8 +337,8 @@ public class RabbitBinderModuleTests {
@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer> containerCustomizer() {
return (c, q, g) -> c.setBeanName("setByCustomizerForQueue:" + q +
(g == null ? "" : ",andGroup:" + g));
return (c, q, g) -> c.setBeanName(
"setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g));
}
}
@@ -336,9 +358,8 @@ public class RabbitBinderModuleTests {
public Cloud cloud() {
Cloud cloud = mock(Cloud.class);
willReturn(new CachingConnectionFactory("localhost"))
.given(cloud)
.getSingletonServiceConnector(ConnectionFactory.class, null);
willReturn(new CachingConnectionFactory("localhost")).given(cloud)
.getSingletonServiceConnector(ConnectionFactory.class, null);
return cloud;
}