'use strict';

Object.defineProperty(exports, "__esModule", {
  value: true
});

var _keys = require('babel-runtime/core-js/object/keys');

var _keys2 = _interopRequireDefault(_keys);

var _promise = require('babel-runtime/core-js/promise');

var _promise2 = _interopRequireDefault(_promise);

var _classCallCheck2 = require('babel-runtime/helpers/classCallCheck');

var _classCallCheck3 = _interopRequireDefault(_classCallCheck2);

var _createClass2 = require('babel-runtime/helpers/createClass');

var _createClass3 = _interopRequireDefault(_createClass2);

var _poolConfig = require('./pool-config');

var _poolConfig2 = _interopRequireDefault(_poolConfig);

var _error = require('../error');

var _logger = require('./logger');

var _logger2 = _interopRequireDefault(_logger);

function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }

var Pool = function () {

  /**
   * @param {function(function): Promise<object>} create  an allocation function that creates a promise with a new resource. It's given
   *                a single argument, a function that will return the resource to
   *                the pool if invoked, which is meant to be called on .dispose
   *                or .close or whatever mechanism the resource uses to finalize.
   * @param {function} destroy called with the resource when it is evicted from this pool
   * @param {function} validate called at various times (like when an instance is acquired and
   *                 when it is returned). If this returns false, the resource will
   *                 be evicted
   * @param {PoolConfig} config configuration for the new driver.
   * @param {Logger} log the driver logger.
   */
  function Pool(create) {
    var destroy = arguments.length > 1 && arguments[1] !== undefined ? arguments[1] : function () {
      return true;
    };
    var validate = arguments.length > 2 && arguments[2] !== undefined ? arguments[2] : function () {
      return true;
    };
    var config = arguments.length > 3 && arguments[3] !== undefined ? arguments[3] : _poolConfig2.default.defaultConfig();
    var log = arguments.length > 4 && arguments[4] !== undefined ? arguments[4] : _logger2.default.noOp();
    (0, _classCallCheck3.default)(this, Pool);

    this._create = create;
    this._destroy = destroy;
    this._validate = validate;
    this._maxSize = config.maxSize;
    this._acquisitionTimeout = config.acquisitionTimeout;
    this._pools = {};
    this._acquireRequests = {};
    this._activeResourceCounts = {};
    this._release = this._release.bind(this);
    this._log = log;
  }

  /**
   * Acquire and idle resource fom the pool or create a new one.
   * @param {string} key the resource key.
   * @return {object} resource that is ready to use.
   */


  (0, _createClass3.default)(Pool, [{
    key: 'acquire',
    value: function acquire(key) {
      var _this = this;

      return this._acquire(key).then(function (resource) {
        if (resource) {
          resourceAcquired(key, _this._activeResourceCounts);
          if (_this._log.isDebugEnabled()) {
            _this._log.debug(resource + ' acquired from the pool');
          }
          return resource;
        }

        // We're out of resources and will try to acquire later on when an existing resource is released.
        var allRequests = _this._acquireRequests;
        var requests = allRequests[key];
        if (!requests) {
          allRequests[key] = [];
        }

        return new _promise2.default(function (resolve, reject) {
          var request = void 0;

          var timeoutId = setTimeout(function () {
            // acquisition timeout fired

            // remove request from the queue of pending requests, if it's still there
            // request might've been taken out by the release operation
            var pendingRequests = allRequests[key];
            if (pendingRequests) {
              allRequests[key] = pendingRequests.filter(function (item) {
                return item !== request;
              });
            }

            if (request.isCompleted()) {
              // request already resolved/rejected by the release operation; nothing to do
            } else {
              // request is still pending and needs to be failed
              request.reject((0, _error.newError)('Connection acquisition timed out in ' + _this._acquisitionTimeout + ' ms.'));
            }
          }, _this._acquisitionTimeout);

          request = new PendingRequest(resolve, reject, timeoutId, _this._log);
          allRequests[key].push(request);
        });
      });
    }

    /**
     * Destroy all idle resources for the given key.
     * @param {string} key the resource key to purge.
     */

  }, {
    key: 'purge',
    value: function purge(key) {
      var pool = this._pools[key] || [];
      while (pool.length) {
        var resource = pool.pop();
        this._destroy(resource);
      }
      delete this._pools[key];
    }

    /**
     * Destroy all idle resources in this pool.
     */

  }, {
    key: 'purgeAll',
    value: function purgeAll() {
      var _this2 = this;

      (0, _keys2.default)(this._pools).forEach(function (key) {
        return _this2.purge(key);
      });
    }

    /**
     * Check if this pool contains resources for the given key.
     * @param {string} key the resource key to check.
     * @return {boolean} `true` when pool contains entries for the given key, <code>false</code> otherwise.
     */

  }, {
    key: 'has',
    value: function has(key) {
      return key in this._pools;
    }

    /**
     * Get count of active (checked out of the pool) resources for the given key.
     * @param {string} key the resource key to check.
     * @return {number} count of resources acquired by clients.
     */

  }, {
    key: 'activeResourceCount',
    value: function activeResourceCount(key) {
      return this._activeResourceCounts[key] || 0;
    }
  }, {
    key: '_acquire',
    value: function _acquire(key) {
      var pool = this._pools[key];
      if (!pool) {
        pool = [];
        this._pools[key] = pool;
      }
      while (pool.length) {
        var resource = pool.pop();

        if (this._validate(resource)) {
          // idle resource is valid and can be acquired
          return _promise2.default.resolve(resource);
        } else {
          this._destroy(resource);
        }
      }

      if (this._maxSize && this.activeResourceCount(key) >= this._maxSize) {
        return _promise2.default.resolve(null);
      }

      // there exist no idle valid resources, create a new one for acquisition
      return this._create(key, this._release);
    }
  }, {
    key: '_release',
    value: function _release(key, resource) {
      var pool = this._pools[key];

      if (pool) {
        // there exist idle connections for the given key
        if (!this._validate(resource)) {
          if (this._log.isDebugEnabled()) {
            this._log.debug(resource + ' destroyed and can\'t be released to the pool ' + key + ' because it is not functional');
          }
          this._destroy(resource);
        } else {
          if (this._log.isDebugEnabled()) {
            this._log.debug(resource + ' released to the pool ' + key);
          }
          pool.push(resource);
        }
      } else {
        // key has been purged, don't put it back, just destroy the resource
        if (this._log.isDebugEnabled()) {
          this._log.debug(resource + ' destroyed and can\'t be released to the pool ' + key + ' because pool has been purged');
        }
        this._destroy(resource);
      }
      resourceReleased(key, this._activeResourceCounts);

      this._processPendingAcquireRequests(key);
    }
  }, {
    key: '_processPendingAcquireRequests',
    value: function _processPendingAcquireRequests(key) {
      var _this3 = this;

      var requests = this._acquireRequests[key];
      if (requests) {
        var pendingRequest = requests.shift(); // pop a pending acquire request

        if (pendingRequest) {
          this._acquire(key).catch(function (error) {
            // failed to acquire/create a new connection to resolve the pending acquire request
            // propagate the error by failing the pending request
            pendingRequest.reject(error);
            return null;
          }).then(function (resource) {
            if (resource) {
              // managed to acquire a valid resource from the pool

              if (pendingRequest.isCompleted()) {
                // request has been completed, most likely failed by a timeout
                // return the acquired resource back to the pool
                _this3._release(key, resource);
              } else {
                // request is still pending and can be resolved with the newly acquired resource
                resourceAcquired(key, _this3._activeResourceCounts); // increment the active counter
                pendingRequest.resolve(resource); // resolve the pending request with the acquired resource
              }
            }
          });
        } else {
          delete this._acquireRequests[key];
        }
      }
    }
  }]);
  return Pool;
}();

/**
 * Increment active (checked out of the pool) resource counter.
 * @param {string} key the resource group identifier (server address for connections).
 * @param {Object.<string, number>} activeResourceCounts the object holding active counts per key.
 */
/**
 * Copyright (c) 2002-2018 "Neo4j,"
 * Neo4j Sweden AB [http://neo4j.com]
 *
 * This file is part of Neo4j.
 *
 * 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.
 */

function resourceAcquired(key, activeResourceCounts) {
  var currentCount = activeResourceCounts[key] || 0;
  activeResourceCounts[key] = currentCount + 1;
}

/**
 * Decrement active (checked out of the pool) resource counter.
 * @param {string} key the resource group identifier (server address for connections).
 * @param {Object.<string, number>} activeResourceCounts the object holding active counts per key.
 */
function resourceReleased(key, activeResourceCounts) {
  var currentCount = activeResourceCounts[key] || 0;
  var nextCount = currentCount - 1;

  if (nextCount > 0) {
    activeResourceCounts[key] = nextCount;
  } else {
    delete activeResourceCounts[key];
  }
}

var PendingRequest = function () {
  function PendingRequest(resolve, reject, timeoutId, log) {
    (0, _classCallCheck3.default)(this, PendingRequest);

    this._resolve = resolve;
    this._reject = reject;
    this._timeoutId = timeoutId;
    this._log = log;
    this._completed = false;
  }

  (0, _createClass3.default)(PendingRequest, [{
    key: 'isCompleted',
    value: function isCompleted() {
      return this._completed;
    }
  }, {
    key: 'resolve',
    value: function resolve(resource) {
      if (this._completed) {
        return;
      }
      this._completed = true;

      clearTimeout(this._timeoutId);
      if (this._log.isDebugEnabled()) {
        this._log.debug(resource + ' acquired from the pool');
      }
      this._resolve(resource);
    }
  }, {
    key: 'reject',
    value: function reject(error) {
      if (this._completed) {
        return;
      }
      this._completed = true;

      clearTimeout(this._timeoutId);
      this._reject(error);
    }
  }]);
  return PendingRequest;
}();

exports.default = Pool;