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

AMQP-451 Override 4 Argument send() (Batching)

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-451

Spring Integration always invokes the 4 arg method.
This commit is contained in:
Gary Russell
2014-12-11 18:15:06 -05:00
parent 107bc52d4c
commit 2f22bb6e42
1 changed files with 26 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplate.java
View File

@@ -22,6 +22,7 @@ import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.support.BatchingStrategy;
import org.springframework.amqp.rabbit.core.support.MessageBatch;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.scheduling.TaskScheduler;
/**
@@ -55,29 +56,39 @@ public class BatchingRabbitTemplate extends RabbitTemplate {
}
@Override
public synchronized void send(String exchange, String routingKey, Message message) throws AmqpException {
if (this.scheduledTask != null) {
this.scheduledTask.cancel(false);
public synchronized void send(String exchange, String routingKey, Message message, CorrelationData correlationData)
throws AmqpException {
if (correlationData != null) {
if (logger.isDebugEnabled()) {
logger.debug("Cannot use batching with correlation data");
}
super.send(exchange, routingKey, message, correlationData);
}
MessageBatch batch = this.batchingStrategy.addToBatch(exchange, routingKey, message);
if (batch != null) {
super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage());
}
Date next = this.batchingStrategy.nextRelease();
if (next != null) {
this.scheduledTask = this.scheduler.schedule(new Runnable() {
else {
if (this.scheduledTask != null) {
this.scheduledTask.cancel(false);
}
MessageBatch batch = this.batchingStrategy.addToBatch(exchange, routingKey, message);
if (batch != null) {
super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage(), null);
}
Date next = this.batchingStrategy.nextRelease();
if (next != null) {
this.scheduledTask = this.scheduler.schedule(new Runnable() {
@Override
public void run() {
releaseBatches();
}}, next);
@Override
public void run() {
releaseBatches();
}
}, next);
}
}
}
private synchronized void releaseBatches() {
MessageBatch batch;
while ((batch = this.batchingStrategy.releaseBatch()) != null) {
super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage());
super.send(batch.getExchange(), batch.getRoutingKey(), batch.getMessage(), null);
}
}