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.