Post event to aggregate topic as well

Author: callmekatootie

README.md

@@ -35,6 +35,7 @@ Configuration for the application is at config/default.js and config/production.
 - UBAHN_CREATE_TOPIC: Kafka topic for create message
 - UBAHN_UPDATE_TOPIC: Kafka topic for update message
 - UBAHN_DELETE_TOPIC: Kafka topic for delete message
+- UBAHN_AGGREGATE_TOPIC: Kafka topic that is used to combine all create, update and delete message(s)
 - ES.HOST: Elasticsearch host
 - ES.API_VERSION: Elasticsearch API version
 - ES.DOCUMENTS: Elasticsearch index, type and id mapping for resources.

config/default.js

@@ -36,6 +36,7 @@ module.exports = {
   UBAHN_CREATE_TOPIC: process.env.UBAHN_CREATE_TOPIC || 'u-bahn.action.create',
   UBAHN_UPDATE_TOPIC: process.env.UBAHN_UPDATE_TOPIC || 'u-bahn.action.update',
   UBAHN_DELETE_TOPIC: process.env.UBAHN_DELETE_TOPIC || 'u-bahn.action.delete',
+  UBAHN_AGGREGATE_TOPIC: process.env.UBAHN_AGGREGATE_TOPIC || 'u-bahn.action.aggregate',
 
   EMSI: {
     CLIENT_ID: process.env.EMSI_CLIENT_ID,

src/common/helper.js

@@ -246,6 +246,12 @@ async function postEvent (topic, payload) {
     payload
   }
   await busApiClient.postEvent(message)
+
+  // Post to the aggregate topic
+  message.payload.originalTopic = topic
+  message.topic = config.UBAHN_AGGREGATE_TOPIC
+  logger.debug(`Posting event to aggregate topic ${message.topic}`)
+  await busApiClient.postEvent(message)
 }
 
 module.exports = {