Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove deprecations; fix corresponding schema

Browse Source 
- add `batch-size`
- deprecate `transaction-size`
- remove deprecated `publisher-confirms`
- add 'consumer-batch-enabled'
This commit is contained in:
Gary Russell
2020-03-10 13:44:47 -04:00
committed by Artem Bilan
parent 3fccb5a176
commit ddccc08bdc
18 changed files with 88 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/RepeatProcessor.java
View File

@@ -1,201 +0,0 @@
/*
* Copyright 2002-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.junit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.After;
import org.junit.Before;
import org.junit.internal.runners.statements.RunAfters;
import org.junit.internal.runners.statements.RunBefores;
import org.junit.rules.MethodRule;
import org.junit.runners.model.FrameworkMethod;
import org.junit.runners.model.Statement;
import org.junit.runners.model.TestClass;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.test.annotation.Repeat;
/**
* A JUnit method @Rule that looks at Spring repeat annotations on methods and executes the test multiple times
* (without re-initializing the test case if necessary). To avoid re-initializing use the {@link #isInitialized()}
* method to protect the @Before and @After methods.
* @deprecated in favor of JUnit 5 {@link org.junit.jupiter.api.RepeatedTest}.
*
* @author Dave Syer
*
*/
@Deprecated
public class RepeatProcessor implements MethodRule {
private static final Log LOGGER = LogFactory.getLog(RepeatProcessor.class);
private final int concurrency;
private volatile boolean initialized = false;
private volatile boolean finalizing = false;
public RepeatProcessor() {
this(0);
}
public RepeatProcessor(int concurrency) {
this.concurrency = concurrency < 0 ? 0 : concurrency;
}
@Override
public Statement apply(final Statement base, FrameworkMethod method, final Object target) {
Repeat repeat = AnnotationUtils.findAnnotation(method.getMethod(), Repeat.class);
if (repeat == null) {
return base;
}
final int repeats = repeat.value();
if (repeats <= 1) {
return base;
}
initializeIfNecessary(target);
if (this.concurrency <= 0) {
return new Statement() {
@Override
public void evaluate() throws Throwable {
try {
for (int i = 0; i < repeats; i++) {
try {
base.evaluate();
}
catch (Throwable t) { // NOSONAR
throw new IllegalStateException(
"Failed on iteration: " + i + " of " + repeats + " (started at 0)", t);
}
}
}
finally {
finalizeIfNecessary(target);
}
}
};
}
return new Statement() { // NOSONAR
@Override
public void evaluate() throws Throwable {
List<Future<Boolean>> results = new ArrayList<Future<Boolean>>();
ExecutorService executor = Executors.newFixedThreadPool(RepeatProcessor.this.concurrency);
CompletionService<Boolean> completionService = new ExecutorCompletionService<Boolean>(executor);
try {
for (int i = 0; i < repeats; i++) {
final int count = i;
results.add(completionService.submit(new Callable<Boolean>() {
@Override
public Boolean call() {
try {
base.evaluate();
}
catch (Throwable t) { // NOSONAR
throw new IllegalStateException("Failed on iteration: " + count, t);
}
return true;
}
}));
}
for (int i = 0; i < repeats; i++) {
Future<Boolean> future = completionService.take();
assertThat(future.get()).as("Null result from completer").isTrue();
}
}
finally {
executor.shutdownNow();
finalizeIfNecessary(target);
}
}
};
}
private void finalizeIfNecessary(Object target) {
this.finalizing = true;
List<FrameworkMethod> afters = new TestClass(target.getClass()).getAnnotatedMethods(After.class);
try {
if (!afters.isEmpty()) {
LOGGER.debug("Running @After methods");
try {
new RunAfters(new Statement() {
@Override
public void evaluate() {
}
}, afters, target).evaluate();
}
catch (Throwable e) { // NOSONAR
fail("Unexpected throwable " + e);
}
}
}
finally {
this.finalizing = false;
}
}
private void initializeIfNecessary(Object target) {
TestClass testClass = new TestClass(target.getClass());
List<FrameworkMethod> befores = testClass.getAnnotatedMethods(Before.class);
if (!befores.isEmpty()) {
LOGGER.debug("Running @Before methods");
try {
new RunBefores(new Statement() {
@Override
public void evaluate() {
}
}, befores, target).evaluate();
}
catch (Throwable e) { // NOSONAR
fail("Unexpected throwable " + e);
}
this.initialized = true;
}
if (!testClass.getAnnotatedMethods(After.class).isEmpty()) {
this.initialized = true;
}
}
public boolean isInitialized() {
return this.initialized;
}
public boolean isFinalizing() {
return this.finalizing;
}
public int getConcurrency() {
return this.concurrency > 0 ? this.concurrency : 1;
}
}

13
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java
View File

@@ -20,7 +20,6 @@ package org.springframework.amqp.rabbit.config;
import java.util.Arrays;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
@@ -346,18 +345,6 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
this.recoveryCallback = recoveryCallback;
}
/**
* A {@link Consumer} that is invoked to enable setting other container properties not
* exposed by this container factory.
* @param configurer the configurer;
* @since 2.1.1
* @deprecated in favor of {@link #setContainerCustomizer(ContainerCustomizer)}.
*/
@Deprecated
public void setContainerConfigurer(Consumer<C> configurer) {
this.containerCustomizer = container -> configurer.accept(container);
}
/**
* Set a {@link ContainerCustomizer} that is invoked after a container is created and
* configured to enable further customization of the container.

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ConnectionFactoryParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -50,7 +50,7 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
private static final String EXECUTOR_ATTRIBUTE = "executor";
private static final String PUBLISHER_CONFIRMS = "publisher-confirms";
private static final String CONFIRM_TYPE = "confirm-type";
private static final String PUBLISHER_RETURNS = "publisher-returns";
@@ -101,7 +101,6 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
NamespaceUtils.setReferenceIfAttributeDefined(builder, element, EXECUTOR_ATTRIBUTE);
NamespaceUtils.setValueIfAttributeDefined(builder, element, ADDRESSES);
NamespaceUtils.setValueIfAttributeDefined(builder, element, SHUFFLE_ADDRESSES);
NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_CONFIRMS);
NamespaceUtils.setValueIfAttributeDefined(builder, element, PUBLISHER_RETURNS);
NamespaceUtils.setValueIfAttributeDefined(builder, element, REQUESTED_HEARTBEAT, "requestedHeartBeat");
NamespaceUtils.setValueIfAttributeDefined(builder, element, CONNECTION_TIMEOUT);
@@ -111,7 +110,7 @@ class ConnectionFactoryParser extends AbstractSingleBeanDefinitionParser {
NamespaceUtils.setValueIfAttributeDefined(builder, element, FACTORY_TIMEOUT, "channelCheckoutTimeout");
NamespaceUtils.setValueIfAttributeDefined(builder, element, CONNECTION_LIMIT);
NamespaceUtils.setReferenceIfAttributeDefined(builder, element, "connection-name-strategy");
NamespaceUtils.setValueIfAttributeDefined(builder, element, "confirm-type", "publisherConfirmType");
NamespaceUtils.setValueIfAttributeDefined(builder, element, CONFIRM_TYPE, "publisherConfirmType");
}
}

13
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ListenerContainerFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -396,16 +396,6 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
this.batchSize = batchSize;
}
/**
* Set the txSize.
* @param txSize the txSize.
* @deprecated in favor of {@link #setBatchSize(int)}.
*/
@Deprecated
public void setTxSize(int txSize) {
setBatchSize(txSize);
}
/**
* Set to true to present a list of messages based on the {@link #setBatchSize(int)},
* if the container and listener support it.
@@ -432,7 +422,6 @@ public class ListenerContainerFactoryBean extends AbstractFactoryBean<AbstractMe
: this.listenerContainer.getClass();
}
@SuppressWarnings("deprecation")
@Override
protected AbstractMessageListenerContainer createInstance() { // NOSONAR complexity
if (this.listenerContainer == null) {

22
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RabbitNamespaceUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -74,6 +74,10 @@ public final class RabbitNamespaceUtils {
private static final String TRANSACTION_SIZE_ATTRIBUTE = "transaction-size";
private static final String CONSUMER_BATCH_ENABLED_ATTRIBUTE = "consumer-batch-enabled";
private static final String BATCH_SIZE_ATTRIBUTE = "batch-size";
private static final String PHASE_ATTRIBUTE = "phase";
private static final String AUTO_STARTUP_ATTRIBUTE = "auto-startup";
@@ -216,9 +220,23 @@ public final class RabbitNamespaceUtils {
containerDef.getPropertyValues().add("channelTransacted", new TypedStringValue(channelTransacted));
}
String consumerBatch = containerEle.getAttribute(CONSUMER_BATCH_ENABLED_ATTRIBUTE);
if (StringUtils.hasText(consumerBatch)) {
containerDef.getPropertyValues().add("consumerBatchEnabled", new TypedStringValue(consumerBatch));
}
String batchSize = containerEle.getAttribute(BATCH_SIZE_ATTRIBUTE);
if (StringUtils.hasText(batchSize)) {
containerDef.getPropertyValues().add("batchSize", new TypedStringValue(batchSize));
}
String transactionSize = containerEle.getAttribute(TRANSACTION_SIZE_ATTRIBUTE);
if (StringUtils.hasText(transactionSize)) {
containerDef.getPropertyValues().add("txSize", new TypedStringValue(transactionSize));
if (StringUtils.hasText(batchSize)) {
parserContext.getReaderContext().error(
"Listener Container - cannot have both 'batch-size' and 'transaction-size'", containerEle);
}
containerDef.getPropertyValues().add("batchSize", new TypedStringValue(transactionSize));
}
String requeueRejected = containerEle.getAttribute(REQUEUE_REJECTED_ATTRIBUTE);

12
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -56,16 +56,6 @@ public class SimpleRabbitListenerContainerFactory
private Boolean consumerBatchEnabled;
/**
* @param txSize the transaction size.
* @see SimpleMessageListenerContainer#setBatchSize
* @deprecated in favor of {@link #setBatchSize(Integer)}
*/
@Deprecated
public void setTxSize(Integer txSize) {
setBatchSize(txSize);
}
/**
* @param batchSize the batch size.
* @since 2.2

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -114,7 +114,6 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint
return this.messageHandlerMethodFactory;
}
@SuppressWarnings("deprecation")
@Override
protected MessagingMessageListenerAdapter createMessageListener(MessageListenerContainer container) {
Assert.state(this.messageHandlerMethodFactory != null,

14
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -353,20 +353,6 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
this.batchSize = batchSize;
}
/**
* Tells the container how many messages to process in a single transaction (if the
* channel is transactional). For best results it should be less than or equal to
* {@link #setPrefetchCount(int) the prefetch count}. Also affects how often acks are
* sent when using {@link org.springframework.amqp.core.AcknowledgeMode#AUTO} - one
* ack per txSize. Default is 1.
* @param txSize the transaction size
* @deprecated since 2.2 in favor of {@link #setBatchSize(int)}.
*/
@Deprecated
public void setTxSize(int txSize) {
setBatchSize(txSize);
}
/**
* Set to true to present a list of messages based on the {@link #setBatchSize(int)},
* if the listener supports it.

19
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 the original author or authors.
* Copyright 2015-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -85,23 +85,6 @@ public class HandlerAdapter {
}
}
/**
* Return the return type for the method that will be chosen for this payload.
* @param payload the payload.
* @return the return type, or null if no handler found.
* @since 2.0
* @deprecated in favor of {@link #getReturnTypeFor(Object)}.
*/
@Deprecated
public Object getReturnType(Object payload) {
if (this.invokerHandlerMethod != null) {
return this.invokerHandlerMethod.getMethod().getReturnType();
}
else {
return this.delegatingHandler.getMethodFor(payload).getReturnType();
}
}
/**
* Return the return type for the method that will be chosen for this payload.
* @param payload the payload.

11
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/InvocationResult.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2019 the original author or authors.
* Copyright 2018-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -45,16 +45,13 @@ public final class InvocationResult {
private final Method method;
/**
* @deprecated in favor of {@link #InvocationResult(Object, Expression, Type, Object, Method)}.
* Construct an instance with the provided properties.
* @param result the result.
* @param sendTo the sendTo expression.
* @param returnType the return type.
* @param bean the bean.
* @param method the method.
*/
@Deprecated
public InvocationResult(Object result, @Nullable Expression sendTo, @Nullable Type returnType) {
this(result, sendTo, returnType, null, null);
}
public InvocationResult(Object result, @Nullable Expression sendTo, @Nullable Type returnType,
@Nullable Object bean, @Nullable Method method) {

22
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessageListenerAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -338,31 +338,13 @@ public class MessageListenerAdapter extends AbstractAdaptableMessageListener {
* This can be overridden to treat special message content such as arrays differently, for example passing in each
* element of the message array as distinct method argument.
* @param extractedMessage the content of the message
* @return the array of arguments to be passed into the listener method (each element of the array corresponding to
* a distinct method argument)
* @deprecated use @{@link #buildListenerArguments(Object, Channel, Message)} to get complete arguments
*/
@Deprecated
protected Object[] buildListenerArguments(Object extractedMessage) {
return new Object[] { extractedMessage };
}
/**
* Build an array of arguments to be passed into the target listener method. Allows for multiple method arguments to
* be built from message object with channel, More detail about {@code extractedMessage} in the method
* {@link #buildListenerArguments(java.lang.Object)}.
* This can be overridden to treat special message content such as arrays differently, and add argument in case of
* receiving Channel and original Message object to invoke basicAck method in the listener by manual acknowledge
* mode.
* @param extractedMessage the content of the message
* @param channel the Rabbit channel to operate on
* @param message the incoming Rabbit message
* @return the array of arguments to be passed into the listener method (each element of the array corresponding to
* a distinct method argument)
*/
@SuppressWarnings("deprecation")
protected Object[] buildListenerArguments(Object extractedMessage, Channel channel, Message message) {
return buildListenerArguments(extractedMessage);
return new Object[] { extractedMessage };
}
/**

119
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java
View File

@@ -43,13 +43,12 @@ import org.apache.logging.log4j.core.appender.AbstractAppender;
import org.apache.logging.log4j.core.appender.AbstractManager;
import org.apache.logging.log4j.core.async.BlockingQueueFactory;
import org.apache.logging.log4j.core.config.Configuration;
import org.apache.logging.log4j.core.config.Property;
import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginBuilderAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginBuilderFactory;
import org.apache.logging.log4j.core.config.plugins.PluginConfiguration;
import org.apache.logging.log4j.core.config.plugins.PluginElement;
import org.apache.logging.log4j.core.config.plugins.PluginFactory;
import org.apache.logging.log4j.core.layout.PatternLayout;
import org.apache.logging.log4j.core.util.Integers;
@@ -138,109 +137,28 @@ public class AmqpAppender extends AbstractAppender {
*/
private final Object layoutMutex = new Object();
@SuppressWarnings("deprecation") // For backward compatibility
/**
* Construct an instance with the provided properties.
* @param name the name.
* @param filter the filter.
* @param layout the layout.
* @param ignoreExceptions true to ignore exceptions.
* @param manager the manager.
* @param eventQueue the event queue.
* @param properties the properties.
*/
public AmqpAppender(String name, Filter filter, Layout<? extends Serializable> layout, boolean ignoreExceptions,
AmqpManager manager, BlockingQueue<Event> eventQueue) {
Property[] properties, AmqpManager manager, BlockingQueue<Event> eventQueue) {
super(name, filter, layout, ignoreExceptions);
super(name, filter, layout, ignoreExceptions, properties);
this.manager = manager;
this.events = eventQueue;
}
@Deprecated // For backward compatibility
@PluginFactory
public static AmqpAppender createAppender(// NOSONAR NCSS line count
@PluginConfiguration final Configuration configuration,
@PluginAttribute("name") String name,
@PluginElement("Layout") Layout<? extends Serializable> layout,
@PluginElement("Filter") Filter filter,
@PluginAttribute("ignoreExceptions") boolean ignoreExceptions,
@PluginAttribute("uri") URI uri,
@PluginAttribute("host") String host,
@PluginAttribute("port") String port,
@PluginAttribute("addresses") String addresses,
@PluginAttribute("user") String user,
@PluginAttribute("password") String password,
@PluginAttribute("virtualHost") String virtualHost,
@PluginAttribute("useSsl") boolean useSsl,
@PluginAttribute("verifyHostname") boolean verifyHostname,
@PluginAttribute("sslAlgorithm") String sslAlgorithm,
@PluginAttribute("sslPropertiesLocation") String sslPropertiesLocation,
@PluginAttribute("keyStore") String keyStore,
@PluginAttribute("keyStorePassphrase") String keyStorePassphrase,
@PluginAttribute("keyStoreType") String keyStoreType,
@PluginAttribute("trustStore") String trustStore,
@PluginAttribute("trustStorePassphrase") String trustStorePassphrase,
@PluginAttribute("trustStoreType") String trustStoreType,
@PluginAttribute("saslConfig") String saslConfig,
@PluginAttribute("senderPoolSize") int senderPoolSize,
@PluginAttribute("maxSenderRetries") int maxSenderRetries,
@PluginAttribute("applicationId") String applicationId,
@PluginAttribute("routingKeyPattern") String routingKeyPattern,
@PluginAttribute("generateId") boolean generateId,
@PluginAttribute("deliveryMode") String deliveryMode,
@PluginAttribute("exchange") String exchange,
@PluginAttribute("exchangeType") String exchangeType,
@PluginAttribute("declareExchange") boolean declareExchange,
@PluginAttribute("durable") boolean durable,
@PluginAttribute("autoDelete") boolean autoDelete,
@PluginAttribute("contentType") String contentType,
@PluginAttribute("contentEncoding") String contentEncoding,
@PluginAttribute("connectionName") String connectionName,
@PluginAttribute("clientConnectionProperties") String clientConnectionProperties,
@PluginAttribute("async") boolean async,
@PluginAttribute("charset") String charset,
@PluginAttribute(value = "bufferSize", defaultInt = Integer.MAX_VALUE) int bufferSize,
@PluginElement(BlockingQueueFactory.ELEMENT_TYPE) BlockingQueueFactory<Event> blockingQueueFactory,
@PluginAttribute(value = "addMdcAsHeaders", defaultBoolean = true) boolean addMdcAsHeaders) {
return new Builder()
.setConfiguration(configuration)
.setName(name)
.setLayout(layout)
.setFilter(filter)
.setIgnoreExceptions(ignoreExceptions)
.setUri(uri)
.setHost(host)
.setPort(port)
.setAddresses(addresses)
.setUser(user)
.setPassword(password)
.setVirtualHost(virtualHost)
.setUseSsl(useSsl)
.setVerifyHostname(verifyHostname)
.setSslAlgorithm(sslAlgorithm)
.setSslPropertiesLocation(sslPropertiesLocation)
.setKeyStore(keyStore)
.setKeyStorePassphrase(keyStorePassphrase)
.setKeyStoreType(keyStoreType)
.setTrustStore(trustStore)
.setTrustStorePassphrase(trustStorePassphrase)
.setTrustStoreType(trustStoreType)
.setSaslConfig(saslConfig)
.setSenderPoolSize(senderPoolSize)
.setMaxSenderRetries(maxSenderRetries)
.setApplicationId(applicationId)
.setRoutingKeyPattern(routingKeyPattern)
.setGenerateId(generateId)
.setDeliveryMode(deliveryMode)
.setExchange(exchange)
.setExchangeType(exchangeType)
.setDeclareExchange(declareExchange)
.setDurable(durable)
.setAutoDelete(autoDelete)
.setContentType(contentType)
.setContentEncoding(contentEncoding)
.setConnectionName(connectionName)
.setClientConnectionProperties(clientConnectionProperties)
.setAsync(async)
.setCharset(charset)
.setBufferSize(bufferSize)
.setBlockingQueueFactory(blockingQueueFactory)
.setAddMdcAsHeaders(addMdcAsHeaders)
.build();
}
/**
* Create a new builder.
* @return the builder.
*/
@PluginBuilderFactory
public static Builder newBuilder() {
return new Builder();
@@ -1203,7 +1121,8 @@ public class AmqpAppender extends AbstractAppender {
*/
protected AmqpAppender buildInstance(String name, Filter filter, Layout<? extends Serializable> layout,
boolean ignoreExceptions, AmqpManager manager, BlockingQueue<Event> eventQueue) {
return new AmqpAppender(name, filter, layout, ignoreExceptions, manager, eventQueue);
return new AmqpAppender(name, filter, layout, ignoreExceptions, Property.EMPTY_ARRAY, manager, eventQueue);
}
}

40
spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-2.2.xsd
View File

@@ -736,9 +736,28 @@
<xsd:attribute name="transaction-size" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Deprecated. Synonym to 'batch-size'.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="consumer-batch-enabled" default="false">
<xsd:annotation>
<xsd:documentation><![CDATA[
Enable consumer-side batching according to 'batch-size' and 'receive-timeout'.
Only applies when the container type is 'simple'.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="batch-size" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Tells the container how many messages to process in a single transaction (if the channel is transactional). For
best results it should be less than or equal to the prefetch count. Also used to determine how often acks
are sent when using AUTO acknowledge mode.
best results it should be less than or equal to the prefetch count.
Also used to determine how often acks are sent when using AUTO acknowledge mode.
Also used to determine the batch size when 'consumer-batch-enabled' is 'true'.
Only applies when the container type is 'simple'.
]]></xsd:documentation>
</xsd:annotation>
@@ -746,10 +765,11 @@
<xsd:attribute name="receive-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
The time for which a consumer waits for a message; used together with 'transaction-size' to determine whether
or not a consumer is idle, why dynamic concurrency is enabled using 'max-concurrency'. When 'transaction-size'
is greater than 1 (default), acks for processed message(s) can be delayed up to ('transaction-size' - 1) *
this value because the ack is sent after `transaction-size` attempts to receive a message have occurred.
The time for which a consumer waits for a message; used together with 'batch-size' to determine whether
or not a consumer is idle, why dynamic concurrency is enabled using 'max-concurrency'. When 'batch-size'
is greater than 1 (default), acks for processed message(s) can be delayed up to ('batch-size' - 1) *
this value because the ack is sent after `batch-size` attempts to receive a message have occurred.
Also used to deliver a short batch when 'consumer-batch-enabled' is 'true'.
Only applies when the container type is 'simple'.
]]></xsd:documentation>
</xsd:annotation>
@@ -1464,14 +1484,6 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="publisher-confirms" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
When true, channels on connections created by this factory support correlated publisher confirms.
DEPRECATED in favor of 'confirm-type'.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="confirm-type" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[

3
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/ListenerContainerParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2010-2019 the original author or authors.
* Copyright 2010-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -180,6 +180,7 @@ public class ListenerContainerParserTests {
SimpleMessageListenerContainer container = beanFactory.getBean("container6", SimpleMessageListenerContainer.class);
assertThat(container.isChannelTransacted()).isTrue();
assertThat(ReflectionTestUtils.getField(container, "batchSize")).isEqualTo(5);
assertThat(ReflectionTestUtils.getField(container, "consumerBatchEnabled")).isEqualTo(Boolean.TRUE);
}
@Test

32
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -1846,34 +1846,4 @@ public class CachingConnectionFactoryTests extends AbstractConnectionFactoryTest
assertThat(firstAddress).containsExactly("host1", "host2", "host3");
}
@SuppressWarnings("deprecation")
@Test
public void confirmsSimple() {
CachingConnectionFactory cf = new CachingConnectionFactory(mock(ConnectionFactory.class));
cf.setSimplePublisherConfirms(false);
assertThat(cf.isSimplePublisherConfirms()).isFalse();
assertThat(cf.getPublisherConnectionFactory().isSimplePublisherConfirms()).isFalse();
cf.setSimplePublisherConfirms(true);
assertThat(cf.isSimplePublisherConfirms()).isTrue();
assertThat(cf.getPublisherConnectionFactory().isSimplePublisherConfirms()).isTrue();
cf.setSimplePublisherConfirms(false);
assertThat(cf.isSimplePublisherConfirms()).isFalse();
assertThat(cf.getPublisherConnectionFactory().isSimplePublisherConfirms()).isFalse();
}
@SuppressWarnings("deprecation")
@Test
public void confirmsCorrelated() {
CachingConnectionFactory cf = new CachingConnectionFactory(mock(ConnectionFactory.class));
cf.setPublisherConfirms(false);
assertThat(cf.getPublisherConnectionFactory().isPublisherConfirms()).isFalse();
assertThat(cf.isPublisherConfirms()).isFalse();
cf.setPublisherConfirms(true);
assertThat(cf.getPublisherConnectionFactory().isPublisherConfirms()).isTrue();
assertThat(cf.isPublisherConfirms()).isTrue();
cf.setPublisherConfirms(false);
assertThat(cf.isPublisherConfirms()).isFalse();
assertThat(cf.getPublisherConnectionFactory().isPublisherConfirms()).isFalse();
}
}

7
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/ExtendAmqpAppender.java
View File

@@ -21,6 +21,7 @@ import java.util.concurrent.BlockingQueue;
import org.apache.logging.log4j.core.Filter;
import org.apache.logging.log4j.core.Layout;
import org.apache.logging.log4j.core.config.Property;
import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.config.plugins.PluginBuilderAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginBuilderFactory;
@@ -36,12 +37,12 @@ import org.springframework.amqp.core.Message;
@Plugin(name = "TestRabbitMQ", category = "Core", elementType = "appender", printObject = true)
public class ExtendAmqpAppender extends AmqpAppender {
private String foo;
private String bar;
private final String foo;
private final String bar;
public ExtendAmqpAppender(String name, Filter filter, Layout<? extends Serializable> layout,
boolean ignoreExceptions, AmqpManager manager, BlockingQueue<Event> eventQueue, String foo, String bar) {
super(name, filter, layout, ignoreExceptions, manager, eventQueue);
super(name, filter, layout, ignoreExceptions, Property.EMPTY_ARRAY, manager, eventQueue);
this.foo = foo;
this.bar = bar;
}

2
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ConnectionFactoryParserTests-context.xml
View File

@@ -9,7 +9,7 @@
<rabbit:connection-factory id="kitchenSink" host="foo" virtual-host="/bar"
channel-cache-size="10" port="6888" username="user" password="password"
publisher-confirms="true" publisher-returns="true" connection-timeout="789"
confirm-type="CORRELATED" publisher-returns="true" connection-timeout="789"
factory-timeout="234" connection-limit="456"
requested-heartbeat="123"
connection-name-strategy="connectionNameStrategy"/>

3
spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/ListenerContainerParserTests-context.xml
View File

@@ -40,7 +40,8 @@
<rabbit:listener id="container5" queues="foo" ref="testBean" method="handle"/>
</rabbit:listener-container>
<rabbit:listener-container connection-factory="connectionFactory" channel-transacted="true" transaction-size="5" >
<rabbit:listener-container connection-factory="connectionFactory" channel-transacted="true" batch-size="5"
consumer-batch-enabled="true" >
<rabbit:listener id="container6" queues="foo" ref="testBean" method="handle"/>
</rabbit:listener-container>