source: node_modules/nodemailer/lib/smtp-pool/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: 19.2 KB
Line 
1'use strict';
2
3const EventEmitter = require('events');
4const PoolResource = require('./pool-resource');
5const SMTPConnection = require('../smtp-connection');
6const wellKnown = require('../well-known');
7const shared = require('../shared');
8const packageData = require('../../package.json');
9
10/**
11 * Creates a SMTP pool transport object for Nodemailer
12 *
13 * @constructor
14 * @param {Object} options SMTP Connection options
15 */
16class SMTPPool extends EventEmitter {
17 constructor(options) {
18 super();
19
20 options = options || {};
21 if (typeof options === 'string') {
22 options = {
23 url: options
24 };
25 }
26
27 let urlData;
28 let service = options.service;
29
30 if (typeof options.getSocket === 'function') {
31 this.getSocket = options.getSocket;
32 }
33
34 if (options.url) {
35 urlData = shared.parseConnectionUrl(options.url);
36 service = service || urlData.service;
37 }
38
39 this.options = shared.assign(
40 false, // create new object
41 options, // regular options
42 urlData, // url options
43 service && wellKnown(service) // wellknown options
44 );
45
46 this.options.maxConnections = this.options.maxConnections || 5;
47 this.options.maxMessages = this.options.maxMessages || 100;
48
49 this.logger = shared.getLogger(this.options, {
50 component: this.options.component || 'smtp-pool'
51 });
52
53 // temporary object
54 let connection = new SMTPConnection(this.options);
55
56 this.name = 'SMTP (pool)';
57 this.version = packageData.version + '[client:' + connection.version + ']';
58
59 this._rateLimit = {
60 counter: 0,
61 timeout: null,
62 waiting: [],
63 checkpoint: false,
64 delta: Number(this.options.rateDelta) || 1000,
65 limit: Number(this.options.rateLimit) || 0
66 };
67 this._closed = false;
68 this._queue = [];
69 this._connections = [];
70 this._connectionCounter = 0;
71
72 this.idling = true;
73
74 setImmediate(() => {
75 if (this.idling) {
76 this.emit('idle');
77 }
78 });
79 }
80
81 /**
82 * Placeholder function for creating proxy sockets. This method immediatelly returns
83 * without a socket
84 *
85 * @param {Object} options Connection options
86 * @param {Function} callback Callback function to run with the socket keys
87 */
88 getSocket(options, callback) {
89 // return immediatelly
90 return setImmediate(() => callback(null, false));
91 }
92
93 /**
94 * Queues an e-mail to be sent using the selected settings
95 *
96 * @param {Object} mail Mail object
97 * @param {Function} callback Callback function
98 */
99 send(mail, callback) {
100 if (this._closed) {
101 return false;
102 }
103
104 this._queue.push({
105 mail,
106 requeueAttempts: 0,
107 callback
108 });
109
110 if (this.idling && this._queue.length >= this.options.maxConnections) {
111 this.idling = false;
112 }
113
114 setImmediate(() => this._processMessages());
115
116 return true;
117 }
118
119 /**
120 * Closes all connections in the pool. If there is a message being sent, the connection
121 * is closed later
122 */
123 close() {
124 let connection;
125 let len = this._connections.length;
126 this._closed = true;
127
128 // clear rate limit timer if it exists
129 clearTimeout(this._rateLimit.timeout);
130
131 if (!len && !this._queue.length) {
132 return;
133 }
134
135 // remove all available connections
136 for (let i = len - 1; i >= 0; i--) {
137 if (this._connections[i] && this._connections[i].available) {
138 connection = this._connections[i];
139 connection.close();
140 this.logger.info(
141 {
142 tnx: 'connection',
143 cid: connection.id,
144 action: 'removed'
145 },
146 'Connection #%s removed',
147 connection.id
148 );
149 }
150 }
151
152 if (len && !this._connections.length) {
153 this.logger.debug(
154 {
155 tnx: 'connection'
156 },
157 'All connections removed'
158 );
159 }
160
161 if (!this._queue.length) {
162 return;
163 }
164
165 // make sure that entire queue would be cleaned
166 let invokeCallbacks = () => {
167 if (!this._queue.length) {
168 this.logger.debug(
169 {
170 tnx: 'connection'
171 },
172 'Pending queue entries cleared'
173 );
174 return;
175 }
176 let entry = this._queue.shift();
177 if (entry && typeof entry.callback === 'function') {
178 try {
179 entry.callback(new Error('Connection pool was closed'));
180 } catch (E) {
181 this.logger.error(
182 {
183 err: E,
184 tnx: 'callback',
185 cid: connection.id
186 },
187 'Callback error for #%s: %s',
188 connection.id,
189 E.message
190 );
191 }
192 }
193 setImmediate(invokeCallbacks);
194 };
195 setImmediate(invokeCallbacks);
196 }
197
198 /**
199 * Check the queue and available connections. If there is a message to be sent and there is
200 * an available connection, then use this connection to send the mail
201 */
202 _processMessages() {
203 let connection;
204 let i, len;
205
206 // do nothing if already closed
207 if (this._closed) {
208 return;
209 }
210
211 // do nothing if queue is empty
212 if (!this._queue.length) {
213 if (!this.idling) {
214 // no pending jobs
215 this.idling = true;
216 this.emit('idle');
217 }
218 return;
219 }
220
221 // find first available connection
222 for (i = 0, len = this._connections.length; i < len; i++) {
223 if (this._connections[i].available) {
224 connection = this._connections[i];
225 break;
226 }
227 }
228
229 if (!connection && this._connections.length < this.options.maxConnections) {
230 connection = this._createConnection();
231 }
232
233 if (!connection) {
234 // no more free connection slots available
235 this.idling = false;
236 return;
237 }
238
239 // check if there is free space in the processing queue
240 if (!this.idling && this._queue.length < this.options.maxConnections) {
241 this.idling = true;
242 this.emit('idle');
243 }
244
245 let entry = (connection.queueEntry = this._queue.shift());
246 entry.messageId = (connection.queueEntry.mail.message.getHeader('message-id') || '').replace(/[<>\s]/g, '');
247
248 connection.available = false;
249
250 this.logger.debug(
251 {
252 tnx: 'pool',
253 cid: connection.id,
254 messageId: entry.messageId,
255 action: 'assign'
256 },
257 'Assigned message <%s> to #%s (%s)',
258 entry.messageId,
259 connection.id,
260 connection.messages + 1
261 );
262
263 if (this._rateLimit.limit) {
264 this._rateLimit.counter++;
265 if (!this._rateLimit.checkpoint) {
266 this._rateLimit.checkpoint = Date.now();
267 }
268 }
269
270 connection.send(entry.mail, (err, info) => {
271 // only process callback if current handler is not changed
272 if (entry === connection.queueEntry) {
273 try {
274 entry.callback(err, info);
275 } catch (E) {
276 this.logger.error(
277 {
278 err: E,
279 tnx: 'callback',
280 cid: connection.id
281 },
282 'Callback error for #%s: %s',
283 connection.id,
284 E.message
285 );
286 }
287 connection.queueEntry = false;
288 }
289 });
290 }
291
292 /**
293 * Creates a new pool resource
294 */
295 _createConnection() {
296 let connection = new PoolResource(this);
297
298 connection.id = ++this._connectionCounter;
299
300 this.logger.info(
301 {
302 tnx: 'pool',
303 cid: connection.id,
304 action: 'conection'
305 },
306 'Created new pool resource #%s',
307 connection.id
308 );
309
310 // resource comes available
311 connection.on('available', () => {
312 this.logger.debug(
313 {
314 tnx: 'connection',
315 cid: connection.id,
316 action: 'available'
317 },
318 'Connection #%s became available',
319 connection.id
320 );
321
322 if (this._closed) {
323 // if already closed run close() that will remove this connections from connections list
324 this.close();
325 } else {
326 // check if there's anything else to send
327 this._processMessages();
328 }
329 });
330
331 // resource is terminated with an error
332 connection.once('error', err => {
333 if (err.code !== 'EMAXLIMIT') {
334 this.logger.error(
335 {
336 err,
337 tnx: 'pool',
338 cid: connection.id
339 },
340 'Pool Error for #%s: %s',
341 connection.id,
342 err.message
343 );
344 } else {
345 this.logger.debug(
346 {
347 tnx: 'pool',
348 cid: connection.id,
349 action: 'maxlimit'
350 },
351 'Max messages limit exchausted for #%s',
352 connection.id
353 );
354 }
355
356 if (connection.queueEntry) {
357 try {
358 connection.queueEntry.callback(err);
359 } catch (E) {
360 this.logger.error(
361 {
362 err: E,
363 tnx: 'callback',
364 cid: connection.id
365 },
366 'Callback error for #%s: %s',
367 connection.id,
368 E.message
369 );
370 }
371 connection.queueEntry = false;
372 }
373
374 // remove the erroneus connection from connections list
375 this._removeConnection(connection);
376
377 this._continueProcessing();
378 });
379
380 connection.once('close', () => {
381 this.logger.info(
382 {
383 tnx: 'connection',
384 cid: connection.id,
385 action: 'closed'
386 },
387 'Connection #%s was closed',
388 connection.id
389 );
390
391 this._removeConnection(connection);
392
393 if (connection.queueEntry) {
394 // If the connection closed when sending, add the message to the queue again
395 // if max number of requeues is not reached yet
396 // Note that we must wait a bit.. because the callback of the 'error' handler might be called
397 // in the next event loop
398 setTimeout(() => {
399 if (connection.queueEntry) {
400 if (this._shouldRequeuOnConnectionClose(connection.queueEntry)) {
401 this._requeueEntryOnConnectionClose(connection);
402 } else {
403 this._failDeliveryOnConnectionClose(connection);
404 }
405 }
406 this._continueProcessing();
407 }, 50);
408 } else {
409 this._continueProcessing();
410 }
411 });
412
413 this._connections.push(connection);
414
415 return connection;
416 }
417
418 _shouldRequeuOnConnectionClose(queueEntry) {
419 if (this.options.maxRequeues === undefined || this.options.maxRequeues < 0) {
420 return true;
421 }
422
423 return queueEntry.requeueAttempts < this.options.maxRequeues;
424 }
425
426 _failDeliveryOnConnectionClose(connection) {
427 if (connection.queueEntry && connection.queueEntry.callback) {
428 try {
429 connection.queueEntry.callback(new Error('Reached maximum number of retries after connection was closed'));
430 } catch (E) {
431 this.logger.error(
432 {
433 err: E,
434 tnx: 'callback',
435 messageId: connection.queueEntry.messageId,
436 cid: connection.id
437 },
438 'Callback error for #%s: %s',
439 connection.id,
440 E.message
441 );
442 }
443 connection.queueEntry = false;
444 }
445 }
446
447 _requeueEntryOnConnectionClose(connection) {
448 connection.queueEntry.requeueAttempts = connection.queueEntry.requeueAttempts + 1;
449 this.logger.debug(
450 {
451 tnx: 'pool',
452 cid: connection.id,
453 messageId: connection.queueEntry.messageId,
454 action: 'requeue'
455 },
456 'Re-queued message <%s> for #%s. Attempt: #%s',
457 connection.queueEntry.messageId,
458 connection.id,
459 connection.queueEntry.requeueAttempts
460 );
461 this._queue.unshift(connection.queueEntry);
462 connection.queueEntry = false;
463 }
464
465 /**
466 * Continue to process message if the pool hasn't closed
467 */
468 _continueProcessing() {
469 if (this._closed) {
470 this.close();
471 } else {
472 setTimeout(() => this._processMessages(), 100);
473 }
474 }
475
476 /**
477 * Remove resource from pool
478 *
479 * @param {Object} connection The PoolResource to remove
480 */
481 _removeConnection(connection) {
482 let index = this._connections.indexOf(connection);
483
484 if (index !== -1) {
485 this._connections.splice(index, 1);
486 }
487 }
488
489 /**
490 * Checks if connections have hit current rate limit and if so, queues the availability callback
491 *
492 * @param {Function} callback Callback function to run once rate limiter has been cleared
493 */
494 _checkRateLimit(callback) {
495 if (!this._rateLimit.limit) {
496 return callback();
497 }
498
499 let now = Date.now();
500
501 if (this._rateLimit.counter < this._rateLimit.limit) {
502 return callback();
503 }
504
505 this._rateLimit.waiting.push(callback);
506
507 if (this._rateLimit.checkpoint <= now - this._rateLimit.delta) {
508 return this._clearRateLimit();
509 } else if (!this._rateLimit.timeout) {
510 this._rateLimit.timeout = setTimeout(() => this._clearRateLimit(), this._rateLimit.delta - (now - this._rateLimit.checkpoint));
511 this._rateLimit.checkpoint = now;
512 }
513 }
514
515 /**
516 * Clears current rate limit limitation and runs paused callback
517 */
518 _clearRateLimit() {
519 clearTimeout(this._rateLimit.timeout);
520 this._rateLimit.timeout = null;
521 this._rateLimit.counter = 0;
522 this._rateLimit.checkpoint = false;
523
524 // resume all paused connections
525 while (this._rateLimit.waiting.length) {
526 let cb = this._rateLimit.waiting.shift();
527 setImmediate(cb);
528 }
529 }
530
531 /**
532 * Returns true if there are free slots in the queue
533 */
534 isIdle() {
535 return this.idling;
536 }
537
538 /**
539 * Verifies SMTP configuration
540 *
541 * @param {Function} callback Callback function
542 */
543 verify(callback) {
544 let promise;
545
546 if (!callback) {
547 promise = new Promise((resolve, reject) => {
548 callback = shared.callbackPromise(resolve, reject);
549 });
550 }
551
552 let auth = new PoolResource(this).auth;
553
554 this.getSocket(this.options, (err, socketOptions) => {
555 if (err) {
556 return callback(err);
557 }
558
559 let options = this.options;
560 if (socketOptions && socketOptions.connection) {
561 this.logger.info(
562 {
563 tnx: 'proxy',
564 remoteAddress: socketOptions.connection.remoteAddress,
565 remotePort: socketOptions.connection.remotePort,
566 destHost: options.host || '',
567 destPort: options.port || '',
568 action: 'connected'
569 },
570 'Using proxied socket from %s:%s to %s:%s',
571 socketOptions.connection.remoteAddress,
572 socketOptions.connection.remotePort,
573 options.host || '',
574 options.port || ''
575 );
576 options = shared.assign(false, options);
577 Object.keys(socketOptions).forEach(key => {
578 options[key] = socketOptions[key];
579 });
580 }
581
582 let connection = new SMTPConnection(options);
583 let returned = false;
584
585 connection.once('error', err => {
586 if (returned) {
587 return;
588 }
589 returned = true;
590 connection.close();
591 return callback(err);
592 });
593
594 connection.once('end', () => {
595 if (returned) {
596 return;
597 }
598 returned = true;
599 return callback(new Error('Connection closed'));
600 });
601
602 let finalize = () => {
603 if (returned) {
604 return;
605 }
606 returned = true;
607 connection.quit();
608 return callback(null, true);
609 };
610
611 connection.connect(() => {
612 if (returned) {
613 return;
614 }
615
616 if (auth && (connection.allowsAuth || options.forceAuth)) {
617 connection.login(auth, err => {
618 if (returned) {
619 return;
620 }
621
622 if (err) {
623 returned = true;
624 connection.close();
625 return callback(err);
626 }
627
628 finalize();
629 });
630 } else if (!auth && connection.allowsAuth && options.forceAuth) {
631 let err = new Error('Authentication info was not provided');
632 err.code = 'NoAuth';
633
634 returned = true;
635 connection.close();
636 return callback(err);
637 } else {
638 finalize();
639 }
640 });
641 });
642
643 return promise;
644 }
645}
646
647// expose to the world
648module.exports = SMTPPool;
Note: See TracBrowser for help on using the repository browser.