Есть ли способ выполнить производителя в kafkajs и выйти из функции без process.exit ()? - PullRequest
Новая
       25

Есть ли способ выполнить производителя в kafkajs и выйти из функции без process.exit ()?

0 голосов
/ 03 апреля 2019

Я устанавливаю сервер в nodejs, который использует kafka. Для этого я использую kafkaJS. Проблема в том, что я не хочу, чтобы сервер отключался каждый раз, когда кто-то отправляет сообщение или читает значение из kafka.

До сих пор я пытался это прочитать данные 

const readData = async (receiver, stream_name) => {
  return new Promise((resolve, reject) => {
    consumer.connect()
    consumer.subscribe({ topic: stream_name })
      .catch(e => {
        console.log(stream_name)
        console.log('failed to subscribe to topic')
        console.log('err: ', e)
        reject(e)
      })
    consumer.run({
      eachMessage: async ({ topic, message }) => {
        receiver(`- ${topic} ${message.timestamp} ${message.key}#${message.value}`)
        resolve('ok')
      }
    })
      .then(e => {
        console.log('reading')
        resolve('read')
      })
      .catch(e => {
        console.log('messed up', e)
        reject('fail')
      })
    setTimeout(() => {
      console.log('egging')
      return 0
    }, 10000)
  })
}

и это для создания данных 

const pushData = async payload => {
  return new Promise((resolve, reject) => {
    producer.send(payload)
            .then( e => {
              resolve(e)
            })
            .catch(e => {
              console.log('Error pushing data to kafka broker:\n', e)
              reject(e)
            })
  })
}

// run ({String topic, {String key, JSON value[]} messages} payload)
const putData = async payload => {
  console.log('Connecting to kafka server...')
  await producer.connect()
  const a = await pushData(payload)
  console.log('Data sent: ', a)
  await producer.disconnect()
  process.abort()
}

Этот код работает хорошо, но его нужно записать, чтобы выйти из метода. Я хотел найти решение, в котором я мог бы либо убить процесс / поток, в котором выполняется интерфейс kafka, либо регулярно завершать работу метода.

Любая помощь приветствуется.

...