Merge pull request #116 from topcoder-platform/Issue_115

cwdcwd · web-flow · commit e8e8cf92d47a · 2019-06-19T13:34:38.000-04:00
Duplicate all events to post to aggregate topic
diff --git a/config/default.js b/config/default.js
@@ -20,6 +20,7 @@ module.exports = {
   },
   BUSAPI_URL: process.env.BUSAPI_URL || 'https://api.topcoder-dev.com/v5',
   KAFKA_ERROR_TOPIC: process.env.KAFKA_ERROR_TOPIC || 'error.notification',
+  KAFKA_AGGREGATE_TOPIC: process.env.KAFKA_AGGREGATE_TOPIC,
   CHALLENGEAPI_URL: process.env.CHALLENGEAPI_URL || 'https://api.topcoder-dev.com/v4/challenges',
   AUTH0_URL: process.env.AUTH0_URL, // Auth0 credentials for Submission Service
   AUTH0_AUDIENCE: process.env.AUTH0_AUDIENCE || 'https://www.topcoder.com',
diff --git a/src/common/helper.js b/src/common/helper.js
@@ -517,17 +517,33 @@ function * downloadFile (fileURL) {
   return downloadedFile.Body
 }
 
+/**
+ * Wrapper function to post to bus api. Ensures that every event posted to bus api
+ * is duplicated and posted to bus api again, but to a different "aggregate" topic
+ * @param {Object} payload Data that needs to be posted to the bus api
+ */
+function * postToBusApi (payload) {
+  const busApiClient = getBusApiClient()
+
+  yield busApiClient.postEvent(payload)
+
+  // Post to aggregate topic
+  payload['topic'] = config.get('KAFKA_AGGREGATE_TOPIC')
+
+  yield busApiClient.postEvent(payload)
+}
+
 module.exports = {
   wrapExpress,
   autoWrapExpress,
   getEsClient,
-  getBusApiClient,
   fetchFromES,
   camelize,
   setPaginationHeaders,
   getSubmissionPhaseId,
   checkCreateAccess,
   checkGetAccess,
   checkReviewGetAccess,
-  downloadFile
+  downloadFile,
+  postToBusApi
 }
diff --git a/src/services/ReviewService.js b/src/services/ReviewService.js
@@ -16,7 +16,6 @@ const {
 const HelperService = require('./HelperService')
 const SubmissionService = require('./SubmissionService')
 
-const busApiClient = helper.getBusApiClient()
 const table = 'Review'
 
 /**
@@ -135,7 +134,7 @@ function * createReview (authUser, entity) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 
   // Inserting records in DynamoDB doesn't return any response
   // Hence returning the same entity to be in compliance with Swagger
@@ -217,7 +216,7 @@ function * _updateReview (authUser, reviewId, entity) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 
   // Updating records in DynamoDB doesn't return any response
   // Hence returning the response which will be in compliance with Swagger
@@ -310,7 +309,7 @@ function * deleteReview (reviewId) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 }
 
 deleteReview.schema = {
diff --git a/src/services/ReviewSummationService.js b/src/services/ReviewSummationService.js
@@ -11,7 +11,6 @@ const helper = require('../common/helper')
 const { originator, mimeType, events } = require('../../constants').busApiMeta
 const HelperService = require('./HelperService')
 
-const busApiClient = helper.getBusApiClient()
 const table = 'ReviewSummation'
 
 /**
@@ -112,7 +111,7 @@ function * createReviewSummation (authUser, entity) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 
   // Inserting records in DynamoDB doesn't return any response
   // Hence returning the same entity to be in compliance with Swagger
@@ -212,7 +211,7 @@ function * _updateReviewSummation (authUser, reviewSummationId, entity) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 
   // Updating records in DynamoDB doesn't return any response
   // Hence returning the response which will be in compliance with Swagger
@@ -302,7 +301,7 @@ function * deleteReviewSummation (reviewSummationId) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 }
 
 deleteReviewSummation.schema = {
diff --git a/src/services/ReviewTypeService.js b/src/services/ReviewTypeService.js
@@ -10,7 +10,6 @@ const dbhelper = require('../common/dbhelper')
 const helper = require('../common/helper')
 const { originator, mimeType, events } = require('../../constants').busApiMeta
 
-const busApiClient = helper.getBusApiClient()
 const table = 'ReviewType'
 
 /**
@@ -93,7 +92,7 @@ function * createReviewType (entity) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 
   // Inserting records in DynamoDB doesn't return any response
   // Hence returning the same entity to be in compliance with Swagger
@@ -157,7 +156,7 @@ function * _updateReviewType (reviewTypeId, entity) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 
   // Updating records in DynamoDB doesn't return any response
   // Hence returning the response which will be in compliance with Swagger
@@ -235,7 +234,7 @@ function * deleteReviewType (reviewTypeId) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 }
 
 deleteReviewType.schema = {
diff --git a/src/services/SubmissionService.js b/src/services/SubmissionService.js
@@ -16,7 +16,6 @@ const { submissionIndex } = require('../../constants')
 const s3 = new AWS.S3()
 const logger = require('winston')
 
-const busApiClient = helper.getBusApiClient()
 const table = 'Submission'
 
 /*
@@ -308,7 +307,7 @@ function * createSubmission (authUser, files, entity) {
   logger.info('Prepared submission create event payload to pass to THE bus')
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 
   // Inserting records in DynamoDB doesn't return any response
   // Hence returning the entity which is in compliance with Swagger
@@ -411,7 +410,7 @@ function * _updateSubmission (authUser, submissionId, entity) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 
   // Updating records in DynamoDB doesn't return any response
   // Hence returning the response which will be in compliance with Swagger
@@ -503,7 +502,7 @@ function * deleteSubmission (submissionId) {
   }
 
   // Post to Bus API using Client
-  yield busApiClient.postEvent(reqBody)
+  yield helper.postToBusApi(reqBody)
 }
 
 deleteSubmission.schema = {

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,6 @@ const {`
`16`	`16`	`const HelperService = require('./HelperService')`
`17`	`17`	`const SubmissionService = require('./SubmissionService')`
`18`	`18`
`19`		`-const busApiClient = helper.getBusApiClient()`
`20`	`19`	`const table = 'Review'`
`21`	`20`
`22`	`21`	`/**`
`@@ -135,7 +134,7 @@ function * createReview (authUser, entity) {`
`135`	`134`	`}`
`136`	`135`
`137`	`136`	`// Post to Bus API using Client`
`138`		`- yield busApiClient.postEvent(reqBody)`
	`137`	`+ yield helper.postToBusApi(reqBody)`
`139`	`138`
`140`	`139`	`// Inserting records in DynamoDB doesn't return any response`
`141`	`140`	`// Hence returning the same entity to be in compliance with Swagger`
`@@ -217,7 +216,7 @@ function * _updateReview (authUser, reviewId, entity) {`
`217`	`216`	`}`
`218`	`217`
`219`	`218`	`// Post to Bus API using Client`
`220`		`- yield busApiClient.postEvent(reqBody)`
	`219`	`+ yield helper.postToBusApi(reqBody)`
`221`	`220`
`222`	`221`	`// Updating records in DynamoDB doesn't return any response`
`223`	`222`	`// Hence returning the response which will be in compliance with Swagger`
`@@ -310,7 +309,7 @@ function * deleteReview (reviewId) {`
`310`	`309`	`}`
`311`	`310`
`312`	`311`	`// Post to Bus API using Client`
`313`		`- yield busApiClient.postEvent(reqBody)`
	`312`	`+ yield helper.postToBusApi(reqBody)`
`314`	`313`	`}`
`315`	`314`
`316`	`315`	`deleteReview.schema = {`
Original file line number	Diff line number	Diff line change
`@@ -11,7 +11,6 @@ const helper = require('../common/helper')`
`11`	`11`	`const { originator, mimeType, events } = require('../../constants').busApiMeta`
`12`	`12`	`const HelperService = require('./HelperService')`
`13`	`13`
`14`		`-const busApiClient = helper.getBusApiClient()`
`15`	`14`	`const table = 'ReviewSummation'`
`16`	`15`
`17`	`16`	`/**`
`@@ -112,7 +111,7 @@ function * createReviewSummation (authUser, entity) {`
`112`	`111`	`}`
`113`	`112`
`114`	`113`	`// Post to Bus API using Client`
`115`		`- yield busApiClient.postEvent(reqBody)`
	`114`	`+ yield helper.postToBusApi(reqBody)`
`116`	`115`
`117`	`116`	`// Inserting records in DynamoDB doesn't return any response`
`118`	`117`	`// Hence returning the same entity to be in compliance with Swagger`
`@@ -212,7 +211,7 @@ function * _updateReviewSummation (authUser, reviewSummationId, entity) {`
`212`	`211`	`}`
`213`	`212`
`214`	`213`	`// Post to Bus API using Client`
`215`		`- yield busApiClient.postEvent(reqBody)`
	`214`	`+ yield helper.postToBusApi(reqBody)`
`216`	`215`
`217`	`216`	`// Updating records in DynamoDB doesn't return any response`
`218`	`217`	`// Hence returning the response which will be in compliance with Swagger`
`@@ -302,7 +301,7 @@ function * deleteReviewSummation (reviewSummationId) {`
`302`	`301`	`}`
`303`	`302`
`304`	`303`	`// Post to Bus API using Client`
`305`		`- yield busApiClient.postEvent(reqBody)`
	`304`	`+ yield helper.postToBusApi(reqBody)`
`306`	`305`	`}`
`307`	`306`
`308`	`307`	`deleteReviewSummation.schema = {`
Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,6 @@ const dbhelper = require('../common/dbhelper')`
`10`	`10`	`const helper = require('../common/helper')`
`11`	`11`	`const { originator, mimeType, events } = require('../../constants').busApiMeta`
`12`	`12`
`13`		`-const busApiClient = helper.getBusApiClient()`
`14`	`13`	`const table = 'ReviewType'`
`15`	`14`
`16`	`15`	`/**`
`@@ -93,7 +92,7 @@ function * createReviewType (entity) {`
`93`	`92`	`}`
`94`	`93`
`95`	`94`	`// Post to Bus API using Client`
`96`		`- yield busApiClient.postEvent(reqBody)`
	`95`	`+ yield helper.postToBusApi(reqBody)`
`97`	`96`
`98`	`97`	`// Inserting records in DynamoDB doesn't return any response`
`99`	`98`	`// Hence returning the same entity to be in compliance with Swagger`
`@@ -157,7 +156,7 @@ function * _updateReviewType (reviewTypeId, entity) {`
`157`	`156`	`}`
`158`	`157`
`159`	`158`	`// Post to Bus API using Client`
`160`		`- yield busApiClient.postEvent(reqBody)`
	`159`	`+ yield helper.postToBusApi(reqBody)`
`161`	`160`
`162`	`161`	`// Updating records in DynamoDB doesn't return any response`
`163`	`162`	`// Hence returning the response which will be in compliance with Swagger`
`@@ -235,7 +234,7 @@ function * deleteReviewType (reviewTypeId) {`
`235`	`234`	`}`
`236`	`235`
`237`	`236`	`// Post to Bus API using Client`
`238`		`- yield busApiClient.postEvent(reqBody)`
	`237`	`+ yield helper.postToBusApi(reqBody)`
`239`	`238`	`}`
`240`	`239`
`241`	`240`	`deleteReviewType.schema = {`