Fix deprecation warnings in AMQP module
This commit is contained in:
Artem Bilan
@@ -70,73 +70,73 @@ import org.springframework.util.ErrorHandler;
|
||||
public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChannel>
|
||||
implements SmartLifecycle, BeanNameAware {
|
||||
|
||||
private volatile AbstractAmqpChannel channel;
|
||||
|
||||
private volatile List<ChannelInterceptor> interceptors;
|
||||
private final AmqpTemplate amqpTemplate = new RabbitTemplate();
|
||||
|
||||
private final boolean messageDriven;
|
||||
|
||||
private final AmqpTemplate amqpTemplate = new RabbitTemplate();
|
||||
private AbstractAmqpChannel channel;
|
||||
|
||||
private volatile AmqpAdmin amqpAdmin;
|
||||
private List<ChannelInterceptor> interceptors;
|
||||
|
||||
private volatile FanoutExchange exchange;
|
||||
private AmqpAdmin amqpAdmin;
|
||||
|
||||
private volatile String queueName;
|
||||
private FanoutExchange exchange;
|
||||
|
||||
private volatile boolean autoStartup = true;
|
||||
private String queueName;
|
||||
|
||||
private volatile Advice[] adviceChain;
|
||||
private boolean autoStartup = true;
|
||||
|
||||
private volatile Integer concurrentConsumers;
|
||||
private Advice[] adviceChain;
|
||||
|
||||
private volatile Integer consumersPerQueue;
|
||||
private Integer concurrentConsumers;
|
||||
|
||||
private volatile ConnectionFactory connectionFactory;
|
||||
private Integer consumersPerQueue;
|
||||
|
||||
private volatile MessagePropertiesConverter messagePropertiesConverter;
|
||||
private ConnectionFactory connectionFactory;
|
||||
|
||||
private volatile ErrorHandler errorHandler;
|
||||
private MessagePropertiesConverter messagePropertiesConverter;
|
||||
|
||||
private volatile Boolean exposeListenerChannel;
|
||||
private ErrorHandler errorHandler;
|
||||
|
||||
private volatile Integer phase;
|
||||
private Boolean exposeListenerChannel;
|
||||
|
||||
private volatile Integer prefetchCount;
|
||||
private Integer phase;
|
||||
|
||||
private volatile boolean isPubSub;
|
||||
private Integer prefetchCount;
|
||||
|
||||
private volatile Long receiveTimeout;
|
||||
private boolean isPubSub;
|
||||
|
||||
private volatile Long recoveryInterval;
|
||||
private Long receiveTimeout;
|
||||
|
||||
private volatile Long shutdownTimeout;
|
||||
private Long recoveryInterval;
|
||||
|
||||
private volatile String beanName;
|
||||
private Long shutdownTimeout;
|
||||
|
||||
private volatile AcknowledgeMode acknowledgeMode;
|
||||
private String beanName;
|
||||
|
||||
private volatile boolean channelTransacted;
|
||||
private AcknowledgeMode acknowledgeMode;
|
||||
|
||||
private volatile Executor taskExecutor;
|
||||
private boolean channelTransacted;
|
||||
|
||||
private volatile PlatformTransactionManager transactionManager;
|
||||
private Executor taskExecutor;
|
||||
|
||||
private volatile TransactionAttribute transactionAttribute;
|
||||
private PlatformTransactionManager transactionManager;
|
||||
|
||||
private volatile Integer txSize;
|
||||
private TransactionAttribute transactionAttribute;
|
||||
|
||||
private volatile Integer maxSubscribers;
|
||||
private Integer batchSize;
|
||||
|
||||
private volatile Boolean missingQueuesFatal;
|
||||
private Integer maxSubscribers;
|
||||
|
||||
private volatile MessageDeliveryMode defaultDeliveryMode;
|
||||
private Boolean missingQueuesFatal;
|
||||
|
||||
private volatile Boolean extractPayload;
|
||||
private MessageDeliveryMode defaultDeliveryMode;
|
||||
|
||||
private volatile AmqpHeaderMapper outboundHeaderMapper = DefaultAmqpHeaderMapper.outboundMapper();
|
||||
private Boolean extractPayload;
|
||||
|
||||
private volatile AmqpHeaderMapper inboundHeaderMapper = DefaultAmqpHeaderMapper.inboundMapper();
|
||||
private AmqpHeaderMapper outboundHeaderMapper = DefaultAmqpHeaderMapper.outboundMapper();
|
||||
|
||||
private AmqpHeaderMapper inboundHeaderMapper = DefaultAmqpHeaderMapper.inboundMapper();
|
||||
|
||||
private boolean headersLast;
|
||||
|
||||
@@ -314,8 +314,18 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify a batch size for consumer.
|
||||
* @param txSize the batch size to use
|
||||
* @deprecated since 5.2 in favor of {@link #setBatchSize(Integer)}
|
||||
*/
|
||||
@Deprecated
|
||||
public void setTxSize(int txSize) {
|
||||
this.txSize = txSize;
|
||||
setBatchSize(txSize);
|
||||
}
|
||||
|
||||
public void setBatchSize(Integer batchSize) {
|
||||
this.batchSize = batchSize;
|
||||
}
|
||||
|
||||
public void setMaxSubscribers(int maxSubscribers) {
|
||||
@@ -360,18 +370,20 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
|
||||
}
|
||||
if (this.isPubSub) {
|
||||
PublishSubscribeAmqpChannel pubsub = new PublishSubscribeAmqpChannel(
|
||||
this.beanName, container, this.amqpTemplate, this.outboundHeaderMapper, this.inboundHeaderMapper);
|
||||
this.beanName, container, this.amqpTemplate, this.outboundHeaderMapper,
|
||||
this.inboundHeaderMapper);
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotNull(this.exchange, pubsub::setExchange)
|
||||
.acceptIfNotNull(this.maxSubscribers, pubsub::setMaxSubscribers);
|
||||
.acceptIfNotNull(this.exchange, pubsub::setExchange)
|
||||
.acceptIfNotNull(this.maxSubscribers, pubsub::setMaxSubscribers);
|
||||
this.channel = pubsub;
|
||||
}
|
||||
else {
|
||||
PointToPointSubscribableAmqpChannel p2p = new PointToPointSubscribableAmqpChannel(
|
||||
this.beanName, container, this.amqpTemplate, this.outboundHeaderMapper, this.inboundHeaderMapper);
|
||||
this.beanName, container, this.amqpTemplate, this.outboundHeaderMapper,
|
||||
this.inboundHeaderMapper);
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfHasText(this.queueName, p2p::setQueueName)
|
||||
.acceptIfNotNull(this.maxSubscribers, p2p::setMaxSubscribers);
|
||||
.acceptIfHasText(this.queueName, p2p::setQueueName)
|
||||
.acceptIfNotNull(this.maxSubscribers, p2p::setMaxSubscribers);
|
||||
this.channel = p2p;
|
||||
}
|
||||
}
|
||||
@@ -380,17 +392,17 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
|
||||
PollableAmqpChannel pollable = new PollableAmqpChannel(this.beanName, this.amqpTemplate,
|
||||
this.outboundHeaderMapper, this.inboundHeaderMapper);
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotNull(this.amqpAdmin, pollable::setAmqpAdmin)
|
||||
.acceptIfHasText(this.queueName, pollable::setQueueName);
|
||||
.acceptIfNotNull(this.amqpAdmin, pollable::setAmqpAdmin)
|
||||
.acceptIfHasText(this.queueName, pollable::setQueueName);
|
||||
this.channel = pollable;
|
||||
}
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotEmpty(this.interceptors, this.channel::setInterceptors);
|
||||
.acceptIfNotEmpty(this.interceptors, this.channel::setInterceptors);
|
||||
this.channel.setBeanName(this.beanName);
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotNull(getBeanFactory(), this.channel::setBeanFactory)
|
||||
.acceptIfNotNull(this.defaultDeliveryMode, this.channel::setDefaultDeliveryMode)
|
||||
.acceptIfNotNull(this.extractPayload, this.channel::setExtractPayload);
|
||||
.acceptIfNotNull(getBeanFactory(), this.channel::setBeanFactory)
|
||||
.acceptIfNotNull(this.defaultDeliveryMode, this.channel::setDefaultDeliveryMode)
|
||||
.acceptIfNotNull(this.extractPayload, this.channel::setExtractPayload);
|
||||
this.channel.setHeadersMappedLast(this.headersLast);
|
||||
this.channel.afterPropertiesSet();
|
||||
return this.channel;
|
||||
@@ -401,9 +413,9 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
|
||||
if (this.consumersPerQueue == null) {
|
||||
SimpleMessageListenerContainer smlc = new SimpleMessageListenerContainer();
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotNull(this.concurrentConsumers, smlc::setConcurrentConsumers)
|
||||
.acceptIfNotNull(this.receiveTimeout, smlc::setReceiveTimeout)
|
||||
.acceptIfNotNull(this.txSize, smlc::setTxSize);
|
||||
.acceptIfNotNull(this.concurrentConsumers, smlc::setConcurrentConsumers)
|
||||
.acceptIfNotNull(this.receiveTimeout, smlc::setReceiveTimeout)
|
||||
.acceptIfNotNull(this.batchSize, smlc::setBatchSize);
|
||||
container = smlc;
|
||||
}
|
||||
else {
|
||||
@@ -412,24 +424,24 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
|
||||
container = dmlc;
|
||||
}
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotNull(this.acknowledgeMode, container::setAcknowledgeMode)
|
||||
.acceptIfNotEmpty(this.adviceChain, container::setAdviceChain);
|
||||
.acceptIfNotNull(this.acknowledgeMode, container::setAcknowledgeMode)
|
||||
.acceptIfNotEmpty(this.adviceChain, container::setAdviceChain);
|
||||
container.setAutoStartup(this.autoStartup);
|
||||
container.setChannelTransacted(this.channelTransacted);
|
||||
container.setConnectionFactory(this.connectionFactory);
|
||||
|
||||
JavaUtils.INSTANCE
|
||||
.acceptIfNotNull(this.errorHandler, container::setErrorHandler)
|
||||
.acceptIfNotNull(this.exposeListenerChannel, container::setExposeListenerChannel)
|
||||
.acceptIfNotNull(this.messagePropertiesConverter, container::setMessagePropertiesConverter)
|
||||
.acceptIfNotNull(this.phase, container::setPhase)
|
||||
.acceptIfNotNull(this.prefetchCount, container::setPrefetchCount)
|
||||
.acceptIfNotNull(this.recoveryInterval, container::setRecoveryInterval)
|
||||
.acceptIfNotNull(this.shutdownTimeout, container::setShutdownTimeout)
|
||||
.acceptIfNotNull(this.taskExecutor, container::setTaskExecutor)
|
||||
.acceptIfNotNull(this.transactionAttribute, container::setTransactionAttribute)
|
||||
.acceptIfNotNull(this.transactionManager, container::setTransactionManager)
|
||||
.acceptIfNotNull(this.missingQueuesFatal, container::setMissingQueuesFatal);
|
||||
.acceptIfNotNull(this.errorHandler, container::setErrorHandler)
|
||||
.acceptIfNotNull(this.exposeListenerChannel, container::setExposeListenerChannel)
|
||||
.acceptIfNotNull(this.messagePropertiesConverter, container::setMessagePropertiesConverter)
|
||||
.acceptIfNotNull(this.phase, container::setPhase)
|
||||
.acceptIfNotNull(this.prefetchCount, container::setPrefetchCount)
|
||||
.acceptIfNotNull(this.recoveryInterval, container::setRecoveryInterval)
|
||||
.acceptIfNotNull(this.shutdownTimeout, container::setShutdownTimeout)
|
||||
.acceptIfNotNull(this.taskExecutor, container::setTaskExecutor)
|
||||
.acceptIfNotNull(this.transactionAttribute, container::setTransactionAttribute)
|
||||
.acceptIfNotNull(this.transactionManager, container::setTransactionManager)
|
||||
.acceptIfNotNull(this.missingQueuesFatal, container::setMissingQueuesFatal);
|
||||
return container;
|
||||
}
|
||||
|
||||
|
||||
@@ -39,11 +39,12 @@ import org.springframework.util.ErrorHandler;
|
||||
*
|
||||
* @author Artem Bilan
|
||||
* @author Gary Russell
|
||||
*
|
||||
* @since 5.0
|
||||
*/
|
||||
public class AmqpMessageChannelSpec<S extends AmqpMessageChannelSpec<S>> extends AmqpPollableMessageChannelSpec<S> {
|
||||
|
||||
private final List<Advice> adviceChain = new LinkedList<Advice>();
|
||||
private final List<Advice> adviceChain = new LinkedList<>();
|
||||
|
||||
AmqpMessageChannelSpec(ConnectionFactory connectionFactory) {
|
||||
super(new AmqpChannelFactoryBean(true), connectionFactory);
|
||||
@@ -205,16 +206,28 @@ public class AmqpMessageChannelSpec<S extends AmqpMessageChannelSpec<S>> extends
|
||||
* Configure the txSize.
|
||||
* @param txSize the txSize.
|
||||
* @return the spec.
|
||||
* @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setTxSize(int)
|
||||
* @deprecated since 5.2 in favor of {@link #batchSize(int)}
|
||||
*/
|
||||
@Deprecated
|
||||
public S txSize(int txSize) {
|
||||
this.amqpChannelFactoryBean.setTxSize(txSize);
|
||||
return batchSize(txSize);
|
||||
}
|
||||
|
||||
/**
|
||||
* Configure the batch size.
|
||||
* @param batchSize the batchSize.
|
||||
* @return the spec.
|
||||
* @since 5.2
|
||||
* @see org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer#setBatchSize(int)
|
||||
*/
|
||||
public S batchSize(int batchSize) {
|
||||
this.amqpChannelFactoryBean.setBatchSize(batchSize);
|
||||
return _this();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected AbstractAmqpChannel doGet() {
|
||||
this.amqpChannelFactoryBean.setAdviceChain(this.adviceChain.toArray(new Advice[this.adviceChain.size()]));
|
||||
this.amqpChannelFactoryBean.setAdviceChain(this.adviceChain.toArray(new Advice[0]));
|
||||
return super.doGet();
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user