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

GH-56: Publisher Confirms/Returns use Future

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp-samples/issues/56

`CorrelationData` now provides a `Future<?>` to wait for confirms/returns.
This commit is contained in:
Gary Russell
2019-11-27 16:25:19 -05:00
committed by Artem Bilan
parent 66e65055da
commit 476338e923
2 changed files with 18 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-rabbit-confirms-returns/README.adoc
View File

@@ -5,3 +5,5 @@ This sample demonstrates how to use publisher confirms and returns.
It uses an auto-delete queue that will be removed after the demo completes.
It is a Spring Boot application and can be run with normal boot run methods, e.g.: `./mvnw spring-boot:run`
It shows how to use the `Future<Confirm>` to wait for a delivery confirmation or returned message.

64
spring-rabbit-confirms-returns/src/main/java/org/springframework/amqp/samples/confirms/SpringRabbitConfirmsReturnsApplication.java
View File

@@ -19,12 +19,13 @@ package org.springframework.amqp.samples.confirms;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
@@ -38,7 +39,6 @@ public class SpringRabbitConfirmsReturnsApplication {
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(SpringRabbitConfirmsReturnsApplication.class,
args);
context.getBean(SpringRabbitConfirmsReturnsApplication.class).runDemo();
context.close();
}
@@ -47,20 +47,13 @@ public class SpringRabbitConfirmsReturnsApplication {
private final CountDownLatch listenLatch = new CountDownLatch(1);
private final CountDownLatch confirmLatch = new CountDownLatch(1);
private final CountDownLatch returnLatch = new CountDownLatch(1);
private void runDemo() throws Exception {
setupCallbacks();
// send a message to the default exchange to be routed to the queue
this.rabbitTemplate.convertAndSend("", QUEUE, "foo", new CorrelationData("Correlation for message 1"));
if (this.confirmLatch.await(10, TimeUnit.SECONDS)) {
System.out.println("Confirm received");
}
else {
System.out.println("Confirm NOT received");
}
CorrelationData correlationData = new CorrelationData("Correlation for message 1");
this.rabbitTemplate.convertAndSend("", QUEUE, "foo", correlationData);
Confirm confirm = correlationData.getFuture().get(10, TimeUnit.SECONDS);
System.out.println("Confirm received, ack = " + confirm.isAck());
if (this.listenLatch.await(10, TimeUnit.SECONDS)) {
System.out.println("Message received by listener");
}
@@ -68,16 +61,13 @@ public class SpringRabbitConfirmsReturnsApplication {
System.out.println("Message NOT received by listener");
}
// send a message to the default exchange to be routed to a non-existent queue
correlationData = new CorrelationData("Correlation for message 2");
this.rabbitTemplate.convertAndSend("", QUEUE + QUEUE, "bar", message -> {
System.out.println("Message after conversion: " + message);
return message;
});
if (this.returnLatch.await(10, TimeUnit.SECONDS)) {
System.out.println("Return received");
}
else {
System.out.println("Return NOT received");
}
}, correlationData);
correlationData.getFuture().get(10, TimeUnit.SECONDS);
System.out.println("Return received:" + correlationData.getReturnedMessage());
}
private void setupCallbacks() {
@@ -88,19 +78,17 @@ public class SpringRabbitConfirmsReturnsApplication {
if (correlation != null) {
System.out.println("Received " + (ack ? " ack " : " nack ") + "for correlation: " + correlation);
}
this.confirmLatch.countDown();
});
this.rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
System.out.println("Returned: " + message + "\nreplyCode: " + replyCode
+ "\nreplyText: " + replyText + "\nexchange/rk: " + exchange + "/" + routingKey);
this.returnLatch.countDown();
});
/*
* Replace the correlation data with one containing the converted message in case
* we want to resend it after a nack.
*/
this.rabbitTemplate.setCorrelationDataPostProcessor((message, correlationData) ->
new CompleteMessageCorrelationData(correlationData != null ? correlationData.getId() : null, message));
}
@Bean
public ApplicationRunner runner() {
return args -> runDemo();
}
@Bean
@@ -114,24 +102,4 @@ public class SpringRabbitConfirmsReturnsApplication {
this.listenLatch.countDown();
}
static class CompleteMessageCorrelationData extends CorrelationData {
private final Message message;
CompleteMessageCorrelationData(String id, Message message) {
super(id);
this.message = message;
}
public Message getMessage() {
return this.message;
}
@Override
public String toString() {
return "CompleteMessageCorrelationData [id=" + getId() + ", message=" + this.message + "]";
}
}
}