ductile/lib/reader.js

// Generated by CoffeeScript 1.11.1
(function() {
  var ReadableSearch, jsonStream, mixin, through2, toBulk, transform;

  ReadableSearch = require('./readable-search');

  through2 = require('through2');

  mixin = require('./mixin');

  toBulk = function(operdelete) {
    return through2.obj(function(doc, enc, callback) {
      var idx;
      idx = {
        _index: doc._index,
        _type: doc._type,
        _id: doc._id
      };
      if (operdelete) {
        this.push({
          "delete": idx
        });
      } else {
        this.push({
          index: idx
        });
        this.push(doc._source);
      }
      return callback();
    });
  };

  transform = function(fn) {
    return through2.obj(function(doc, enc, callback) {
      var tdoc;
      tdoc = fn(doc);
      if (tdoc) {
        this.push(tdoc);
      }
      return callback();
    });
  };

  jsonStream = function() {
    return through2.obj(function(chunk, enc, callback) {
      this.push(JSON.stringify(chunk) + "\n");
      return callback();
    });
  };

  module.exports = function(client, _opts, operdelete, trans) {
    var last, opts, readable, scrollExec, stream;
    opts = mixin(_opts, {
      scroll: '60s',
      size: 200
    });
    if (!opts.body && !opts.q) {
      opts.body = {
        query: {
          match_all: {}
        }
      };
    }
    if (opts.body) {
      delete opts.q;
    } else {
      delete opts.body;
    }
    scrollExec = (function() {
      var scrollId;
      scrollId = null;
      return function(from, callback) {
        if (scrollId) {
          return client.scroll({
            scrollId: scrollId,
            scroll: '60s'
          }, callback);
        } else {
          return client.search(opts, function(err, res) {
            scrollId = res != null ? res._scroll_id : void 0;
            return callback(err, res);
          });
        }
      };
    })();
    readable = new ReadableSearch(scrollExec).on('error', function(err) {
      return stream.emit('error', err);
    });
    last = -1;
    stream = readable.pipe(through2.obj(function(hit, enc, callback) {
      this.push(hit);
      if (readable.from !== last) {
        last = readable.from;
        stream.emit('progress', {
          from: last,
          total: readable.total
        });
      }
      return callback();
    })).pipe(transform(trans)).pipe(toBulk(operdelete)).pipe(jsonStream()).on('end', function() {
      if (readable.from !== last) {
        return stream.emit('progress', {
          from: readable.total,
          total: readable.total
        });
      }
    });
    return stream;
  };

}).call(this);

//# sourceMappingURL=reader.js.map