add filter for specific log key

src/amqp-writable-stream.js

@@ -1,3 +1,4 @@
+const config = require('./config');
 const { Writable } = require('stream');
 const stringify = require('json-stringify-safe');
 
@@ -12,6 +13,11 @@ module.exports = class AmqpWritableStream extends Writable {
     channel.on('drain', () => this.emit('drain'));
   }
   write (chunk, encoding, callback) {
+    if (typeof chunk !== 'object') return true; // chunk should be an object, return true to keep going.
+    if (config.requiredLogKey) { // check to see if the required log key is in this chunk
+      const logs = Array.isArray(chunk.value) ? chunk.value : chunk.value.logs;
+      if (!logs.some(log => Object.keys(log).includes(config.requiredLogKey))) return true;
+    }
     return this.channel.sendToQueue(this.queueName, Buffer.from(stringify(chunk)), callback);
   }
 };

src/config.js

@@ -12,6 +12,7 @@ module.exports = {
   db: parseConnection(process.env.DATABASE_URL),
   tableName: process.env.DATABASE_TABLE_NAME,
   logOffset: parseInt(process.env.LOG_OFFSET) || 0,
+  requiredLogKey: process.env.REQUIRED_LOG_KEY,
   amqp: {
     url: process.env.AMQP_URL,
     queueName: process.env.AMQP_SEARCH_QUEUE