<commit_before>var Promise = require('native-promise-only');

var Client = require('pigato').Client;
var Broker = require('pigato').Broker;
var Worker = require('pigato').Worker;

module.exports = function(servicifyOptions) {
  function listen(opts) {
    var brokerAddress = 'tcp://' + opts.host + ':' + opts.port;

    return start(new Broker(brokerAddress)).then(function (broker) {
      broker.on('error', function(err) {
        console.error(err);
      });

      return {
        stop: stop.bind(null, broker)
      };
    });
  }

  function offer(spec, invoke, opts) {
    var worker = new Worker('tcp://' + opts.host + ':' + opts.port, spec.name + '@' + spec.version);

    worker.on('error', function(err) {
      console.error(err);
    });

    worker.on('request', function (inp, rep, opts) {
      invoke(inp.args, opts).then(function (result) {
        rep.end(result);
      }, function (err) {
        rep.error(err.message);
      });
    });

    return start(worker).then(function () {
      return {
        invoke: function () {
          return invoke([].slice.call(arguments));
        },
        stop: function () {
          return stop(worker);
        }
      };
    });
  }

  function request(spec, args, requestOptions) {
    var opts = Object.assign({}, servicifyOptions, requestOptions, {
      timeout: 10000
    });

    return new Promise(function (resolve, reject) {
      var client = new Client('tcp://' + opts.host + ':' + opts.port);
      client.once('start', resolve.bind(null, client));
      client.once('error', function(err) {
        reject(err);
      });
      client.start();
    }).then(function(client) {
      return new Promise(function (resolve, reject) {
        client.request(spec.name + '@' + spec.version, {args: args}, false, function (err, response) {
          stop(client).then(function () {
            return err ? reject(err) : resolve(response);
          });
        }, {timeout: opts.timeout});
      });
    });
  }

  return {
    listen: listen,
    offer: offer,
    request: request,
    name: 'pigato'
  };
};

function start(startable) {
  return new Promise(function (resolve) {
    startable.once('start', resolve.bind(null, startable));
    startable.start();
  });
}

function stop(stoppable) {
  return new Promise(function (resolve) {
    stoppable.once('stop',  resolve.bind(null, stoppable));
    stoppable.stop();
  });
}<commit_msg>simplifying the output from offerng and listening<commit_after>var Promise = require('native-promise-only');

var Client = require('pigato').Client;
var Broker = require('pigato').Broker;
var Worker = require('pigato').Worker;

module.exports = function(servicifyOptions) {
  function listen(opts) {
    var brokerAddress = 'tcp://' + opts.host + ':' + opts.port;

    return start(new Broker(brokerAddress)).then(function (broker) {
      broker.on('error', function(err) {
        console.error(err);
      });

      return stop.bind(null, broker);
    });
  }

  function offer(spec, invoke, opts) {
    var worker = new Worker('tcp://' + opts.host + ':' + opts.port, spec.name + '@' + spec.version);

    worker.on('error', function(err) {
      console.error(err);
    });

    worker.on('request', function (inp, rep, opts) {
      invoke(inp.args, opts).then(function (result) {
        rep.end(result);
      }, function (err) {
        rep.error(err.message);
      });
    });

    return start(worker).then(function () {
      return stop.bind(null, worker);
    });
  }

  function request(spec, args, requestOptions) {
    var opts = Object.assign({}, servicifyOptions, requestOptions, {
      timeout: 10000
    });

    return new Promise(function (resolve, reject) {
      var client = new Client('tcp://' + opts.host + ':' + opts.port);
      client.once('start', resolve.bind(null, client));
      client.once('error', function(err) {
        reject(err);
      });
      client.start();
    }).then(function(client) {
      return new Promise(function (resolve, reject) {
        client.request(spec.name + '@' + spec.version, {args: args}, false, function (err, response) {
          stop(client).then(function () {
            return err ? reject(err) : resolve(response);
          });
        }, {timeout: opts.timeout});
      });
    });
  }

  return {
    listen: listen,
    offer: offer,
    request: request,
    name: 'pigato'
  };
};

function start(startable) {
  return new Promise(function (resolve) {
    startable.once('start', resolve.bind(null, startable));
    startable.start();
  });
}

function stop(stoppable) {
  return new Promise(function (resolve) {
    stoppable.once('stop',  resolve.bind(null, stoppable));
    stoppable.stop();
  });
}
{'fraction_non_alphanumeric': 0.16896624920399067, 'fraction_numerical': 0.0021226915729144554, 'mean_word_length': 2.7847389558232933, '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': '4767196', 'n_tokens_mistral': 1473, 'n_tokens_neox': 1442, 'n_words': 297}