Support for delayClose.

Author: mihxil

vpro-shared-util/src/main/java/nl/vpro/util/locker/ObjectLocker.java

@@ -28,6 +28,8 @@
 @Slf4j
 public class ObjectLocker {
 
+
+
     public static BiPredicate<StackTraceElement, AtomicInteger> summaryBiPredicate =
         (e, count) -> {
             boolean match = e.getClassName().startsWith("nl.vpro") &&
@@ -205,7 +207,7 @@ public static <K extends Serializable> LockHolderCloser<Serializable> acquireLoc
 
 
     @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
-    public static  <K extends Serializable> LockHolderCloser<K> acquireLock(
+    private static  <K extends Serializable> LockHolderCloser<K> acquireLock(
         final  K key,
         final @NonNull String reason,
         final @NonNull Map<K, LockHolder<K>> locks,
@@ -227,7 +229,7 @@ public static  <K extends Serializable> LockHolderCloser<K> acquireLock(
                         locks.remove(key);
                         continue;
                     }
-                    closer = new LockHolderCloser<>(nanoStart, locks, holder, true);
+                    closer = new LockHolderCloser<>(nanoStart, locks, holder, comparable);
                     if (holder.lock.isLocked() && !holder.lock.isHeldByCurrentThread()) {
                         log.debug("There are already threads ({}) for {}, waiting", holder.lock.getQueueLength(), key);
                         alreadyWaiting = true;
@@ -324,31 +326,29 @@ private static <K extends Serializable>  LockHolder<K> computeLock(
     @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
     private static  <K extends Serializable> void releaseLock(
         final long nanoStart,
-        final @NonNull K key,
-        final @NonNull  String reason,
         final @NonNull Map<K, LockHolder<K>> locks,
-        final @NonNull LockHolder<K> lock,
-        final boolean checkIfCurrentThread) {
+        final @NonNull LockHolder<K> lock) {
         synchronized (locks) {
             if (lock.lock.getHoldCount() == 1) {
                 if (!lock.lock.hasQueuedThreads()) {
-                    log.trace("Removed {}", key);
-                    locks.remove(key);
+                    log.trace("Removed {}", lock.key);
+                    locks.remove(lock.key);
                 }
                 final Duration duration = Duration.ofNanos(System.nanoTime() - nanoStart);
                 for (Listener listener : LISTENERS) {
                     listener.unlock(lock, duration);
                 }
 
                 Slf4jHelper.log(log, duration.compareTo(lock.warnTime)> 0 ? Level.WARN :  Level.DEBUG,
-                    "Released lock for {} ({}) in {}", key, reason, Duration.ofNanos(System.nanoTime() - nanoStart));
+                    "Released lock for {} ({}) in {}", lock.key, lock.reason, Duration.ofNanos(System.nanoTime() - nanoStart));
             }
-            if (! checkIfCurrentThread || lock.lock.isHeldByCurrentThread()) { // MSE-4946
+            if (lock.lock.isHeldByCurrentThread()) { // MSE-4946
                 HOLDS.get().remove(lock);
                 lock.lock.unlock();
             } else {
                 // can happen if 'continuing without lock'
-                log.warn("Current lock {} not hold by current thread {} but by {}", lock, Thread.currentThread().getName(), Optional.ofNullable(lock.thread.get()).map(Thread::getName).orElse(null));
+                Thread currentThread = Thread.currentThread();
+                log.warn("Current lock {} not hold by current thread {} ({}) but by {} ({})", lock, currentThread.getName(), currentThread, Optional.ofNullable(lock.thread.get()).map(Thread::getName).orElse(null), lock.thread.get(), new Exception());
             }
 
             locks.notifyAll();
@@ -383,6 +383,7 @@ public static class LockHolder<K> {
         @Setter
         private Duration warnTime = ObjectLocker.defaultWarnTime;
 
+
         LockHolder(K k, String reason, ReentrantLock lock) {
             this.key = k;
             this.lock = lock;
@@ -391,6 +392,8 @@ public static class LockHolder<K> {
             this.reason = reason;
         }
 
+
+
         @SuppressWarnings("rawtypes")
         @Override
         public boolean equals(Object o) {
@@ -457,25 +460,83 @@ public static class LockHolderCloser<K extends Serializable> implements AutoClos
 
         final long nanoStart;
 
-        @With
-        final boolean checkIfCurrentThread;
-
         final @NonNull Map<K, LockHolder<K>> locks;
 
+        final BiPredicate<Serializable, K> comparable;
+        boolean closed = false;
+
         private LockHolderCloser(
             final long nanoStart,
             @NonNull Map<K, LockHolder<K>> locks,
             @NonNull LockHolder<K> lockHolder,
-            final boolean checkIfCurrentThread) {
+            BiPredicate<Serializable, K> comparable
+            ) {
             this.nanoStart = nanoStart;
             this.locks = locks;
             this.lockHolder = lockHolder;
-            this.checkIfCurrentThread = checkIfCurrentThread;
+            this.comparable = comparable;
+        }
+
+        public CompletableFuture<LockHolderCloser<K>> forTransfer() {
+            close();
+            CompletableFuture<LockHolderCloser<K>> future = new CompletableFuture<>();
+
+
+            future.thenAccept((o) -> {
+                if (log.isDebugEnabled()) {
+                    log.debug("Completed transfer {} ({}) -> {} ({})", LockHolderCloser.this, LockHolderCloser.this.lockHolder.thread.get(), o, o.lockHolder.thread.get());
+                }
+            });
+            log.debug("Future to use {}", future);
+            return future;
+        }
+
+        public LockHolderCloser<K> transfer(CompletableFuture<LockHolderCloser<K>> future) {
+            synchronized (locks) {
+                try {
+
+                    LockHolderCloser<K> closer = acquireLock(lockHolder.key, lockHolder.reason, locks, comparable);
+                    future.complete(closer);
+                    return closer;
+                } catch (Exception e) {
+                    return null;
+                }
+            }
+        }
+        public CompletableFuture<Void> delayedClose(Duration duration) {
+            CompletableFuture<Void> ready = new CompletableFuture<>();
+            final CompletableFuture<LockHolderCloser<K>> future = forTransfer();
+            ForkJoinPool.commonPool().execute(() -> {
+
+                try (var closer = transfer(future)) {
+                    log.debug("{} Starting delay", lockHolder.key);
+                    Thread.sleep(duration.toMillis());
+                    log.info("{} Ready after delay {}", lockHolder.key, duration);
+                    ready.complete(null);
+                } catch (InterruptedException e) {
+                    log.warn(e.getMessage(), e);
+                } finally {
+
+                }
+            });
+            return  ready;
         }
 
         @Override
         public void close() {
-            releaseLock(nanoStart, lockHolder.key, lockHolder.reason, locks, lockHolder, checkIfCurrentThread);
+            synchronized (locks) {
+                if (!closed) {
+                    releaseLock(nanoStart, locks, lockHolder);
+                } else {
+                    log.debug("Closed already");
+                }
+                closed = true;
+            }
+        }
+
+        @Override
+        public  String toString() {
+            return lockHolder + (closed ? " (closed)" : "");
         }
     }
 

vpro-shared-util/src/test/java/nl/vpro/util/locker/ObjectLockerTest.java

@@ -7,8 +7,8 @@
 import java.io.Serial;
 import java.io.Serializable;
 import java.time.Duration;
-import java.util.*;
 import java.util.AbstractMap.SimpleEntry;
+import java.util.List;
 import java.util.concurrent.*;
 
 import org.junit.jupiter.api.*;
@@ -222,6 +222,19 @@ public void twoDifferentLocksStrictly() {
         }).isInstanceOf(IllegalStateException.class);
     }
 
+    @Test
+    public void delayedClose() throws InterruptedException, ExecutionException {
+        CompletableFuture<Void> voidCompletableFuture;
+        try (var closer = ObjectLocker.acquireLock("bla", "test", ObjectLocker.LOCKED_OBJECTS)) {
+            Thread.sleep(100);
+            log.info("DelayClosing");
+            voidCompletableFuture = closer.delayedClose(Duration.ofSeconds(10));
+        }
+        log.info("going ahhead..");
+        voidCompletableFuture.get();
+
+    }
+
 
 
     @SneakyThrows