histograph-elasticsearch.js

delete-index.js

var elasticsearch = require('elasticsearch'),
    config = require(process.env.HISTOGRAPH_CONFIG),
    esClient = new elasticsearch.Client({
      host: config.elasticsearch.host + ':' + config.elasticsearch.port,
      //log: 'trace'
    });

esClient.indices.delete({
  index: config.elasticsearch.index
}, function(body) {
  console.log("Done...");
  esClient.close();
});

histograph-elasticsearch.js

var elasticsearch = require('elasticsearch'),
    redis = require("redis"),
    async = require("async"),
    config = require(process.env.HISTOGRAPH_CONFIG),
    type = "pit",
    queue = config.redis.queues.elasticsearch,
    mapping = require(config.schemas.dir + "elasticsearch/" + type + ".json"),
    esClient = new elasticsearch.Client({
      host: config.elasticsearch.host + ':' + config.elasticsearch.port,
      //log: 'trace'
    }),
    redisClient = redis.createClient(config.redis.port, config.redis.host);

function initializeIndex(callback) {
  esClient.indices.exists({
    index: config.elasticsearch.index
  }).then(function(exists) {
    if (exists) {
      callback();
    } else {
      esClient.indices.create({
        index: config.elasticsearch.index
      }).then(function (body) {        
        esClient.indices.putMapping({
          index: config.elasticsearch.index,
          type: type,
          body: mapping
        }, function(body) {
          callback();
        });
      });
    }    
  });  
}

initializeIndex(function() {
  redisClient.llen(queue, function (err, length) {
    
    var index = 0;
    async.whilst(
      function () { return index < length; },
      function (callback) {
        getFromQueue(index, function(i, data) {

          var item = JSON.parse(data);

          if (item.type == type) {

            // TODO check item.action (now always 'add')
            var pit = {
              hgid: item.layer + "/" + item.data.id,
              name: item.data.name,
              type: type,
              layer: item.layer,
              uri: item.data.uri,
              geometry: item.data.geometry,
              startDate: item.data.startDate,
              endDate: item.data.endDate              
            }

            esClient.index({
                index: config.elasticsearch.index,
                type: type,
                id: pit.hgid,
                body: pit
              },
              function(body) {
                console.log("" + i + "/" + length + ": indexed \"" + pit.hgid + "\"");
                index += 1;
                callback();
              });
          } else {
            index += 1;
            callback();
          }

        });

      },
      function (err) {
      }
    );
    

    // for (var i = 0; i < length; i++) {

    // }
  });
});

function getFromQueue(index, callback) {
  redisClient.lindex(queue, index, function(err, data) {
    callback(index, data);
  });
}

// TODO:
// client.bulk({
//   body: [
//     // action description
//     { index:  { _index: 'myindex', _type: 'mytype', _id: 1 } },
//      // the document to index
//     { title: 'foo' },
//     // action description
//     { update: { _index: 'myindex', _type: 'mytype', _id: 2 } },
//     // the document to update
//     { doc: { title: 'foo' } },
//     // action description
//     { delete: { _index: 'myindex', _type: 'mytype', _id: 3 } },
//     // no document needed for this delete
//   ]
// }, function (err, resp) {
//   // ...
// });