var _ = require('lodash');
var optimist = require('optimist');

var argv = optimist
  .alias('s', 'services')
  .describe('services', 'Comma separated list of services to start. Default: run all services')
  .argv;

// This export must return an object (or class) with start/stop functions.
// When mixdown calls start, mixdown.apps is already initialized.  This is true for start and reload scenarios.
module.exports = function(mixdown, options) {
  this.mixdown = mixdown;
  this.options = options;
  var self = this;
  var writeCount = {};
  var lastSaveTime = {};
  var lastSaveCount = {};
  var appsRunning = 0;
  this.streams = {};

  // Start doing stuff.  This should start service or resume the service if stop was previously called.
  this.start = function(done) {

    runnableApps = mixdown.apps;

    // if the services flag was passed from cmd line, then only run the specified services.
    if (argv.services) {
      logger.info('Filtering services');
      runnableApps = _.pick(runnableApps, argv.services.trim().split(/\s*,\s*/g));
    }
    logger.info('Apps starting', Object.keys(runnableApps));
    logger.info(runnableApps);

    _.each(runnableApps, function(app, appid) {
      try {
        writeCount[appid] = 0;
        lastSaveCount[appid] = 0;
        lastSaveTime[appid] = new Date();
        appsRunning++;

        // logger.debug(app.transformer.source_stream);
        var input = app.source.create_stream(app);
        var transform = app.transform.create_stream(app);
        var output = app.output.create_stream(app);

        // attach this state to the delegate so that outside modules can access the latest streams.  
        // This was added for unit test support.
        self.streams[appid] = {
          input: input,
          transform: transform,
          output: output
        };

        input.on('warn', function(err, chunk) {
          logger.info('Input Warn', err, err.stack, JSON.stringify(chunk));
        });

        transform.on('warn', function(err, chunk) {
          logger.info('Transform Warn', err, err.stack);
        });

        output.on('error', function(err) {
          logger.error('Output Error', err, err.stack);
        });

        output.on('finish', function() {
          if (!isNaN(output.push_count)) {
            logger.info(output.push_count, 'outputs');
            logger.info(output.error_count, 'errors');
          }
          appsRunning--;
          logger.info(appid + ' finished running');

          if (appsRunning === 0) {
            self.stop();
          }
        });


        output.on('save', function(obj) {
          var dt = lastSaveTime[appid];
          var dtNow = new Date();
          var secs = (Math.abs(dtNow - dt) / 1000);

          // logger.info('save ' + obj.id);
          logger.debug(obj);
          writeCount[appid] += _.isArray(obj) ? obj.length : 1;

          if (secs > 1) {
            var diff = writeCount[appid] - lastSaveCount[appid];

            // update last log counters
            lastSaveTime[appid] = dtNow;
            lastSaveCount[appid] = writeCount[appid];

            // Send to logs.
            logger.info(appid, diff, secs + 's', (diff / secs).toFixed(1) + ' rec/sec.');
            // logger.debug('write doc', obj);
          }

        });

        input.pipe(transform).pipe(output);
      } catch (ex) {
        logger.error(ex);
        logger.error(ex.stack)
      }
    });

    done();
  };

  // stop running the app.  stop() is responsible for pausing processing, saving state, or whatever so that it can pick up where it left off.
  // This is called on shutdown and also called on reload from config.
  this.stop = function(done) {
    this.mixdown.emit('stop');
    return typeof(done) === 'function' ? done() : null;
  };
};
@@ -1122,20 +1122,21 @@
  logger.
-info
+debug
 (runnabl

drop log level for runnable apps