Postgres streams to elastic streams

[matt_baker]

Hi,

Hi,
i wanted to append below object to every object in stream

{"index":{"_index":"tvseries","_type":"internindex"}}
my stream looks like this

[
  {"showname":"The X Files","episode":"04","content":"Before what?","season":"1"},
  {"showname":"The X Files","episode":"04","content":"Before what?","season":"1"},
{"showname":"The X Files","episode":"01","content":"What?","season":"1"}
]

what my stream should look like !

-> POST http://localhost:9200/_bulk
  {"index":{"_index":"tvseries","_type":"internindex"}}
  {"showname":"The X Files","episode":"04","content":"Before what?","season":"1"}
  {"index":{"_index":"tvseries","_type":"internindex"}}
  {"showname":"The X Files","episode":"04","content":"Great.","season":"1"}
  {"index":{"_index":"tvseries","_type":"internindex"}}
  {"showname":"The X Files","episode":"01","content":"What?","season":"1"}

how can i achieve this using jsonstream in my existing belowcodebase

    var stream = new ElasticsearchWritableStream(client, {
      highWaterMark: 256,
      flushTimeout: 500
    });
  
pg.connect(connectionString,function(err, client, done) {
  if(err) throw err;
  var query = new QueryStream('SELECT * FROM srt limit 2')
  var streams = client.query(query)
  
  //release the client when the stream is finished
  streams.on('end', done)
  streams.pipe(JSONStream.stringify()).pipe(stream)
}

any other npm package which converts pgstreams directly to elastic search streams with some basic input params !

the npm packages currently i am using

1. For bulk insertion in elasticsearch !
   elasticsearch-writable-stream
2. for getting data from postgres into streams !
   pg-query-stream
3. missing piece is converting postgres streams into elastic writable streams !

any suggestion and recommendations on how to resolve this