forked from ejjoo/node-elasticsearch-broker
-
Notifications
You must be signed in to change notification settings - Fork 0
/
example_stream.js
65 lines (56 loc) · 1.25 KB
/
example_stream.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
var es_broker = require('./index'),
fs = require('fs'),
eventstream = require('event-stream');
var argv = require('optimist')
.boolean('v')
.alias('t', 'target')
.alias('l', 'logpath')
.alias('v', 'verbose')
.describe('t', 'elasticsearch url')
.describe('l', 'log file path')
.describe('v', 'verbose')
.demand(['l'])
.argv;
var logpath = argv.l;
var estarget = argv.t;
var verbose = argv.v;
if (!fs.existsSync(logpath)) {
throw new Error("log file is not exist");
}
var frs = fs.createReadStream(logpath);
if (frs == null) {
throw new Error("failed to load file");
}
var parser = function(str, callback) {
if (str == null) {
callback('parse queue received null', null);
return;
}
var data;
try {
data = JSON.parse(str);
} catch(e) {
callback(e, null);
return;
}
data.timestamp = data.timestamp || Date.now();
callback(null, data);
};
var broker = es_broker.create({
parser: parser,
target: estarget,
max_bulk_qtty: 1000,
max_request_num: 10,
verbose: true,
verbose_data: true,
index: 'index'
});
frs.pipe(eventstream.split())
.pipe(eventstream.mapSync(function(line) {
broker.push(line);
}).on('end', function() {
broker.close(function(total_task) {
console.log(total_task + ' has done.');
});
})
);