/*
 *
 *  *  Copyright 2014 Orient Technologies LTD (info(at)orientechnologies.com)
 *  *
 *  *  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.
 *  *
 *  * For more information: http://www.orientechnologies.com
 *
 */
package com.orientechnologies.common.concur.resource;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import com.orientechnologies.common.concur.lock.OInterruptedException;
import com.orientechnologies.common.concur.lock.OLockException;

/**
 * Generic non reentrant implementation about pool of resources. It pre-allocates a semaphore of maxResources. Resources are lazily
 * created by invoking the listener.
 * 
 * @author Luca Garulli (l.garulli--at--orientechnologies.com)
 * @param <K>
 *          Resource's Key
 * @param <V>
 *          Resource Object
 */
public class OResourcePool<K, V> {
  protected final Semaphore             sem;
  protected final Queue<V>              resources    = new ConcurrentLinkedQueue<V>();
  protected final Queue<V>              resourcesOut = new ConcurrentLinkedQueue<V>();
  protected final Collection<V>         unmodifiableresources;
  protected OResourcePoolListener<K, V> listener;
  protected volatile int                created      = 0;

  public OResourcePool(final int maxResources, final OResourcePoolListener<K, V> listener) {
    if (maxResources < 1)
      throw new IllegalArgumentException("iMaxResource must be major than 0");

    this.listener = listener;
    sem = new Semaphore(maxResources, true);
    unmodifiableresources = Collections.unmodifiableCollection(resources);
  }

  public V getResource(K key, final long maxWaitMillis, Object... additionalArgs) throws OLockException {
    // First, get permission to take or create a resource
    try {
      if (!sem.tryAcquire(maxWaitMillis, TimeUnit.MILLISECONDS))
        throw new OLockException("No more resources available in pool. Requested resource: " + key);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new OInterruptedException(e);
    }

    V res;
    do {
      // POP A RESOURCE
      res = resources.poll();
      if (res != null) {
        // TRY TO REUSE IT
        if (listener.reuseResource(key, additionalArgs, res)) {
          // OK: REUSE IT
          break;
        } else
          res = null;

        // UNABLE TO REUSE IT: THE RESOURE WILL BE DISCARDED AND TRY WITH THE NEXT ONE, IF ANY
      }
    } while (!resources.isEmpty());

    // NO AVAILABLE RESOURCES: CREATE A NEW ONE
    try {
      if (res == null)
        res = listener.createNewResource(key, additionalArgs);
      resourcesOut.add(res);
      created++;
      return res;
    } catch (RuntimeException e) {
      sem.release();
      // PROPAGATE IT
      throw e;
    } catch (Exception e) {
      sem.release();

      throw new OLockException("Error on creation of the new resource in the pool", e);
    }
  }

  public int getMaxResources() {
    return sem.availablePermits();
  }

  public int getAvailableResources() {
    return resources.size();
  }

  public boolean returnResource(final V res) {
    if (resourcesOut.remove(res)) {
      resources.add(res);
      sem.release();
    }
    return true;
  }

  public Collection<V> getResources() {
    return unmodifiableresources;
  }

  public void close() {
    sem.drainPermits();
  }

  public Collection<V> getAllResources() {
    List<V> all = new ArrayList<V>(resources);
    all.addAll(resourcesOut);
    return all;
  }

  public void remove(final V res) {
    if (resourcesOut.remove(res)) {
      this.resources.remove(res);
      sem.release();
    }
  }

  public int getCreatedInstances() {
    return created;
  }
}

@@ -1207,16 +1207,69 @@
 ception;
+%0Aimport com.orientechnologies.common.log.OLogManager;
 %0A%0A/**%0A *
@@ -3330,16 +3330,18 @@
 == null)
+ %7B
 %0A       
@@ -3406,46 +3406,459 @@
     
-resourcesOut.add(res);%0A      created++
+  created++;%0A        if (OLogManager.instance().isDebugEnabled())%0A          OLogManager.instance().debug(this, %22pool:'%25s' created new resource '%25s', new resource count '%25d'%22, this, res, created);%0A      %7D%0A      resourcesOut.add(res);%0A      if (OLogManager.instance().isDebugEnabled())%0A        OLogManager.instance().debug(this, %22pool:'%25s' acquired resource '%25s' available %25d out %25d %22, this, res,%0A            sem.availablePermits(), resourcesOut.size())
 ;%0A  
@@ -4383,32 +4383,251 @@
  sem.release();%0A
+      if (OLogManager.instance().isDebugEnabled())%0A        OLogManager.instance().debug(this, %22pool:'%25s' returned resource '%25s' available %25d out %25d%22, this, res,%0A            sem.availablePermits(), resourcesOut.size());%0A
     %7D%0A    return
@@ -5034,24 +5034,242 @@
 .release();%0A
+      if (OLogManager.instance().isDebugEnabled())%0A        OLogManager.instance().debug(this, %22pool:'%25s' removed resource '%25s' available %25d out %25d%22, this, res,%0A            sem.availablePermits(), resourcesOut.size());%0A
     %7D%0A  %7D%0A%0A 

add logging information in the pool with available an used resurces. issue #2891