source: node_modules/nodemailer/lib/ses-transport/index.js

main
Last change on this file was 81bc7da, checked in by Klimentina Efremova <klimentina08642@…>, 3 months ago

Initial commit
  • Property mode set to 100644
File size: 10.8 KB
Line 
1'use strict';
2
3const EventEmitter = require('events');
4const packageData = require('../../package.json');
5const shared = require('../shared');
6const LeWindows = require('../mime-node/le-windows');
7
8/**
9 * Generates a Transport object for AWS SES
10 *
11 * Possible options can be the following:
12 *
13 * * **sendingRate** optional Number specifying how many messages per second should be delivered to SES
14 * * **maxConnections** optional Number specifying max number of parallel connections to SES
15 *
16 * @constructor
17 * @param {Object} optional config parameter
18 */
19class SESTransport extends EventEmitter {
20 constructor(options) {
21 super();
22 options = options || {};
23
24 this.options = options || {};
25 this.ses = this.options.SES;
26
27 this.name = 'SESTransport';
28 this.version = packageData.version;
29
30 this.logger = shared.getLogger(this.options, {
31 component: this.options.component || 'ses-transport'
32 });
33
34 // parallel sending connections
35 this.maxConnections = Number(this.options.maxConnections) || Infinity;
36 this.connections = 0;
37
38 // max messages per second
39 this.sendingRate = Number(this.options.sendingRate) || Infinity;
40 this.sendingRateTTL = null;
41 this.rateInterval = 1000; // milliseconds
42 this.rateMessages = [];
43
44 this.pending = [];
45
46 this.idling = true;
47
48 setImmediate(() => {
49 if (this.idling) {
50 this.emit('idle');
51 }
52 });
53 }
54
55 /**
56 * Schedules a sending of a message
57 *
58 * @param {Object} emailMessage MailComposer object
59 * @param {Function} callback Callback function to run when the sending is completed
60 */
61 send(mail, callback) {
62 if (this.connections >= this.maxConnections) {
63 this.idling = false;
64 return this.pending.push({
65 mail,
66 callback
67 });
68 }
69
70 if (!this._checkSendingRate()) {
71 this.idling = false;
72 return this.pending.push({
73 mail,
74 callback
75 });
76 }
77
78 this._send(mail, (...args) => {
79 setImmediate(() => callback(...args));
80 this._sent();
81 });
82 }
83
84 _checkRatedQueue() {
85 if (this.connections >= this.maxConnections || !this._checkSendingRate()) {
86 return;
87 }
88
89 if (!this.pending.length) {
90 if (!this.idling) {
91 this.idling = true;
92 this.emit('idle');
93 }
94 return;
95 }
96
97 let next = this.pending.shift();
98 this._send(next.mail, (...args) => {
99 setImmediate(() => next.callback(...args));
100 this._sent();
101 });
102 }
103
104 _checkSendingRate() {
105 clearTimeout(this.sendingRateTTL);
106
107 let now = Date.now();
108 let oldest = false;
109 // delete older messages
110 for (let i = this.rateMessages.length - 1; i >= 0; i--) {
111 if (this.rateMessages[i].ts >= now - this.rateInterval && (!oldest || this.rateMessages[i].ts < oldest)) {
112 oldest = this.rateMessages[i].ts;
113 }
114
115 if (this.rateMessages[i].ts < now - this.rateInterval && !this.rateMessages[i].pending) {
116 this.rateMessages.splice(i, 1);
117 }
118 }
119
120 if (this.rateMessages.length < this.sendingRate) {
121 return true;
122 }
123
124 let delay = Math.max(oldest + 1001, now + 20);
125 this.sendingRateTTL = setTimeout(() => this._checkRatedQueue(), now - delay);
126
127 try {
128 this.sendingRateTTL.unref();
129 } catch (E) {
130 // Ignore. Happens on envs with non-node timer implementation
131 }
132
133 return false;
134 }
135
136 _sent() {
137 this.connections--;
138 this._checkRatedQueue();
139 }
140
141 /**
142 * Returns true if there are free slots in the queue
143 */
144 isIdle() {
145 return this.idling;
146 }
147
148 /**
149 * Compiles a mailcomposer message and forwards it to SES
150 *
151 * @param {Object} emailMessage MailComposer object
152 * @param {Function} callback Callback function to run when the sending is completed
153 */
154 _send(mail, callback) {
155 let statObject = {
156 ts: Date.now(),
157 pending: true
158 };
159 this.connections++;
160 this.rateMessages.push(statObject);
161
162 let envelope = mail.data.envelope || mail.message.getEnvelope();
163 let messageId = mail.message.messageId();
164
165 let recipients = [].concat(envelope.to || []);
166 if (recipients.length > 3) {
167 recipients.push('...and ' + recipients.splice(2).length + ' more');
168 }
169 this.logger.info(
170 {
171 tnx: 'send',
172 messageId
173 },
174 'Sending message %s to <%s>',
175 messageId,
176 recipients.join(', ')
177 );
178
179 let getRawMessage = next => {
180 // do not use Message-ID and Date in DKIM signature
181 if (!mail.data._dkim) {
182 mail.data._dkim = {};
183 }
184 if (mail.data._dkim.skipFields && typeof mail.data._dkim.skipFields === 'string') {
185 mail.data._dkim.skipFields += ':date:message-id';
186 } else {
187 mail.data._dkim.skipFields = 'date:message-id';
188 }
189
190 let sourceStream = mail.message.createReadStream();
191 let stream = sourceStream.pipe(new LeWindows());
192 let chunks = [];
193 let chunklen = 0;
194
195 stream.on('readable', () => {
196 let chunk;
197 while ((chunk = stream.read()) !== null) {
198 chunks.push(chunk);
199 chunklen += chunk.length;
200 }
201 });
202
203 sourceStream.once('error', err => stream.emit('error', err));
204
205 stream.once('error', err => {
206 next(err);
207 });
208
209 stream.once('end', () => next(null, Buffer.concat(chunks, chunklen)));
210 };
211
212 setImmediate(() =>
213 getRawMessage((err, raw) => {
214 if (err) {
215 this.logger.error(
216 {
217 err,
218 tnx: 'send',
219 messageId
220 },
221 'Failed creating message for %s. %s',
222 messageId,
223 err.message
224 );
225 statObject.pending = false;
226 return callback(err);
227 }
228
229 let sesMessage = {
230 RawMessage: {
231 // required
232 Data: raw // required
233 },
234 Source: envelope.from,
235 Destinations: envelope.to
236 };
237
238 Object.keys(mail.data.ses || {}).forEach(key => {
239 sesMessage[key] = mail.data.ses[key];
240 });
241
242 let ses = (this.ses.aws ? this.ses.ses : this.ses) || {};
243 let aws = this.ses.aws || {};
244
245 let getRegion = cb => {
246 if (ses.config && typeof ses.config.region === 'function') {
247 // promise
248 return ses.config
249 .region()
250 .then(region => cb(null, region))
251 .catch(err => cb(err));
252 }
253 return cb(null, (ses.config && ses.config.region) || 'us-east-1');
254 };
255
256 getRegion((err, region) => {
257 if (err || !region) {
258 region = 'us-east-1';
259 }
260
261 let sendPromise;
262 if (typeof ses.send === 'function' && aws.SendRawEmailCommand) {
263 // v3 API
264 sendPromise = ses.send(new aws.SendRawEmailCommand(sesMessage));
265 } else {
266 // v2 API
267 sendPromise = ses.sendRawEmail(sesMessage).promise();
268 }
269
270 sendPromise
271 .then(data => {
272 if (region === 'us-east-1') {
273 region = 'email';
274 }
275
276 statObject.pending = false;
277 callback(null, {
278 envelope: {
279 from: envelope.from,
280 to: envelope.to
281 },
282 messageId: '<' + data.MessageId + (!/@/.test(data.MessageId) ? '@' + region + '.amazonses.com' : '') + '>',
283 response: data.MessageId,
284 raw
285 });
286 })
287 .catch(err => {
288 this.logger.error(
289 {
290 err,
291 tnx: 'send'
292 },
293 'Send error for %s: %s',
294 messageId,
295 err.message
296 );
297 statObject.pending = false;
298 callback(err);
299 });
300 });
301 })
302 );
303 }
304
305 /**
306 * Verifies SES configuration
307 *
308 * @param {Function} callback Callback function
309 */
310 verify(callback) {
311 let promise;
312 let ses = (this.ses.aws ? this.ses.ses : this.ses) || {};
313 let aws = this.ses.aws || {};
314
315 const sesMessage = {
316 RawMessage: {
317 // required
318 Data: 'From: invalid@invalid\r\nTo: invalid@invalid\r\n Subject: Invalid\r\n\r\nInvalid'
319 },
320 Source: 'invalid@invalid',
321 Destinations: ['invalid@invalid']
322 };
323
324 if (!callback) {
325 promise = new Promise((resolve, reject) => {
326 callback = shared.callbackPromise(resolve, reject);
327 });
328 }
329 const cb = err => {
330 if (err && (err.code || err.Code) !== 'InvalidParameterValue') {
331 return callback(err);
332 }
333 return callback(null, true);
334 };
335
336 if (typeof ses.send === 'function' && aws.SendRawEmailCommand) {
337 // v3 API
338 sesMessage.RawMessage.Data = Buffer.from(sesMessage.RawMessage.Data);
339 ses.send(new aws.SendRawEmailCommand(sesMessage), cb);
340 } else {
341 // v2 API
342 ses.sendRawEmail(sesMessage, cb);
343 }
344
345 return promise;
346 }
347}
348
349module.exports = SESTransport;
Note: See TracBrowser for help on using the repository browser.