<commit_before>var audio = require('./audio');
var inherits = require('util').inherits;
var Readable = require('stream').Readable;
var debug = require('debug')('osxaudio:input');

var MAX_BUFFERS = 16;

var input = null;

function Input(opts) {
  if (!(this instanceof Input)) {
    return new Input(opts);
  }
  Readable.call(this, opts);

  var self = this;
  if (!opts) opts = {};

  self._audioBuffers = [];

  if (!input) {
    input = new audio.input();
  }

  input.on('message', function(size, message) {
    if (self._audioBuffers.push(message) > MAX_BUFFERS) {
      debug('exceeded MAX_BUFFERS, shifting older buffers');
      self._audioBuffers.splice(0, self._audioBuffers.length - MAX_BUFFERS);
    }
    self.read(0);
  });

  if (!input.isOpen()) {
    debug('opening input');
    input.openInput();
  }

  debug("instantiated.");

  return self;
}
inherits(Input, Readable);

Input.prototype._read = function(n) {
  debug("_read()");
  var chunk = this._audioBuffers.pop();
  if (typeof chunk !== 'undefined') {
    debug("_read() pushing data");
    this.push(chunk);
  }
  else {
    debug("_read() pushing blank string");
    this.push('');
  }
};

Input.prototype.openInput = function() {
  if(!input) {
    input = new audio.input();
  }

  if (!input.isOpen()) {
    input.openInput();
  }
};

Input.prototype.closeInput = function() {
  if (input.isOpen()) {
    debug("closeInput() input was open, shutting down");
    input.closeInput();
    
    // push null to shut down the stream
    this.push(null);
  }
};

module.exports = Input;<commit_msg>separate out the audio callback, and clean up closing of stream<commit_after>var audio = require('./audio');
var inherits = require('util').inherits;
var Readable = require('stream').Readable;
var debug = require('debug')('osxaudio:input');

var MAX_BUFFERS = 16;

var input = null;

function Input(opts) {
  if (!(this instanceof Input)) {
    return new Input(opts);
  }
  Readable.call(this, opts);

  if (!opts) opts = {};

  this._audioBuffers = [];

  if (!input) {
    input = new audio.input();
  }

  input.on('message', this._audioCallback.bind(this));

  if (!input.isOpen()) {
    debug('opening input');
    input.openInput();
  }

  debug("instantiated.");

  return this;
}
inherits(Input, Readable);

Input.prototype._audioCallback = function(size, message) {
  if (this._audioBuffers.push(message) > MAX_BUFFERS) {
    debug('exceeded MAX_BUFFERS, shifting older buffers');
    this._audioBuffers.splice(0, this._audioBuffers.length - MAX_BUFFERS);
  }
  this.read(0);
};

Input.prototype._read = function(n) {
  debug("_read()");
  var chunk = this._audioBuffers.pop();
  if (typeof chunk !== 'undefined') {
    debug("_read() pushing data");
    this.push(chunk);
  }
  else {
    if (input.isOpen()) {
      debug("_read() pushing blank string");
      this.push('');
    }
    else {
      debug("_read() input is closed. push null");
      this.push(null);
    }
  }
};

Input.prototype.openInput = function() {
  if(!input) {
    input = new audio.input();
  }

  if (!input.isOpen()) {
    input.openInput();
  }
};

Input.prototype.closeInput = function() {
  if (input.isOpen()) {
    debug("closeInput() input was open, shutting down");
    input.closeInput();
  }
};

module.exports = Input;
{'fraction_non_alphanumeric': 0.17248709383540844, 'fraction_numerical': 0.0024293956878226543, 'mean_word_length': 3.4816326530612245, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, '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': '3041405', 'n_tokens_mistral': 1115, 'n_tokens_neox': 1085, 'n_words': 242}