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

module.exports = function dest (glob, opt) {
  var stream = h();
  var _stream = stream;
  opt = opt || {};

  var client = new elasticsearch.Client(opt);

  if (opt.rateLimit) {
    _stream = stream.ratelimit(1, opt.rateLimit);
  }

  _stream
    .consume(function(error, file, push, next) {
      if (error) {
        push(error);
        next();
      } else if (file === h.nil) {
        push(null, file);
      } else {
        if (!(file.data || file.contents)) {
          console.error('No content to index!', file);
          push(null, file);
          next();
        } else {
          var req = {
            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())
          };

          client.index(req, function(error, response) {
            if (error) {
              console.error('Error writing to ES:', file.data, error);
              push(error);
            } else {
              push(null, file);
            }
            next();
          });
        }
      }
    })

    .resume()
    ;

  return stream;
};


var util = require('util');

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

module.exports = function dest (glob, opt) {
  var stream = h();
  var _stream = stream;
  opt = opt || {};

  var client = new elasticsearch.Client(opt);

  if (opt.rateLimit) {
    _stream = stream.ratelimit(1, opt.rateLimit);
  }

  _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;
    })

    /**
     * Doesn't feel right to have to do this, but no other way to get things
     * going:
     */

    .resume()
    ;

  return stream;
};

split functionality into more steps