wasm defintion for indexing record values to Elastic Search server

elastic-simple.js

const {
  SimpleTransform,
  PolicyError,
} = require("@vectorizedio/wasm-api");

const { Client } = require('@elastic/elasticsearch')

const client = new Client({node: 'http://localhost:9200'})

const transform = new SimpleTransform();
/* Topics that fire the transform function */
transform.subscribe(["produce"]);
/* The strategy the transform engine will use when handling errors */
transform.errorHandler(PolicyError.SkipOnFailure);
/* Auxiliar transform function for records */

/* Transform function */
transform.processRecord((recordBatch) => {
  const result = new Map();

  return Promise.all(
    recordBatch.records.map((record) => {
      console.log("applying...")
      if (
        Buffer.from("save").equals(record.headers[0].headerKey) ||
        Buffer.from("elastic").equals(record.headers[0].value)
      ) {
        return client.index({
          index: "result_wasm",
          type: "sometype",
          body: {
            record_message: record.value.toString()
          }
        })
      }
    })
  ).then(() => {
    result.set("elastic", recordBatch)
    return result
  }).catch((e) => console.log(e))
});

exports["default"] = transform;