<commit_before>
/*
 * Stream-NG - Next generation of Stream interface for NodeJS
 *
 * Copyright (c) 2016 vmolsa <<EMAIL>> (http://github.com/vmolsa)
 * MIT License <https://github.com/vmolsa/stream-ng/blob/master/LICENSE>
 *
 */

var _ = {
  isArray: function(arg) {
    return (Object.prototype.toString.call(arg) === '[object Array]');
  },
  isObject: function(arg) {
    return (arg && typeof(arg) === 'object');
  },
  isString: function(arg) {
    return (typeof(arg) === 'string');
  },
  isUndefined: function(arg) {
    return (typeof(arg) === 'undefined');
  },
  isFunction: function(arg) {
    return (typeof(arg) === 'function');
  },
  isNumber: function(arg) {
    return (typeof(arg) === 'number');
  },
  isTypedArray: function(arg) {
    return ArrayBuffer.isView(arg);
  },
  isArrayBuffer: function(arg) {
    return (arg instanceof ArrayBuffer);
  },
  isError: function(arg) {
    return (arg instanceof Error);
  },
  isPromise: function(arg) {
    return (arg && typeof(arg.then) === 'function');
  },
  size: function(arg) {
    if (_.isObject(arg)) {
      return Object.keys(arg).length;
    } else if (_.isArray(arg)) {
      return arg.length;
    } else if (_.isString(arg)) {
      return arg.length;
    } else if (_.isTypedArray(arg) || _.isArrayBuffer(arg)) {
      return arg.byteLength;
    } else {
      return -1;
    }
  },
  now: function() {
    return new Date().getTime();
  },
  extend: function(dst, src) {
    Object.keys(src).forEach(function(key) {
      dst[key] = src[key];
    });
    
    return dst;
  },
  forEach: function(collection, callback) {
    if (_.isFunction(callback)) {
      if (_.isObject(collection)) {
        Object.keys(collection).forEach(function(key) {
          callback(collection[key], key);
        });
      } else if (_.isArray(collection)) {
        collection.forEach(callback);
      }
    }
    return collection;
  },
  once: function(callback, self) {
    return function(arg) {     
      if (_.isFunction(callback)) {
        callback.call(self || this, arg);
      }
      
      callback = undefined;
    };
  },
  delay: process.nextTick,
  random: function() {
    return (Math.random() * (-1 >>> 0)) >>> 0;
  },
  sll_add: function(collection, arg) {
    return {
      element: arg,
      next: collection,
    };
  },
  sll_forEach: function(collection, callback) {
    if (_.isFunction(callback)) {
      for (var index = collection; index; index = index.next) {
        callback(index.element);
      }
    }
    
    return collection;
  },
};

function Promise(callback) {
  var self = this;
 
  if (!(self instanceof Promise)) {
    return new Promise(callback);
  }
  
  self.pending = true;
  self.fulfilled = false;
  self.rejected = false;
  
  function onReject(arg) {
    _.delay(function() {    
      if (!self.fulfilled && !self.rejected) {
        self.pending = false;
        self.fulfilled = false;
        self.rejected = true;
        
        _.sll_forEach(self.onreject, function(callback) {
          callback.call(self, arg);
        });
        
        self.destroy(arg);
      }
    });
  }
  
  function onResolve(arg) {
    if (_.isPromise(arg)) {
      arg.then(function(reply) {
        onResolve(reply);
      }, function(error) {
        onReject(error);
      });
    } else {
      _.delay(function() {      
        if (!self.fulfilled && !self.rejected) {
          self.pending = false;
          self.fulfilled = true;
          self.rejected = false;
          
          _.sll_forEach(self.onresolve, function(callback) {
            try {
              callback.call(self, arg);
            } catch(error) {
              onReject(error);
            }
          });

          self.destroy();
        }
      });
    }
  }
  
  if (_.isFunction(callback)) {    
    try {
      callback(onResolve, onReject);
    } catch(error) {
      onReject(error);
    }
  } else {
    throw new Error('Promise resolver ' + typeof(callback) + ' is not a function');
  }
}

Promise.prototype.destroy = function() {
  this.onresolve = null;
  this.onreject = null;
}

Promise.prototype.then = function(onResolve, onReject) {  
  if (this.pending) {
    if (_.isFunction(onResolve)) {
      this.onresolve = _.sll_add(this.onresolve, onResolve);
    }
    
    if (_.isFunction(onReject)) {
      this.onreject = _.sll_add(this.onreject, onReject);
    }
  }
  
  return this;
};

Promise.prototype.catch = function(onReject) {  
  if (this.pending) {
    if (_.isFunction(onReject)) {
      this.onreject = _.sll_add(this.onreject, onReject);
    }
  }
  
  return this;
};

Promise.prototype.finally = function(onResolve) {  
  return this.then(function(arg) {
    onResolve(arg);
  }, function(error) {
    onResolve(error);
  });
};

Promise.all = function(collection) {
  return new Promise(function(resolve, reject) {
    var len = _.size(collection);
    
    if (_.isArray(collection) && len) {
      var args = [];
      
      collection.forEach(function(value, index) {
        if (_.isPromise(value)) {
          var curIndex = index;
          
          value.then(function(reply) {
            args[curIndex] = reply;
            len--;
            
            if (!len) {
              resolve(args);
            }
          }).catch(function(error) {
            reject(error);
          });
        } else {
          args[index] = value;
          len--;
            
          if (!len) {
            resolve(args);
          }
        }
      });
    } else {
      resolve([]);
    }
  });
}

Promise.forEach = function(collection, callback) {
  return new Promise(function(resolve, reject) {
    var len = _.size(collection);

    if (len && _.isFunction(callback)) {
      var args = [];
      
      if (_.isArray(collection)) {
        collection.forEach(function(value, index) {
          var curIndex = index;
          var entry = new Promise(function(resolve, reject) {
            try {
              callback(resolve, reject, value, index);
            } catch(error) {
              reject(error); 
            }
          });
          
          entry.then(function(reply) {
            args[curIndex] = reply;
            len--;
            
            if (!len) {
              resolve(args);
            }
          }).catch(function(error) {
            reject(error);
          });
        });
      } else if (_.isObject(collection)) {
        var index = 0;
        
        _.forEach(collection, function(value, key) {
          var curIndex = index;
          var entry = new Promise(function(resolve, reject) {
            try {
              callback(resolve, reject, value, key);
            } catch(error) {
              reject(error); 
            }
          });
          
          entry.then(function(reply) {
            args[curIndex] = reply;
            len--;
            
            if (!len) {
              resolve(args);
            }
          }).catch(function(error) {
            reject(error);
          });
          
          index++;
        });
      } else {
        resolve([collection]);
      }
    } else {
      resolve([]);
    }
  });
}

var STREAM_OPENING = 1 << 1;
var STREAM_RUNNING = 1 << 2;
var STREAM_CLOSING = 1 << 3
var STREAM_CLOSED  = 1 << 4;

function Stream(options) {
  var self = this;
  
  if (!(self instanceof Stream)) {
    return new Stream(options);
  }
  
  options = options || {};
  
  Promise.call(this, function(resolve, reject) {
    self._resolve = resolve;
    self._reject = reject;
  });
  
  self.maxThresholdSize = options.maxThresholdSize || 16384;
  self.objectMode = options.objectMode || false;
  self.state = options.state || STREAM_RUNNING;
  self.threshold = 0;
  
  if (_.isFunction(options.write)) {
    self._write = options.write;
  }
}

Object.setPrototypeOf(Stream.prototype, Promise.prototype);

Object.defineProperty(Stream.prototype, 'readable', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.pending && this.ondata && this.state & (STREAM_OPENING | STREAM_RUNNING | STREAM_CLOSING)) ? true : false; },  
});

Object.defineProperty(Stream.prototype, 'writable', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.pending && this._write && this.state & (STREAM_OPENING | STREAM_RUNNING | STREAM_CLOSING)) ? true : false; },  
});

Object.defineProperty(Stream.prototype, 'isOpening', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.state & STREAM_OPENING) ? true : false; },  
});

Object.defineProperty(Stream.prototype, 'isRunning', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.state & STREAM_RUNNING) ? true : false; },   
});

Object.defineProperty(Stream.prototype, 'isClosing', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.state & STREAM_CLOSING) ? true : false; },   
});

Object.defineProperty(Stream.prototype, 'isClosed', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.state & STREAM_CLOSED) ? true : false; },    
});

Object.defineProperty(Stream, 'OPENING', {
  enumerable: false,
  configurable: false,
  writable: false,
  value: STREAM_OPENING,
});

Object.defineProperty(Stream, 'RUNNING', {
  enumerable: false,
  configurable: false,
  writable: false,
  value: STREAM_RUNNING,
});

Object.defineProperty(Stream, 'CLOSING', {
  enumerable: false,
  configurable: false,
  writable: false,
  value: STREAM_CLOSING,
});

Object.defineProperty(Stream, 'CLOSED', {
  enumerable: false,
  configurable: false,
  writable: false,
  value: STREAM_CLOSED,
});

Stream.prototype.destroy = function(error) {  
  this.heap = null;
  this.tail = null;
  this.threshold = 0;
  this.state = STREAM_CLOSED;
  
  this.onopen = null;
  this.onclose = null;
  this.ondata = null;
  this.ondrain = null;
  this.onpause = null;
  this.onresume = null;
  
  return Promise.prototype.destroy.call(this, error);
};

Stream.prototype.setState = function(state) {
  var self = this;
  
  if (self.pending) {
    if (state & STREAM_OPENING && self.state & STREAM_CLOSED) {
      self.state = STREAM_OPENING;
    } else if (state & STREAM_RUNNING && self.state & STREAM_OPENING) {
      self.state = STREAM_RUNNING;
      
      _.sll_forEach(self.onopen, function(callback) {
        try {
          callback.call(self);
        } catch(error) {
          self.end(error);
        }
      });
      
      self.onopen = null;
    } else if (state & STREAM_CLOSING && self.state & (STREAM_OPENING | STREAM_RUNNING)) {
      self.state = STREAM_CLOSING;
    } else if (state & STREAM_CLOSED && self.state & ~(STREAM_CLOSED)) {
      self.state = STREAM_CLOSED;
      
      _.sll_forEach(self.onclose, function(callback) {
        try {
          callback.call(self);
        } catch(error) {
          self.end(error);
        }
      });
      
      self.onclose = null;
    }
  }
  
  return self;
};

Stream.prototype.close = function(onClose) {
  var self = this;
  
  if (self.pending) {
    if (self.state & (STREAM_OPENING | STREAM_RUNNING | STREAM_CLOSING)) {
      if (_.isFunction(onClose)) {
        self.onclose = _.sll_add(self.onclose, onClose);
      }
    } else {
      if (_.isFunction(onClose)) {
        onClose.call(self);
      }
    }
  }
  
  return self;
};

Stream.prototype.open = function(onOpen) {
  var self = this;
  
  if (self.pending) {
    if (self.state & STREAM_OPENING) {
      if (_.isFunction(onOpen)) {
        self.onopen = _.sll_add(self.onopen, onOpen);
      }
    } else {
      if (self.state & (STREAM_RUNNING | STREAM_CLOSING) && _.isFunction(onOpen)) {
        onOpen.call(self);
      }
    }
  }
  
  return self;
};

Stream.prototype.data = function(onData) {
  var self = this;
  
  if (self.pending) {
    if (_.isFunction(onData)) {
      self.ondata = _.sll_add(self.ondata, onData);
    }
  }
  
  return self;
};

Stream.prototype.drain = function(onDrain) {
  var self = this;
  
  _.delay(function() {
    if (self.pending) {
      if (self.heap) {
        if (_.isFunction(onDrain)) {
          self.ondrain = _.sll_add(self.ondrain, onDrain);
        }
      } else {
        if (_.isFunction(onDrain)) {
          onDrain.call(self);
        }
      }
    }
  });
  
  return self;
};

Stream.prototype.resume = function(onResume) {
  var self = this;
  
  if (self.pending) {
    if (_.isFunction(onResume)) {
      self.onresume = _.sll_add(self.onresume, onResume);
    }
  }
  
  return self;
};

Stream.prototype.pause = function(onPause) {
  var self = this;
  
  if (self.pending) {
    if (_.isFunction(onPause)) {
      self.onpause = _.sll_add(self.onpause, onPause);
    }
  }
  
  return self;
};

Stream.prototype.end = function(arg) {
  var self = this;
  
  if (_.isError(arg)) {
    self._reject(arg);
  } else if (_.isPromise(arg)) {
    arg.then(function(reply) {
      self.end(reply);
    }, function(error) {
      self.end(error);
    });
  } else {
    if (self.state & (STREAM_RUNNING | STREAM_CLOSING)) {      
      self.drain(function() {
        self._resolve(arg);
      });
    } else if (self.state & STREAM_OPENING) {
      self.open(function() {
        self.drain(function() {
          self._resolve(arg);<commit_msg>Remove destroy() and set CLOSED state before resolve or reject.<commit_after>
/*
 * Stream-NG - Next generation of Stream interface for NodeJS
 *
 * Copyright (c) 2016 vmolsa <<EMAIL>> (http://github.com/vmolsa)
 * MIT License <https://github.com/vmolsa/stream-ng/blob/master/LICENSE>
 *
 */

var _ = {
  isArray: function(arg) {
    return (Object.prototype.toString.call(arg) === '[object Array]');
  },
  isObject: function(arg) {
    return (arg && typeof(arg) === 'object');
  },
  isString: function(arg) {
    return (typeof(arg) === 'string');
  },
  isUndefined: function(arg) {
    return (typeof(arg) === 'undefined');
  },
  isFunction: function(arg) {
    return (typeof(arg) === 'function');
  },
  isNumber: function(arg) {
    return (typeof(arg) === 'number');
  },
  isTypedArray: function(arg) {
    return ArrayBuffer.isView(arg);
  },
  isArrayBuffer: function(arg) {
    return (arg instanceof ArrayBuffer);
  },
  isError: function(arg) {
    return (arg instanceof Error);
  },
  isPromise: function(arg) {
    return (arg && typeof(arg.then) === 'function');
  },
  size: function(arg) {
    if (_.isObject(arg)) {
      return Object.keys(arg).length;
    } else if (_.isArray(arg)) {
      return arg.length;
    } else if (_.isString(arg)) {
      return arg.length;
    } else if (_.isTypedArray(arg) || _.isArrayBuffer(arg)) {
      return arg.byteLength;
    } else {
      return -1;
    }
  },
  now: function() {
    return new Date().getTime();
  },
  extend: function(dst, src) {
    Object.keys(src).forEach(function(key) {
      dst[key] = src[key];
    });
    
    return dst;
  },
  forEach: function(collection, callback) {
    if (_.isFunction(callback)) {
      if (_.isObject(collection)) {
        Object.keys(collection).forEach(function(key) {
          callback(collection[key], key);
        });
      } else if (_.isArray(collection)) {
        collection.forEach(callback);
      }
    }
    return collection;
  },
  once: function(callback, self) {
    return function(arg) {     
      if (_.isFunction(callback)) {
        callback.call(self || this, arg);
      }
      
      callback = undefined;
    };
  },
  delay: process.nextTick,
  random: function() {
    return (Math.random() * (-1 >>> 0)) >>> 0;
  },
  sll_add: function(collection, arg) {
    return {
      element: arg,
      next: collection,
    };
  },
  sll_forEach: function(collection, callback) {
    if (_.isFunction(callback)) {
      for (var index = collection; index; index = index.next) {
        callback(index.element);
      }
    }
    
    return collection;
  },
};

function Promise(callback) {
  var self = this;
 
  if (!(self instanceof Promise)) {
    return new Promise(callback);
  }
  
  self.pending = true;
  self.fulfilled = false;
  self.rejected = false;
  
  function onReject(arg) {
    _.delay(function() {    
      if (!self.fulfilled && !self.rejected) {
        self.pending = false;
        self.fulfilled = false;
        self.rejected = true;
        self.onresolve = null;
        
        _.sll_forEach(self.onreject, function(callback) {
          callback.call(self, arg);
        });
        
        self.onreject = null;
      }
    });
  }
  
  function onResolve(arg) {
    if (_.isPromise(arg)) {
      arg.then(function(reply) {
        onResolve(reply);
      }, function(error) {
        onReject(error);
      });
    } else {
      _.delay(function() {      
        if (!self.fulfilled && !self.rejected) {
          self.pending = false;
          self.fulfilled = true;
          self.rejected = false;
          self.onreject = null;
          
          _.sll_forEach(self.onresolve, function(callback) {
            try {
              callback.call(self, arg);
            } catch(error) {
              onReject(error);
            }
          });

          self.onresolve = null;
        }
      });
    }
  }
  
  if (_.isFunction(callback)) {    
    try {
      callback(onResolve, onReject);
    } catch(error) {
      onReject(error);
    }
  } else {
    throw new Error('Promise resolver ' + typeof(callback) + ' is not a function');
  }
}

Promise.prototype.then = function(onResolve, onReject) {  
  if (this.pending) {
    if (_.isFunction(onResolve)) {
      this.onresolve = _.sll_add(this.onresolve, onResolve);
    }
    
    if (_.isFunction(onReject)) {
      this.onreject = _.sll_add(this.onreject, onReject);
    }
  }
  
  return this;
};

Promise.prototype.catch = function(onReject) {  
  if (this.pending) {
    if (_.isFunction(onReject)) {
      this.onreject = _.sll_add(this.onreject, onReject);
    }
  }
  
  return this;
};

Promise.prototype.finally = function(onResolve) {  
  return this.then(function(arg) {
    onResolve(arg);
  }, function(error) {
    onResolve(error);
  });
};

Promise.all = function(collection) {
  return new Promise(function(resolve, reject) {
    var len = _.size(collection);
    
    if (_.isArray(collection) && len) {
      var args = [];
      
      collection.forEach(function(value, index) {
        if (_.isPromise(value)) {
          var curIndex = index;
          
          value.then(function(reply) {
            args[curIndex] = reply;
            len--;
            
            if (!len) {
              resolve(args);
            }
          }).catch(function(error) {
            reject(error);
          });
        } else {
          args[index] = value;
          len--;
            
          if (!len) {
            resolve(args);
          }
        }
      });
    } else {
      resolve([]);
    }
  });
}

Promise.forEach = function(collection, callback) {
  return new Promise(function(resolve, reject) {
    var len = _.size(collection);

    if (len && _.isFunction(callback)) {
      var args = [];
      
      if (_.isArray(collection)) {
        collection.forEach(function(value, index) {
          var curIndex = index;
          var entry = new Promise(function(resolve, reject) {
            try {
              callback(resolve, reject, value, index);
            } catch(error) {
              reject(error); 
            }
          });
          
          entry.then(function(reply) {
            args[curIndex] = reply;
            len--;
            
            if (!len) {
              resolve(args);
            }
          }).catch(function(error) {
            reject(error);
          });
        });
      } else if (_.isObject(collection)) {
        var index = 0;
        
        _.forEach(collection, function(value, key) {
          var curIndex = index;
          var entry = new Promise(function(resolve, reject) {
            try {
              callback(resolve, reject, value, key);
            } catch(error) {
              reject(error); 
            }
          });
          
          entry.then(function(reply) {
            args[curIndex] = reply;
            len--;
            
            if (!len) {
              resolve(args);
            }
          }).catch(function(error) {
            reject(error);
          });
          
          index++;
        });
      } else {
        resolve([collection]);
      }
    } else {
      resolve([]);
    }
  });
}

var STREAM_OPENING = 1 << 1;
var STREAM_RUNNING = 1 << 2;
var STREAM_CLOSING = 1 << 3
var STREAM_CLOSED  = 1 << 4;

function Stream(options) {
  var self = this;
  
  if (!(self instanceof Stream)) {
    return new Stream(options);
  }
  
  options = options || {};
  
  Promise.call(this, function(resolve, reject) {
    self._resolve = resolve;
    self._reject = reject;
  });
  
  self.maxThresholdSize = options.maxThresholdSize || 16384;
  self.objectMode = options.objectMode || false;
  self.state = options.state || STREAM_RUNNING;
  self.threshold = 0;
  
  if (_.isFunction(options.write)) {
    self._write = options.write;
  }
  
  self.finally(function() {
    self.heap = null;
    self.tail = null;
    self.threshold = 0;
    self.state = STREAM_CLOSED;
    
    self.onopen = null;
    self.onclose = null;
    self.ondata = null;
    self.ondrain = null;
    self.onpause = null;
    self.onresume = null;
  });
}

Object.setPrototypeOf(Stream.prototype, Promise.prototype);

Object.defineProperty(Stream.prototype, 'readable', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.pending && this.ondata && this.state & (STREAM_OPENING | STREAM_RUNNING | STREAM_CLOSING)) ? true : false; },  
});

Object.defineProperty(Stream.prototype, 'writable', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.pending && this._write && this.state & (STREAM_OPENING | STREAM_RUNNING | STREAM_CLOSING)) ? true : false; },  
});

Object.defineProperty(Stream.prototype, 'isOpening', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.state & STREAM_OPENING) ? true : false; },  
});

Object.defineProperty(Stream.prototype, 'isRunning', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.state & STREAM_RUNNING) ? true : false; },   
});

Object.defineProperty(Stream.prototype, 'isClosing', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.state & STREAM_CLOSING) ? true : false; },   
});

Object.defineProperty(Stream.prototype, 'isClosed', {
  enumerable: true,
  configurable: true,
  get: function() { return (this.state & STREAM_CLOSED) ? true : false; },    
});

Object.defineProperty(Stream, 'OPENING', {
  enumerable: false,
  configurable: false,
  writable: false,
  value: STREAM_OPENING,
});

Object.defineProperty(Stream, 'RUNNING', {
  enumerable: false,
  configurable: false,
  writable: false,
  value: STREAM_RUNNING,
});

Object.defineProperty(Stream, 'CLOSING', {
  enumerable: false,
  configurable: false,
  writable: false,
  value: STREAM_CLOSING,
});

Object.defineProperty(Stream, 'CLOSED', {
  enumerable: false,
  configurable: false,
  writable: false,
  value: STREAM_CLOSED,
});

Stream.prototype.setState = function(state) {
  var self = this;
  
  if (self.pending) {
    if (state & STREAM_OPENING && self.state & STREAM_CLOSED) {
      self.state = STREAM_OPENING;
    } else if (state & STREAM_RUNNING && self.state & STREAM_OPENING) {
      self.state = STREAM_RUNNING;
      
      _.sll_forEach(self.onopen, function(callback) {
        try {
          callback.call(self);
        } catch(error) {
          self.end(error);
        }
      });
      
      self.onopen = null;
    } else if (state & STREAM_CLOSING && self.state & (STREAM_OPENING | STREAM_RUNNING)) {
      self.state = STREAM_CLOSING;
    } else if (state & STREAM_CLOSED && self.state & ~(STREAM_CLOSED)) {
      self.state = STREAM_CLOSED;
      
      _.sll_forEach(self.onclose, function(callback) {
        try {
          callback.call(self);
        } catch(error) {
          self.end(error);
        }
      });
      
      self.onclose = null;
    }
  }
  
  return self;
};

Stream.prototype.close = function(onClose) {
  var self = this;
  
  if (self.pending) {
    if (self.state & (STREAM_OPENING | STREAM_RUNNING | STREAM_CLOSING)) {
      if (_.isFunction(onClose)) {
        self.onclose = _.sll_add(self.onclose, onClose);
      }
    } else {
      if (_.isFunction(onClose)) {
        onClose.call(self);
      }
    }
  }
  
  return self;
};

Stream.prototype.open = function(onOpen) {
  var self = this;
  
  if (self.pending) {
    if (self.state & STREAM_OPENING) {
      if (_.isFunction(onOpen)) {
        self.onopen = _.sll_add(self.onopen, onOpen);
      }
    } else {
      if (self.state & (STREAM_RUNNING | STREAM_CLOSING) && _.isFunction(onOpen)) {
        onOpen.call(self);
      }
    }
  }
  
  return self;
};

Stream.prototype.data = function(onData) {
  var self = this;
  
  if (self.pending) {
    if (_.isFunction(onData)) {
      self.ondata = _.sll_add(self.ondata, onData);
    }
  }
  
  return self;
};

Stream.prototype.drain = function(onDrain) {
  var self = this;
  
  _.delay(function() {
    if (self.pending) {
      if (self.heap) {
        if (_.isFunction(onDrain)) {
          self.ondrain = _.sll_add(self.ondrain, onDrain);
        }
      } else {
        if (_.isFunction(onDrain)) {
          onDrain.call(self);
        }
      }
    }
  });
  
  return self;
};

Stream.prototype.resume = function(onResume) {
  var self = this;
  
  if (self.pending) {
    if (_.isFunction(onResume)) {
      self.onresume = _.sll_add(self.onresume, onResume);
    }
  }
  
  return self;
};

Stream.prototype.pause = function(onPause) {
  var self = this;
  
  if (self.pending) {
    if (_.isFunction(onPause)) {
      self.onpause = _.sll_add(self.onpause, onPause);
    }
  }
  
  return self;
};

Stream.prototype.end = function(arg) {
  var self = this;
  
  if (_.isError(arg)) {
    self.setState(STREAM_CLOSED);
    self._reject(arg);
  } else if (_.isPromise(arg)) {
    arg.then(function(reply) {
      self.end(reply);
    }, function(error) {
      self.end(error);
    });
  } else {
    if (self.state & (STREAM_RUNNING | STREAM_CLOSING)) {      
      self.drain(function() {
        self.setState(STREAM_CLOSED);
        self._resolve(arg);
      });
    } else if (self.state & STREAM_OPENING) {
      self.open(function() {
        self.drain(function() {
          self.setState(STREAM_CLOSED);
          self._resolve(arg);
{'fraction_non_alphanumeric': 0.14434036790429974, 'fraction_numerical': 0.001805665274799684, 'mean_word_length': 2.1894421115776845, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 21, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3411668', 'n_tokens_mistral': 8659, 'n_tokens_neox': 8708, 'n_words': 1631}