<commit_before>var stream = require('stream'),
  async = require('async'),
  protocol = require('usb-daemon-parser'),
  util = require('util'),
  EventEmitter = require('events').EventEmitter
  ;

// The largest packet we can send over USB
var MAX_PACKET_SIZE = 255;
// The size of the back pressure buffers
var MAX_BUFFER_SIZE = 4096;

// A process is running remotely on the Tessel and data is piped in over USB
function USBProcess(id, daemon) {
  var self = this;
  // The numerical identifer for this process
  this.id = id;
  // Status of whether this process is killed or active
  this.active = true;
  // 4 remote process streams
  this.control = new RemoteWritableStream(id, daemon, protocol.controlWrite, protocol.controlClose);
  this.stdin = new RemoteWritableStream(id, daemon, protocol.stdinWrite, protocol.stdinClose);
  this.stdout = new RemoteReadableStream(id, daemon, protocol.stdoutAck);
  this.stderr = new RemoteReadableStream(id, daemon, protocol.stderrAck);

  // Once the process is killed
  self.once('death', function(data) {
    // Close the input streams because the remote proc can't handle data
    self.control.end();
    self.stdin.end();

    // Mark this process as inactive
    self.active = false;
  });

  // Tell the USBDaemon to close the remote process
  self.close = function() {
    // Write the close command
    daemon.write(protocol.closeProcess(self.id));
  };

  // Send a signal to the remote process
  self.kill = function(signal) {
    // Write the signal command
    daemon.write(protocol.killProcess(self.id, signal));
  }

  // Checks if this process is ready to be closed
  // and closes it if it can
  self.closeProcessIfReady = function() {
    // If this process hasn't been closed before
    // and all of the streams are closed
    if (self.active && self.stdout.closed
            && self.stderr.closed
            && self.stdin.closed
            && self.control.closed) {
      // Close this process
      self.close();
    }
  };

  // When any of the streams complete, check if the process is ready to close
  self.stdin.once('finish', self.closeProcessIfReady);
  self.stdout.once('end', self.closeProcessIfReady);
  self.stderr.once('end', self.closeProcessIfReady);

  // This function is primarily intended for compatibility with the ssh2 stream API
  self.stdout.signal = function(signalToSend) {
    switch(signalToSend) {
      case 'KILL':
        self.kill(9);
        break;
    }
  }

  // Tell the remote daemon that this process was created
  daemon.write(protocol.newProcess(self.id));
  // Tell the remote daemon that we can start receiving outbound data
  self.stdout.ack(MAX_BUFFER_SIZE);
  self.stderr.ack(MAX_BUFFER_SIZE);

}

util.inherits(USBProcess, EventEmitter);

// A wrapper on the Node Writable Stream to encapsulate stdin & control stream functionality
function RemoteWritableStream(id, daemon, writeHeaderFunc, closeHeaderFunc) {
  var self = this;
  // Inherit from Writable Streams
  stream.Writable.call(self);
  // The id of the process that this is a stream of
  self.id = id;
  // The daemon to write data to
  self.daemon = daemon;
  // The amount of credit allocated to the stream (backpressure)
  self.credit = 0;
  // An array of backpressure entries
  self.backPressure = new Array(0);<commit_msg>Cleans up process closing logic and adds support for sigint signals<commit_after>var stream = require('stream'),
  async = require('async'),
  protocol = require('usb-daemon-parser'),
  util = require('util'),
  EventEmitter = require('events').EventEmitter
  ;

// The largest packet we can send over USB
var MAX_PACKET_SIZE = 255;
// The size of the back pressure buffers
var MAX_BUFFER_SIZE = 4096;

// A process is running remotely on the Tessel and data is piped in over USB
function USBProcess(id, daemon) {
  var self = this;
  // The numerical identifer for this process
  this.id = id;
  // Status of whether this process is killed or active
  this.active = true;
  // 4 remote process streams
  this.control = new RemoteWritableStream(id, daemon, protocol.controlWrite, protocol.controlClose);
  this.stdin = new RemoteWritableStream(id, daemon, protocol.stdinWrite, protocol.stdinClose);
  this.stdout = new RemoteReadableStream(id, daemon, protocol.stdoutAck);
  this.stderr = new RemoteReadableStream(id, daemon, protocol.stderrAck);

  // Once the process is killed
  self.once('death', function(data) {
    // Close the input streams because the remote proc can't handle data
    self.control.end();
    self.stdin.end();

    // Mark this process as inactive
    self.active = false;

    // Close the process
    self.closeProcessIfReady();
  });

  // Tell the USBDaemon to close the remote process
  self.close = function() {
    // Write the close command
    daemon.write(protocol.closeProcess(self.id));
  };

  // Send a signal to the remote process
  self.kill = function(signal) {
    // Write the signal command
    daemon.write(protocol.killProcess(self.id, signal));
  }

  // Checks if this process is ready to be closed
  // and closes it if it can
  self.closeProcessIfReady = function() {
    // If this process hasn't been closed before
    // and all of the streams are closed
    if (!self.active && self.stdout.closed && self.stderr.closed) {
      // Close this process
      self.close();
    }
  };

  // When any of the streams complete, check if the process is ready to close
  self.stdout.once('end', self.closeProcessIfReady);
  self.stderr.once('end', self.closeProcessIfReady);

  // This function is primarily intended for compatibility with the ssh2 stream API
  self.stdout.signal = function(signalToSend) {
    switch(signalToSend) {
      case 'KILL':
        self.kill(9);
        break;
      case 'SIGINT':
        self.kill(2);
    }
  }

  // Tell the remote daemon that this process was created
  daemon.write(protocol.newProcess(self.id));
  // Tell the remote daemon that we can start receiving outbound data
  self.stdout.ack(MAX_BUFFER_SIZE);
  self.stderr.ack(MAX_BUFFER_SIZE);
}

util.inherits(USBProcess, EventEmitter);

// A wrapper on the Node Writable Stream to encapsulate stdin & control stream functionality
function RemoteWritableStream(id, daemon, writeHeaderFunc, closeHeaderFunc) {
  var self = this;
  // Inherit from Writable Streams
  stream.Writable.call(self);
  // The id of the process that this is a stream of
  self.id = id;
  // The daemon to write data to
  self.daemon = daemon;
  // The amount of credit allocated to the stream (backpressure)
  self.credit = 0;
  // An array of backpressure entries
  self.backPressure = new Array(0);
{'fraction_non_alphanumeric': 0.100120809423135, 'fraction_numerical': 0.0037752944729688917, 'mean_word_length': 3.3344240837696337, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '4500504', 'n_tokens_mistral': 1962, 'n_tokens_neox': 1910, 'n_words': 664}