Added support for stream pause and resume callbacks (#5)

* Implement stream pausing callback support

* Update examples and add an example for pausing

* Cleaned up the code

Author: OlliV

examples/first.js

@@ -1,7 +1,6 @@
-#!/bin/sh
-// >&/dev/null;exec node --harmony_async_await $0 $@
+#!/usr/bin/env node
 
-const Sema = require('./index.js')
+const Sema = require('../index.js')
 
 function getRnd (min, max) {
   min = Math.ceil(min)

examples/second.js

@@ -1,7 +1,6 @@
-#!/bin/sh
-// >&/dev/null;exec node --harmony_async_await $0 $@
+#!/usr/bin/env node
 
-const Sema = require('./index.js')
+const Sema = require('../index.js')
 const redis = require('promise-redis')
 
 async function f () {

examples/third.js

@@ -0,0 +1,33 @@
+#!/usr/bin/env node
+
+// Usage ./third.js < VERY_LARGE_FILE
+
+const Sema = require('../index.js')
+const readline = require('readline')
+
+const rl = readline.createInterface({
+  input: process.stdin,
+  output: process.stdout,
+  terminal: false
+})
+
+function pause() {
+  console.log('Pausing the stream')
+}
+
+function resume() {
+  console.log('Resuming the stream')
+}
+
+const s = new Sema(5, { pauseFn: pause, resumeFn: resume })
+async function parse(line) {
+  await s.v()
+
+  console.log(line)
+
+  s.p()
+}
+
+rl.on('line', (line) => {
+  parse(line).catch(console.error)
+})

index.js

@@ -11,22 +11,33 @@ function defaultInit () {
 }
 
 class Sema {
-  constructor (nr, { initFn, capacity }) {
+  constructor (nr, { initFn, pauseFn, resumeFn, capacity }) {
     initFn = initFn || defaultInit
     capacity = capacity || 10
 
+    if (pauseFn ^ resumeFn) {
+      throw new Error('pauseFn and resumeFn must be both set for pausing')
+    }
+
     this.nrTokens = nr
     this.free = new Deque(nr)
     this.waiting = new Deque(capacity)
     this.releaseEmitter = new ReleaseEmitter()
     this.noTokens = initFn === defaultInit
+    this.pauseFn = pauseFn
+    this.resumeFn = resumeFn
 
     this.releaseEmitter.on('release', (token) => {
       const p = this.waiting.shift()
       if (p) {
         p.resolve(token)
       } else {
-       this.free.push(token)
+        if (this.resumeFn && this.paused) {
+          this.paused = false
+          this.resumeFn()
+        }
+
+        this.free.push(token)
       }
     })
 
@@ -37,10 +48,17 @@ class Sema {
 
   async v () {
     let token = this.free.pop()
-    if (token)
+
+    if (token) {
       return token
+    }
 
     return new Promise((resolve, reject) => {
+      if (this.pauseFn && !this.paused) {
+        this.paused = true
+        this.pauseFn()
+      }
+
       this.waiting.push({ resolve, reject })
     })
   }