feat: implement our own semaphore

Replacing 'await-semaphore' since we need the `test()` function
and there does not appear to be a widely used/maintained semaphore
implementation that provides it and doesn't also depend on the
`async/await` keywords at runtime.

lib/circuit.js

@@ -3,7 +3,7 @@
 const EventEmitter = require('events');
 const Status = require('./status');
 const HystrixStats = require('./hystrix-stats');
-const Semaphore = require('await-semaphore').Semaphore;
+const Semaphore = require('./semaphore');
 
 const STATE = Symbol('state');
 const OPEN = Symbol('open');
@@ -79,6 +79,7 @@ class CircuitBreaker extends EventEmitter {
     this.on('cacheMiss', increment('cacheMisses'));
     this.on('open', _ => this[STATUS].open());
     this.on('close', _ => this[STATUS].close());
+    this.on('semaphore-locked', increment('semaphoreRejections'));
 
     /**
      * Emitted after `options.resetTimeout` has elapsed, allowing for
@@ -277,9 +278,9 @@ class CircuitBreaker extends EventEmitter {
 
     let timeout;
     let timeoutError = false;
-    return this.semaphore.acquire().then((release) => {
-      return new Promise((resolve, reject) => {
-        const latencyStartTime = Date.now();
+    return new Promise((resolve, reject) => {
+      const latencyStartTime = Date.now();
+      if (this.semaphore.test()) {
         timeout = setTimeout(
           () => {
             timeoutError = true;
@@ -290,7 +291,7 @@ class CircuitBreaker extends EventEmitter {
              * @event CircuitBreaker#timeout
              */
             const latency = Date.now() - latencyStartTime;
-            release();
+            this.semaphore.release();
             this.emit('timeout', error, latency);
             resolve(handleError(error, this, timeout, args, latency, resolve, reject));
           }, this.options.timeout);
@@ -309,24 +310,30 @@ class CircuitBreaker extends EventEmitter {
                * @event CircuitBreaker#success
                */
               this.emit('success', result, (Date.now() - latencyStartTime));
-              release();
+              this.semaphore.release();
               resolve(result);
               if (this.options.cache) {
                 CACHE.set(this, promise);
               }
             }
           })
           .catch((error) => {
-            release();
+            this.semaphore.release();
             const latencyEndTime = Date.now() - latencyStartTime;
             handleError(error, this, timeout, args, latencyEndTime, resolve, reject);
           });
         } catch (error) {
-          release();
+          this.semaphore.release();
           const latency = Date.now() - latencyStartTime;
           handleError(error, this, timeout, args, latency, resolve, reject);
         }
-      });
+      } else {
+        const latency = Date.now() - latencyStartTime;
+        const err = new Error('Semaphore locked');
+        err.code = 'ESEMLOCKED';
+        this.emit('semaphore-locked', err, latency);
+        handleError(err, this, timeout, args, latency, resolve, reject);
+      }
     });
   }
 

lib/semaphore.js

@@ -0,0 +1,54 @@
+'use strict';
+
+module.exports = exports = semaphore;
+
+function semaphore (count) {
+  const resolvers = [];
+  let counter = count;
+
+  let sem = {
+    take,
+    release,
+    test
+  };
+
+  Object.defineProperty(sem, 'count', {
+    get: _ => counter,
+    enumerable: true
+  });
+
+  return sem;
+
+  function take (timeout) {
+    if (counter > 0) {
+      --counter;
+      return Promise.resolve(release);
+    }
+    return new Promise((resolve, reject) => {
+      resolvers.push(_ => {
+        --counter;
+        resolve(release);
+      });
+      if (timeout) {
+        setTimeout(_ => {
+          resolvers.shift();
+          const err = new Error(`Timed out after ${timeout}ms`);
+          err.code = 'ETIMEDOUT';
+          reject(err);
+        }, timeout);
+      }
+    });
+  }
+
+  function release () {
+    counter++;
+    if (resolvers.length > 0) {
+      resolvers.shift()();
+    }
+  }
+
+  function test () {
+    if (counter < 1) return false;
+    return take() && true;
+  }
+}

lib/status.js

@@ -164,6 +164,7 @@ const bucket = _ => ({
   timeouts: 0,
   cacheHits: 0,
   cacheMisses: 0,
+  semaphoreRejections: 0,
   percentiles: {},
   latencyTimes: []
 });