pelias-dbclient

'use strict'; const streamify = require('stream-array'); const through = require('through2'); const Document = require('pelias-model').Document; const dbMapper = require('pelias-model').createDocumentMapperStream; const dbclient = require('pelias-dbclient'); const elasticsearch = require('elasticsearch'); const config = require('pelias-config').generate(); // Ensure pelias-config is properly set up const elasticDeleteQuery = require('elastic-deletebyquery'); const timestamp = Date.now(); // Simulate an upstream data source const stream = streamify([1, 2, 3, 4, 5]) .pipe(through.obj((item, enc, next) => { // Create a Pelias Document for each item const uniqueId = [ 'docType', item ].join(':'); const doc = new Document( 'sourceType', 'venue', uniqueId ); doc.timestamp = timestamp; doc.setName('default', `Test Venue ${item}`); doc.setCentroid(item * 0.1, item * 0.2); next(null, doc); })) .pipe(dbMapper()) // Map Pelias Document to Elasticsearch format .pipe(dbclient()); // Bulk-insert documents into Elasticsearch stream.on('finish', () => { console.log('All documents processed and sent to Elasticsearch.'); const client = new elasticsearch.Client(config.esclient); // Example of a post-import operation: clean up old documents const options = { index: config.schema.indexName, body: { query: { "bool": { "must": [ {"term": { "source": "sourceType" }} ], "must_not": [ {"term": { "timestamp": timestamp }} ] } } } }; client.deleteByQuery(options, (err, response) => { if (err) { console.error('Error during cleanup:', err); } else { console.log(`Cleaned up ${response.elements || response.deleted} old elements.`); } client.close(); }); }); stream.on('error', (err) => { console.error('Stream encountered an error:', err); });