From 6cc596157c596a53a3574604f46a8b36e5da7180 Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Thu, 10 Aug 2023 17:14:13 +0200 Subject: [PATCH] coordinate the task completion with lifecycle stop --- .../com/example/amqp/AmqpRabbitApplication.java | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/integration/spring-amqp-rabbit/src/main/java/com/example/amqp/AmqpRabbitApplication.java b/integration/spring-amqp-rabbit/src/main/java/com/example/amqp/AmqpRabbitApplication.java index c501627..5d76da1 100644 --- a/integration/spring-amqp-rabbit/src/main/java/com/example/amqp/AmqpRabbitApplication.java +++ b/integration/spring-amqp-rabbit/src/main/java/com/example/amqp/AmqpRabbitApplication.java @@ -15,6 +15,7 @@ */ package com.example.amqp; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -78,16 +79,21 @@ public class AmqpRabbitApplication implements SmartLifecycle { return new Queue("cf2"); } + private AtomicBoolean isRunning = new AtomicBoolean(false); + + private Semaphore taskCompletionSemaphore = new Semaphore(1); + @Scheduled(fixedDelay = 1, timeUnit = TimeUnit.SECONDS) public void doSendMessage() { + this.taskCompletionSemaphore.acquireUninterruptibly(); if (this.isRunning.get()) { System.out.println("++++++ Received: " + template.convertSendAndReceive("", "cf1", "one")); } + this.taskCompletionSemaphore.release(); + } // Use the SmartLifecycle to synch the message sending with the checkpoint/restore. - private AtomicBoolean isRunning = new AtomicBoolean(false); - @Override public void start() { this.isRunning.set(true); @@ -96,6 +102,9 @@ public class AmqpRabbitApplication implements SmartLifecycle { @Override public void stop() { this.isRunning.set(false); + // wait until the running task complete. + this.taskCompletionSemaphore.acquireUninterruptibly(); + this.taskCompletionSemaphore.release(); } @Override