Skip to content

Commit 4992dd7

Browse files
author
Hamid Tavakoli
committed
WIP- Branch change
1 parent aca56bc commit 4992dd7

File tree

7 files changed

+105
-94
lines changed

7 files changed

+105
-94
lines changed

‎connect/connectEmailServer.js

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -9,27 +9,27 @@ const logger = require('../src/common/logger');
99

1010
// set configuration for the server, see ../config/default.js for available config parameters
1111
// setConfig should be called before initDatabase and start functions
12-
// emailServer.setConfig({ LOG_LEVEL: config.LOG_LEVEL });
12+
emailServer.setConfig({ LOG_LEVEL: config.LOG_LEVEL });
1313

1414
// add topic handlers,
1515
// handler is used build a notification list for a message of a topic,
16-
// it is defined as: function(topic, message, callback),
16+
// it is defined as: function(topic, message),
1717
// the topic is topic name,
1818
// the message is JSON event message,
19-
// the callback is function(error, templateId), where templateId is the used SendGrid template id
20-
const handler = (topic, message, callback) => {
19+
const handler = async (topic, message) => {
2120
let templateId = config.TEMPLATE_MAP[topic];
2221
templateId = _.get(message, config.PAYLOAD_SENDGRID_TEMPLATE_KEY, templateId);
2322
if (!templateId) {
24-
return callback(null, { success: false, error: `Template not found for topic ${topic}` });
23+
return { success: false, error: `Template not found for topic ${topic}` };
2524
}
2625

27-
service.sendEmail(templateId, message).then(() => {
28-
callback(null, { success: true });
29-
}).catch((err) => {
26+
try {
27+
await service.sendEmail(templateId, message)
28+
return { success: true };
29+
} catch (err) {
3030
logger.error("Error occurred in sendgrid api calling:", err);
31-
callback(null, { success: false, error: err });
32-
});
31+
return { success: false, error: err };
32+
}
3333

3434
};
3535

‎connect/service.js

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ const sendEmail = async (templateId, message) => { // send email
4949
bcc,
5050
};
5151
}
52-
return sgMail.send(msg)
52+
return await sgMail.send(msg)
5353
}
5454
module.exports = {
5555
sendEmail,

‎index.js

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ const config = require('config');
33
const jwtAuth = require('tc-core-library-js').middleware.jwtAuthenticator;
44
const express = require('express');
55
const _ = require('lodash');
6-
const schedule = require('node-schedule');
76
const cors = require('cors');
87
const bodyParser = require('body-parser');
98
const helper = require('./src/common/helper');
@@ -112,7 +111,7 @@ app.use((req, res) => {
112111
res.status(404).json({ error: 'route not found' });
113112
});
114113

115-
app.use((err, req, res, next) => { // eslint-disable-line
114+
app.use((err, req, res) => { // eslint-disable-line
116115
logger.logFullError(err, req.signature);
117116
let status = err.httpStatus || 500;
118117
if (err.isJoi) {

‎src/init.js

Lines changed: 77 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ const models = require('./models');
1414
* @param {Object} handlers the handlers
1515
*/
1616
async function configureKafkaConsumer(handlers) {
17+
1718
// create group consumer
1819
const options = { brokers: config.KAFKA_URL.split(',') };
1920
if (config.KAFKA_CLIENT_CERT && config.KAFKA_CLIENT_CERT_KEY) {
@@ -23,17 +24,22 @@ async function configureKafkaConsumer(handlers) {
2324
const consumer = kafka.consumer({ groupId: config.KAFKA_GROUP_ID });
2425
await consumer.connect()
2526
await consumer.subscribe({ topics: _.keys(handlers) });
26-
dataHandler(consumer).catch((err) => {
27+
dataHandler(consumer, handlers).catch((err) => {
2728
logger.error(err);
2829
});
2930
}
3031

3132

32-
async function dataHandler(consumer) {
33+
async function dataHandler(consumer, handlers) {
3334
await consumer.run({
34-
eachMessage: async ({ topic, partition, msg }) => {
35+
eachMessage: async (data) => {
36+
const topic = data.topic
37+
const msg = data.message
38+
const partition = data.partition
39+
//If there is no message, return
40+
if (!msg) return
3541
const message = msg.value.toString('utf8')
36-
logger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Offset: ${m.offset}; Message: ${message}.`);
42+
logger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Message: ${message}.`);
3743
// ignore configured Kafka topic prefix
3844
let topicName = topic;
3945
// find handler
@@ -43,37 +49,49 @@ async function dataHandler(consumer) {
4349
// return null to ignore this message
4450
return null;
4551
}
46-
let emailModel = {};
52+
console.log([1])
53+
const emailModel = await models.loadEmailModule()
4754
const busPayload = JSON.parse(message);
4855
const messageJSON = busPayload.payload;
4956
try {
5057

51-
const result = await models.Email.create({
58+
const emailInfo = {
5259
status: 'PENDING',
5360
topicName,
5461
data: JSON.stringify(messageJSON),
5562
recipients: JSON.stringify(messageJSON.recipients),
56-
})
63+
}
64+
65+
try {
66+
console.log(emailModel)
67+
await emailModel.create(emailInfo)
5768

58-
logger.log('info', 'Email sent', {
69+
} catch (err) {
70+
console.log(err)
71+
}
72+
const result = await handler(topicName, messageJSON);
73+
74+
logger.info('info', 'Email sent', {
5975
sender: 'Connect',
6076
to_address: messageJSON.recipients.join(','),
6177
from_address: config.EMAIL_FROM,
6278
status: result.success ? 'Message accepted' : 'Message rejected',
6379
error: result.error ? result.error.toString() : 'No error message',
6480
});
81+
console.log("******************* result *******************", result)
6582

6683
if (result.success) {
6784
emailTries[topicName] = 0;
6885
emailModel.status = 'SUCCESS';
69-
emailModel.save();
86+
await emailModel.save();
7087
} else {
7188
// emailTries[topicName] += 1; //temporary disabling this feature
7289
if (result.error) {
73-
logger.log('error', 'Send email error details', result.error);
90+
logger.error('error', 'Send email error details', result.error);
7491
}
7592
}
7693
} catch (e) {
94+
console.log(e)
7795
logger.error(e)
7896
}
7997

@@ -83,31 +101,31 @@ async function dataHandler(consumer) {
83101
},
84102
})
85103

86-
const errorTypes = ['unhandledRejection', 'uncaughtException']
87-
const signalTraps = ['SIGTERM', 'SIGINT', 'SIGUSR2']
88-
89-
errorTypes.forEach(type => {
90-
process.on(type, async e => {
91-
try {
92-
console.log(`process.on ${type}`)
93-
console.error(e)
94-
await consumer.disconnect()
95-
process.exit(0)
96-
} catch (_) {
97-
process.exit(1)
98-
}
99-
})
100-
})
101-
102-
signalTraps.forEach(type => {
103-
process.once(type, async () => {
104-
try {
105-
await consumer.disconnect()
106-
} finally {
107-
process.kill(process.pid, type)
108-
}
109-
})
110-
})
104+
// const errorTypes = ['unhandledRejection', 'uncaughtException']
105+
// const signalTraps = ['SIGTERM', 'SIGINT', 'SIGUSR2']
106+
107+
// errorTypes.forEach(type => {
108+
// process.on(type, async e => {
109+
// try {
110+
// console.log(`process.on ${type}`)
111+
// console.error(e)
112+
// await consumer.disconnect()
113+
// process.exit(0)
114+
// } catch (_) {
115+
// process.exit(1)
116+
// }
117+
// })
118+
// })
119+
120+
// signalTraps.forEach(type => {
121+
// process.once(type, async () => {
122+
// try {
123+
// await consumer.disconnect()
124+
// } finally {
125+
// process.kill(process.pid, type)
126+
// }
127+
// })
128+
// })
111129

112130
}
113131

@@ -116,34 +134,32 @@ async function dataHandler(consumer) {
116134
* Callback to retry sending email.
117135
* @param {Object} handlers the handlers
118136
*/
119-
function retryEmail(handlers) {
120-
return models.Email.findAll({ where: { status: 'FAILED', createdAt: { $gt: new Date(new Date() - config.EMAIL_RETRY_MAX_AGE) } } })
121-
.then((models) => {
122-
if (models.length > 0) {
123-
logger.info(`Found ${models.length} e-mails to be resent`);
124-
return Promise.each(models, (m) => {
125-
// find handler
126-
const handler = handlers[m.topicName];
127-
if (!handler) {
128-
logger.warn(`No handler configured for topic: ${m.topicName}`);
129-
return m;
130-
}
131-
const handlerAsync = Promise.promisify(handler);
132-
const messageJSON = { data: JSON.parse(m.data), recipients: JSON.parse(m.recipients) };
133-
return handlerAsync(m.topicName, messageJSON).then((result) => { // save email
134-
if (result.success) {
135-
logger.info(`Email model with ${m.id} was sent correctly`);
136-
m.status = 'SUCCESS';
137-
return m.save();
138-
}
139-
logger.info(`Email model with ${m.id} wasn't sent correctly`);
140-
return m;
141-
});
142-
});
143-
} else {
144-
return models;
137+
async function retryEmail(handlers) {
138+
const models = await models.Email.findAll({ where: { status: 'FAILED', createdAt: { $gt: new Date(new Date() - config.EMAIL_RETRY_MAX_AGE) } } })
139+
140+
if (models.length > 0) {
141+
logger.info(`Found ${models.length} e-mails to be resent`);
142+
models.map(async m => {
143+
// find handler
144+
const handler = handlers[m.topicName];
145+
if (!handler) {
146+
logger.warn(`No handler configured for topic: ${m.topicName}`);
147+
return m;
148+
}
149+
const messageJSON = { data: JSON.parse(m.data), recipients: JSON.parse(m.recipients) };
150+
const result = await handler(m.topicName, messageJSON);
151+
if (result.success) {
152+
logger.info(`Email model with ${m.id} was sent correctly`);
153+
m.status = 'SUCCESS';
154+
return m.save();
145155
}
156+
logger.info(`Email model with ${m.id} wasn't sent correctly`);
157+
return m;
146158
});
159+
} else {
160+
return models;
161+
}
162+
147163
}
148164

149165
async function initServer(handlers) {

‎src/models/Email.js

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,20 +8,20 @@
88
* @author TCSCODER
99
* @version 1.0
1010
*/
11-
async function Email(sequelize, DataTypes) {
11+
async function defineEmailModel(sequelize, DataTypes) {
1212
const Email = sequelize.define('Email', {
1313
id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true },
1414
topicName: { type: DataTypes.STRING, allowNull: true, field: 'topic_name' },
1515
data: { type: DataTypes.TEXT, allowNull: false },
1616
recipients: { type: DataTypes.STRING, allowNull: false },
1717
status: { type: DataTypes.STRING, allowNull: false },
18-
}, {});
18+
});
1919
await Email.sync();
2020
return Email;
2121
}
2222

2323
module.exports = {
24-
Email
24+
defineEmailModel
2525
}
2626

2727
// sequelize will generate and manage createdAt, updatedAt fields

‎src/models/index.js

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,14 +11,14 @@
1111

1212
const sequelizeInstance = require('./datasource');
1313
const DataTypes = require('sequelize/lib/data-types');
14-
const Email = require('./Email')
14+
const defineEmailModel = require('./Email')
1515

1616
async function loadSequelizeModule() {
1717
return await sequelizeInstance.getSequelize();
1818
}
1919
async function loadEmailModule() {
2020
const sequelize = await loadSequelizeModule();
21-
return await Email(sequelize, DataTypes);
21+
return defineEmailModel(sequelize, DataTypes);
2222
}
2323

2424
async function init() {
@@ -28,6 +28,6 @@ async function init() {
2828

2929

3030
module.exports = {
31-
Email: loadEmailModule,
31+
loadEmailModule,
3232
init,
3333
};

‎test/emailServiceTest.js

Lines changed: 10 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,30 +5,26 @@
55
const _ = require('lodash');
66
const sgMail = require('@sendgrid/mail');
77
const config = require('config');
8-
const assert = require('chai').assert;
98
const Kafka = require('no-kafka');
109

1110
const emailServer = require('../index');
1211
const service = require('../connect/service');
1312

1413
let globalDone = null;
1514

16-
const defaultHandler = (topic, message, callback) => {
15+
const defaultHandler = async (topic, message, callback) => {
1716
const templateId = config.TEMPLATE_MAP[topic];
1817
if (templateId === undefined) {
19-
callback(null, { success: false, error: `Template not found for topic ${topic}` });
20-
return finish(`Template not found for topic ${topic}`);
18+
return { success: false, error: `Template not found for topic ${topic}` };
2119
}
22-
23-
// send email
24-
service.sendEmail(templateId, message).then(() => {
25-
callback(null, { success: true });
26-
finish(null);
27-
}).catch((err) => {
28-
callback(null, { success: false, error: err });
29-
finish(err);
30-
});
31-
};
20+
try {
21+
// send email
22+
await service.sendEmail(templateId, message)
23+
return { success: true }
24+
} catch (e) {
25+
return { success: true }
26+
}
27+
}
3228

3329
// init all events
3430
_.keys(config.TEMPLATE_MAP).forEach((eventType) => {

0 commit comments

Comments
 (0)