better support for host argument, and handling of 'no es' scenarios

_client.js

@@ -1,18 +1,49 @@
 var argv = require('./argv');
-var es = require('elasticsearch');
+var _ = require('lodash');
+var Promise = require('bluebird');
+var through2 = require('through2');
+var parse = require('url').parse;
+var Client = require('elasticsearch').Client;
+var NoConnections = require('elasticsearch').errors.NoConnections;
+var RequestTimeout = require('elasticsearch').errors.RequestTimeout;
 
-var clientConfig = {};
+var host = String(argv.host);
+var proto = _.contains(host, '//') ? '' : '//';
+var parsed = parse(proto + host, false, true);
 
-if (argv.trace) {
-  clientConfig.log = 'trace';
-}
-
-clientConfig.hosts = [
-  {
-    host: argv.host.split(":")[0],
-    port: argv.host.split(":")[1] ? argv.host.split(":")[1] : "9200",
+var ms = 1000;
+var client = module.exports = new Client({
+  log: {
+    type: 'stream',
+    level: argv.trace ? 'trace' : 'warning',
+    stream: through2(function (chunk, enc, cb) {
+      client.usable.then(function () {
+        process.stdout.write(chunk, enc);
+        cb();
+      });
+    })
+  },
+  host: {
+    host: parsed.hostname,
+    port: parsed.port,
     auth: argv.auth
   }
-];
+});
+
+client.usable = client.ping({
+  requestTimeout: ms
+})
+.catch(function (err) {
+  var notAlive = err instanceof NoConnections;
+  var timeout = err instanceof RequestTimeout;
+
+  if (notAlive || timeout) {
+    console.error('Unable to connect to elasticsearch at %s within %d seconds', host, ms / 1000);
+  } else {
+    console.log('unknown ping error', err);
+  }
 
-module.exports = new es.Client(clientConfig);
+  client.close();
+  // prevent the promise from ever resolving or rejecting
+  return new Promise(_.noop);
+});

argv/index.js

@@ -19,7 +19,7 @@ var optimist = require('optimist')
     host: {
       alias: 'h',
       describe: 'The host name and port',
-      default: null
+      default: 'localhost:9200'
     },
     auth: {
       describe: 'user:password when you want to connect to a secured elasticsearch cluster over basic auth',

eventBuffer/_bulkQueue.js

@@ -11,7 +11,7 @@ module.exports = function (eventBuffer) {
 
   var queue = async.queue(function (events, done) {
     var body = [];
-    var deps = [];
+    var deps = [client.usable];
     var esBulkQueueOverflow = 0;
 
     events.forEach(function (event) {

eventBuffer/_createIndex.js

@@ -120,10 +120,13 @@ module.exports = function createIndex(indexName) {
     }
   };
 
-  return client.indices.create({
-    ignore: 400,
-    index: indexName,
-    body: indexBody
+  return client.usable
+  .then(function () {
+    return client.indices.create({
+      ignore: 400,
+      index: indexName,
+      body: indexBody
+    });
   })
   .then(function () {
     return client.cluster.health({

index.js

@@ -15,9 +15,10 @@ var total = argv.total;
 var startingMoment = argv.start;
 var endingMoment = argv.end;
 
-console.log('Generating', total, 'events from', startingMoment.format(), 'to', endingMoment.format());
-
-Promise.resolve()
+client.usable
+.then(function () {
+  console.log('Generating', total, 'events from', startingMoment.format(), 'to', endingMoment.format());
+})
 .then(function () {
   if (argv.dry) return;
   if (argv.reset) {

package.json

@@ -34,6 +34,7 @@
     "lodash": "~2.4.1",
     "moment": "~2.7.0",
     "optimist": "~0.6.1",
+    "through2": "~0.6.3",
     "update-notifier": "~0.2.1"
   }
 }