package com.orientechnologies.orient.server.hazelcast;

import com.hazelcast.core.HazelcastInstance;
import com.orientechnologies.orient.server.distributed.ODistributedLockManager;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

public class OHazelcastLockManager implements ODistributedLockManager {
  public OHazelcastLockManager(HazelcastInstance hazelcast) {
    this.hazelcast = hazelcast;
  }

  private HazelcastInstance hazelcast;
  private Set<String> lockedResurces = new HashSet<>();

  @Override
  public void acquireExclusiveLock(String resource, String nodeSource, long timeout) {
    if (timeout != 0) {
      try {
        hazelcast.getLock(resource).tryLock(timeout, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    } else {
      hazelcast.getLock(resource).lock();
    }
    synchronized (this) {
      lockedResurces.add(resource);
    }
  }

  @Override
  public void releaseExclusiveLock(String resource, String nodeSource) {
    hazelcast.getLock(resource).unlock();
    synchronized (this) {
      lockedResurces.remove(resource);
    }
  }

  @Override
  public void handleUnreachableServer(String nodeLeftName) {}

  @Override
  public void shutdown() {
    synchronized (this) {
      HashSet<String> res = new HashSet<>(lockedResurces);
      for (String resource : res) {
        releaseExclusiveLock(resource, null);
      }
    }
  }
}

@@ -94,16 +94,78 @@
 stance;%0A
+import com.orientechnologies.common.types.OModifiableInteger;%0A
 import c
@@ -232,24 +232,50 @@
 ockManager;%0A
+import java.util.HashMap;%0A
 import java.
@@ -284,24 +284,46 @@
 il.HashSet;%0A
+import java.util.Map;%0A
 import java.
@@ -443,16 +443,301 @@
 nager %7B%0A
+  private HazelcastInstance hazelcast;%0A  private Set%3CString%3E lockedResurces = new HashSet%3C%3E();%0A%0A  private static final ThreadLocal%3CMap%3CString, OModifiableInteger%3E%3E ACQUIRED_REENTRANT =%0A      ThreadLocal.withInitial(%0A          () -%3E %7B%0A            return new HashMap%3C%3E();%0A          %7D);%0A%0A
   public
@@ -831,104 +831,8 @@
  %7D%0A%0A
-  private HazelcastInstance hazelcast;%0A  private Set%3CString%3E lockedResurces = new HashSet%3C%3E();%0A%0A
   @O
@@ -922,24 +922,174 @@
  timeout) %7B%0A
+    OModifiableInteger counter;%0A    if ((counter = ACQUIRED_REENTRANT.get().get(resource)) != null) %7B%0A      counter.increment();%0A      return;%0A    %7D%0A%0A
     if (time
@@ -1403,32 +1403,103 @@
 esource);%0A    %7D%0A
+    ACQUIRED_REENTRANT.get().put(resource, new OModifiableInteger(0));%0A
   %7D%0A%0A  @Override
@@ -1568,24 +1568,335 @@
 deSource) %7B%0A
+    OModifiableInteger counter = ACQUIRED_REENTRANT.get().get(resource);%0A    if (counter == null)%0A      // DO NOTHING BECAUSE THE ACQUIRE DIDN'T HAPPEN IN DISTRIBUTED%0A      return;%0A%0A    if (counter.getValue() %3E 0) %7B%0A      counter.decrement();%0A      return;%0A    %7D%0A%0A    ACQUIRED_REENTRANT.get().remove(resource);%0A
     hazelcas
@@ -2151,44 +2151,31 @@
     
-HashSet%3C
+for (
 String
-%3E
  res
- = new HashSet%3C%3E(
+ource : 
 lock
@@ -2189,99 +2189,94 @@
 ces)
-;
+ %7B
 %0A      
-for (String resource : res) %7B%0A
+  hazelcast.getLock(resource).unlock();%0A   
    
+%7D%0A
      
-releaseExclusiveLock(r
+ lockedR
 es
-o
 urce
-, null);%0A      %7D
+s.clear();
 %0A   

add reentrant logic to distributed lock hazelcast implementation