1 | through2 = require 'through2'
|
2 | mixin = require './mixin'
|
3 |
|
4 |
|
5 | toBulk = -> through2.obj (doc, enc, callback) ->
|
6 | this.push alias:doc
|
7 | callback()
|
8 |
|
9 |
|
10 | jsonStream = -> through2.obj (chunk, enc, callback) ->
|
11 | this.push(JSON.stringify(chunk) + "\n")
|
12 | callback()
|
13 |
|
14 |
|
15 | module.exports = (client, _opts) ->
|
16 |
|
17 | opts = mixin _opts
|
18 |
|
19 | instream = toBulk()
|
20 | sink = instream.write.bind instream
|
21 |
|
22 | exec = ->
|
23 | client.indices.getAlias(opts).then (v) ->
|
24 | index for index, {aliases} of v
|
25 | .then (indices) ->
|
26 | Promise.all indices.map (index) -> client.indices.getAlias {index}
|
27 | .then (vs) ->
|
28 | col = {}
|
29 | vs.map (v) -> for i, {aliases} of v
|
30 | for n of aliases
|
31 | (col[n] = (col[n] ? [])).push i
|
32 | {_name, _index:(if i.length == 1 then i[0] else i)} for _name, i of col
|
33 |
|
34 | exec().then (docs) ->
|
35 | docs.forEach sink
|
36 | .catch (err) ->
|
37 | stream.emit 'error', err
|
38 |
|
39 | stream = instream.pipe jsonStream()
|