87 lines
2.0 KiB
JavaScript
87 lines
2.0 KiB
JavaScript
const Queue = require('../queue');
|
|
|
|
module.exports = class QueueFk extends Queue {
|
|
constructor(consumer, name, conf) {
|
|
super(consumer, name, conf);
|
|
this.reset();
|
|
}
|
|
|
|
async consume() {
|
|
await super.consume();
|
|
this.flush();
|
|
}
|
|
|
|
reset() {
|
|
this.lastMessage = null;
|
|
this.nMessages = 0;
|
|
this.ids = new Set();
|
|
}
|
|
|
|
flush(flushInterval) {
|
|
if (this.timeout) {
|
|
clearTimeout(this.timeout);
|
|
this.timeout = null;
|
|
}
|
|
this.timeout = setTimeout(
|
|
() => this.onFlushTimeout(), flushInterval);
|
|
}
|
|
|
|
async onFlushTimeout() {
|
|
const consumer = this.consumer;
|
|
|
|
if (this.ids.size) {
|
|
if (consumer.conf.debug)
|
|
console.debug('Flush:'.blue, this.name.yellow, this.ids);
|
|
|
|
const ids = Array.from(this.ids);
|
|
const lastMessage = this.lastMessage;
|
|
this.reset();
|
|
|
|
try {
|
|
for (const id of ids) {
|
|
const params = {id};
|
|
//const sql = consumer.db.format(this.conf.query, params);
|
|
const sql = this.conf.query;
|
|
if (consumer.conf.debug)
|
|
console.debug('SQL:'.blue, sql, params);
|
|
if (!consumer.conf.testMode)
|
|
await consumer.db.query(sql, params);
|
|
}
|
|
|
|
await this.channel.ack(lastMessage, true);
|
|
} catch(err) {
|
|
await this.channel.nack(lastMessage, true);
|
|
console.error(err);
|
|
}
|
|
}
|
|
|
|
this.flush(this.conf.flushInterval);
|
|
}
|
|
|
|
async onConsume(msg) {
|
|
const consumer = this.consumer;
|
|
const data = JSON.parse(msg.content.toString());
|
|
|
|
if (consumer.conf.debug)
|
|
console.debug('Message:'.blue, this.name.yellow, data.table);
|
|
|
|
const ids = this.ids;
|
|
const key = this.conf.includeSchema[data.schema][data.table].key;
|
|
|
|
if (data.eventName === 'updaterows') {
|
|
for (const row of data.rows) {
|
|
ids.add(row.before[key]);
|
|
ids.add(row.after[key]);
|
|
}
|
|
} else
|
|
for (const row of data.rows)
|
|
ids.add(row[key]);
|
|
|
|
this.nMessages++;
|
|
this.lastMessage = msg;
|
|
|
|
if (this.nMessages == consumer.conf.amqpPrefetch)
|
|
this.flush();
|
|
}
|
|
}
|