var util = require('util');

var h = require('highland');
var elasticsearch = require('elasticsearch');

module.exports = function dest (glob, opt) {
  glob = glob || {};
  opt = opt || {};

  /**
   * Return a pipeline for use in non-Highland situations:
   */

  return h.pipeline(function(stream) {
    var _stream = (opt.rateLimit) ? stream.ratelimit(1, opt.rateLimit) : stream;
    var client = new elasticsearch.Client(opt);

    return _stream
      /**
       * Ensure there is some data to index:
       */

      .filter(function(file) {
        return file.data || file.contents;
      })

      /**
       * Change the layout and use default values where necessary:
       */

      .map(function(file) {
        return {
          index: glob.index || opt.index,
          type: glob.type || opt.type || (file.data && file.data.type) ||
            file.base,
          id: file.id || file.path || glob.id || opt.id,
          body: file.data || JSON.parse(file.contents.toString())
        };
      })

      /**
       * Now push each new object to ES:
       */

      .map(function(file) {
        client.index(file, function(error, response) {
          if (error) {
            throw new Error(util.format('Error writing to ES: %s %s %s', file,
              error, response));
          }
        });
        return file;
      });
  });
};

@@ -744,20 +744,19 @@
  index: 
-glob
+opt
 .index %7C
@@ -757,19 +757,20 @@
 ndex %7C%7C 
-opt
+glob
 .index,%0A
@@ -788,21 +788,8 @@
 ype:
- glob.type %7C%7C
  opt
@@ -841,16 +841,29 @@
         
+ glob.type %7C%7C
  file.ba
@@ -879,16 +879,26 @@
      id:
+ opt.id %7C%7C
  file.id
@@ -925,18 +925,8 @@
 b.id
- %7C%7C opt.id
 ,%0A  

reorder priorities for index, type and id