gRPC + SSL with Avro encoding in node.js

grpc+avro.js

const grpc = require('grpc')
const fs = require('fs')
const avro = require('avsc')
const hexdump = require('hexdump-nodejs')

const encodeMessage = (avroType, request) => {
  let opts = {
    codec: null,
    writeHeader: true
  }
  return new Promise((resolve, reject) => {
    let buffers = []
    let encoder = new avro.streams.BlockEncoder(avroType, opts)
      .on('error', reject)
      .on('data', (value) => buffers.push(value))
      .on('end', () => resolve(Buffer.concat(buffers)))
    encoder.write(request)
    encoder.end()
  })
}

const decodeMessage = (data) => {
  return new Promise((resolve, reject) => {
    let opts = {}
    let avroSchema
    let events = []
    let decoder = new avro.streams.BlockDecoder(opts)
      .on('error', reject)
      .on('metadata', (type) => avroSchema = type)
      .on('data', (value) => events.push(value))
      .on('end', () => resolve({ avroSchema: avroSchema, data: events }))
    decoder.write(data)
    decoder.end()
  })
}

const sendGrpcMessage = async (client, requestType, requestBody, path) => {
  const encodedMessage = await encodeMessage(requestType, {})
  const encoder = () => encodedMessage
  const decoder = (res) => res
  const metadata = {}
  const options = {}
  return new Promise((resolve, reject) => {
    client.makeUnaryRequest(path, encoder, decoder, metadata, options, (err, res) => {
      if (err) {
        return reject(err)
      }
      resolve(res)
    })
  })
}

const run = async () => {
  const rootCert = fs.readFileSync('redacted.pem')
  const credentials = grpc.credentials.createSsl(rootCert)
  const client = new grpc.Client('redacted', credentials)
  const protocol = await new Promise((resolve, reject) => {
    avro.assembleProtocol('./PingService.avdl', (err, protocol) => {
      if (err) {
        return reject(err)
      }
      resolve(protocol)
    })
  })
  const pingType = protocol.types.find((type) => type.name === 'Ping')
  const path = '/PingService/ping'
  const response = await sendGrpcMessage(client, pingType, {}, path)
  console.log(await decodeMessage(response))
}

run()
.catch((err) => {
  console.error(err)
})