/**
 *    Copyright 2009-2019 the original author or authors.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package org.apache.ibatis.cache.decorators;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.ibatis.cache.Cache;
import org.apache.ibatis.cache.CacheException;

/**
 * Simple blocking decorator
 *
 * Simple and inefficient version of EhCache's BlockingCache decorator.
 * It sets a lock over a cache key when the element is not found in cache.
 * This way, other threads will wait until this element is filled instead of hitting the database.
 *
 * @author Eduardo Macarron
 *
 */
public class BlockingCache implements Cache {

  private long timeout;
  private final Cache delegate;
  private final ConcurrentHashMap<Object, ReentrantLock> locks;

  public BlockingCache(Cache delegate) {
    this.delegate = delegate;
    this.locks = new ConcurrentHashMap<>();
  }

  @Override
  public String getId() {
    return delegate.getId();
  }

  @Override
  public int getSize() {
    return delegate.getSize();
  }

  @Override
  public void putObject(Object key, Object value) {
    try {
      delegate.putObject(key, value);
    } finally {
      releaseLock(key);
    }
  }

  @Override
  public Object getObject(Object key) {
    acquireLock(key);
    Object value = delegate.getObject(key);
    if (value != null) {
      releaseLock(key);
    }
    return value;
  }

  @Override
  public Object removeObject(Object key) {
    // despite of its name, this method is called only to release locks
    releaseLock(key);
    return null;
  }

  @Override
  public void clear() {
    delegate.clear();
  }

  private ReentrantLock getLockForKey(Object key) {
    return locks.computeIfAbsent(key, k -> new ReentrantLock());
  }

  private void acquireLock(Object key) {
    Lock lock = getLockForKey(key);
    if (timeout > 0) {
      try {
        boolean acquired = lock.tryLock(timeout, TimeUnit.MILLISECONDS);
        if (!acquired) {
          throw new CacheException("Couldn't get a lock in " + timeout + " for the key " +  key + " at the cache " + delegate.getId());
        }
      } catch (InterruptedException e) {
        throw new CacheException("Got interrupted while trying to acquire lock for key " + key, e);
      }
    } else {
      lock.lock();
    }
  }

  private void releaseLock(Object key) {
    ReentrantLock lock = locks.get(key);
    if (lock.isHeldByCurrentThread()) {
      lock.unlock();
    }
  }

  public long getTimeout() {
    return timeout;
  }

  public void setTimeout(long timeout) {
    this.timeout = timeout;
  }
}

@@ -24,10 +24,10 @@
 9-20
-19
+20
  the
@@ -688,16 +688,48 @@
 ators;%0A%0A
+import java.text.MessageFormat;%0A
 import j
@@ -799,56 +799,22 @@
 ent.
-TimeUnit;%0Aimport java.util.concurrent.locks.Lock
+CountDownLatch
 ;%0Aim
@@ -843,27 +843,16 @@
 ent.
-locks.ReentrantLock
+TimeUnit
 ;%0A%0Ai
@@ -1401,29 +1401,30 @@
 Object, 
-ReentrantLock
+CountDownLatch
 %3E locks;
@@ -2297,74 +2297,149 @@
 ate 
-ReentrantLock getLockForKey(Object key) %7B%0A    return
+void acquireLock(Object key) %7B%0A    CountDownLatch newLatch = new CountDownLatch(1);%0A    while (true) %7B%0A      CountDownLatch latch =
  locks.
-com
 put
-e
 IfAb
@@ -2452,117 +2452,85 @@
 ey, 
-k -%3E new ReentrantLock());%0A  %7D%0A%0A  private void acquireLock(Object key) %7B%0A    Lock lock = getLockForKey(key);%0A
+newLatch);%0A      if (latch == null) %7B%0A        break;%0A      %7D%0A      try %7B%0A    
     
@@ -2550,26 +2550,16 @@
  0) %7B%0A  
-    try %7B%0A
         
@@ -2582,19 +2582,18 @@
  = l
-ock.tryLock
+atch.await
 (tim
@@ -2622,16 +2622,18 @@
 CONDS);%0A
+  
         
@@ -2651,32 +2651,34 @@
 ed) %7B%0A          
+  
 throw new CacheE
@@ -2686,16 +2686,33 @@
 ception(
+%0A                
 %22Couldn'
@@ -2758,17 +2758,16 @@
  key %22 +
- 
  key + %22
@@ -2803,16 +2803,70 @@
 tId());%0A
+          %7D%0A        %7D else %7B%0A          latch.await();%0A
         
@@ -3020,40 +3020,8 @@
   %7D%0A
-    %7D else %7B%0A      lock.lock();%0A
     
@@ -3076,26 +3076,28 @@
     
-ReentrantLock lock
+CountDownLatch latch
  = l
@@ -3101,19 +3101,22 @@
 = locks.
-get
+remove
 (key);%0A 
@@ -3127,65 +3127,170 @@
 f (l
-ock.isHeldByCurrentThread()) %7B%0A      lock.unlock();%0A    %7D
+atch == null) %7B%0A      throw new IllegalStateException(%22Detected an attempt at releasing unacquired lock. This should never happen.%22);%0A    %7D%0A    latch.countDown();
 %0A  %7D

fix #1261 Replace ReentrantLock with CountDownLatch to avoid OutOfMemoryError