Node PubSub: publish many messages without batching

Viewed 16

I have a GCF that publishes messages from a new-line-delimited JSON from GCS:

const { PubSub } = require('@google-cloud/pubsub')
const { Storage } = require('@google-cloud/storage')
const readline = require('readline')

const pubSubClient = new PubSub()

const publish = async (topic, rowData) => {
  const dataBuffer = Buffer.from(JSON.stringify(JSON.parse(rowData)))
  try {
    await topic.publishMessage({data: dataBuffer})
  } catch (error) {
    console.error(`Received error while publishing: ${error.message}`)
  }
}

exports['gcs-to-pubsub'] = async fileRef => {
  console.log(`processing ${fileRef.name}...`)
  const storage = new Storage()
  const myBucket = storage.bucket(fileRef.bucket)
  const file = myBucket.file(fileRef.name)
  const topicName = fileRef.name.split('.')[0]
  console.log(`publishing to ${topicName}`)
  const topicKey = `projects/${process.env.GCP_PROJECT}/topics/${topicName}`
  const topic = await pubSubClient.topic(topicKey)
  return await new Promise(resolve => {
    const stream = file
    .createReadStream()
    .on('error', error => {
      throw Error(error)
    })
    .on('end', () => {
      console.log(`Parsed all rows`)
      file.delete()
      console.log('file deleted')
      resolve()
    })
    const rl = readline.createInterface({
      input: stream,
      crlfDelay: Infinity
    });
    rl.on('line', (line) => {
      publish(topic, line)
    })
  })
}

The problem is that when a file contains over 100k rows, the publishing part starts throwing errors:

Received error while publishing: Total timeout of API google.pubsub.v1.Publisher exceeded 60000 milliseconds before any response was received.

I see where the problem is, but I'm not sure how to solve it.

0 Answers
Related