From a3ed443bad5e623c39284f3c49cfe7bd04da77d5 Mon Sep 17 00:00:00 2001
From: Andrea Cavalli <andrea@cavallium.it>
Date: Thu, 28 Jul 2022 23:44:23 +0200
Subject: [PATCH] Fix disposable

---
 .../it/cavallium/dbengine/database/LLUtils.java  | 16 +++++++++++++---
 1 file changed, 13 insertions(+), 3 deletions(-)

diff --git a/src/main/java/it/cavallium/dbengine/database/LLUtils.java b/src/main/java/it/cavallium/dbengine/database/LLUtils.java
index a950b6d..d302d1a 100644
--- a/src/main/java/it/cavallium/dbengine/database/LLUtils.java
+++ b/src/main/java/it/cavallium/dbengine/database/LLUtils.java
@@ -41,6 +41,7 @@ import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.ToIntFunction;
 import org.apache.logging.log4j.LogManager;
@@ -725,14 +726,23 @@ public class LLUtils {
 	}
 
 	public static Disposable scheduleRepeated(Scheduler scheduler, Runnable action, Duration delay) {
-		return scheduler.schedule(() -> {
+		var currentDisposable = new AtomicReference<Disposable>();
+		scheduleRepeatedInternal(scheduler, action, delay, currentDisposable);
+		return () -> currentDisposable.get().dispose();
+	}
+
+	private static void scheduleRepeatedInternal(Scheduler scheduler,
+			Runnable action,
+			Duration delay,
+			AtomicReference<Disposable> currentDisposable) {
+		currentDisposable.set(scheduler.schedule(() -> {
 			try {
 				action.run();
 			} catch (Throwable ex) {
 				logger.error(ex);
 			}
-			scheduleRepeated(scheduler, action, delay);
-		}, delay.toMillis(), TimeUnit.MILLISECONDS);
+			scheduleRepeatedInternal(scheduler, action, delay, currentDisposable);
+		}, delay.toMillis(), TimeUnit.MILLISECONDS));
 	}
 
 	@Deprecated