pool-resource.js 7.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252
  1. 'use strict';
  2. const SMTPConnection = require('../smtp-connection');
  3. const assign = require('../shared').assign;
  4. const XOAuth2 = require('../xoauth2');
  5. const EventEmitter = require('events');
  6. /**
  7. * Creates an element for the pool
  8. *
  9. * @constructor
  10. * @param {Object} options SMTPPool instance
  11. */
  12. class PoolResource extends EventEmitter {
  13. constructor(pool) {
  14. super();
  15. this.pool = pool;
  16. this.options = pool.options;
  17. this.logger = this.pool.logger;
  18. if (this.options.auth) {
  19. switch ((this.options.auth.type || '').toString().toUpperCase()) {
  20. case 'OAUTH2': {
  21. let oauth2 = new XOAuth2(this.options.auth, this.logger);
  22. oauth2.provisionCallback = (this.pool.mailer && this.pool.mailer.get('oauth2_provision_cb')) || oauth2.provisionCallback;
  23. this.auth = {
  24. type: 'OAUTH2',
  25. user: this.options.auth.user,
  26. oauth2,
  27. method: 'XOAUTH2'
  28. };
  29. oauth2.on('token', token => this.pool.mailer.emit('token', token));
  30. oauth2.on('error', err => this.emit('error', err));
  31. break;
  32. }
  33. default:
  34. if (!this.options.auth.user && !this.options.auth.pass) {
  35. break;
  36. }
  37. this.auth = {
  38. type: 'LOGIN',
  39. user: this.options.auth.user,
  40. credentials: {
  41. user: this.options.auth.user || '',
  42. pass: this.options.auth.pass
  43. },
  44. method: (this.options.auth.method || '').trim().toUpperCase() || false
  45. };
  46. }
  47. }
  48. this._connection = false;
  49. this._connected = false;
  50. this.messages = 0;
  51. this.available = true;
  52. }
  53. /**
  54. * Initiates a connection to the SMTP server
  55. *
  56. * @param {Function} callback Callback function to run once the connection is established or failed
  57. */
  58. connect(callback) {
  59. this.pool.getSocket(this.options, (err, socketOptions) => {
  60. if (err) {
  61. return callback(err);
  62. }
  63. let returned = false;
  64. let options = this.options;
  65. if (socketOptions && socketOptions.connection) {
  66. this.logger.info(
  67. {
  68. tnx: 'proxy',
  69. remoteAddress: socketOptions.connection.remoteAddress,
  70. remotePort: socketOptions.connection.remotePort,
  71. destHost: options.host || '',
  72. destPort: options.port || '',
  73. action: 'connected'
  74. },
  75. 'Using proxied socket from %s:%s to %s:%s',
  76. socketOptions.connection.remoteAddress,
  77. socketOptions.connection.remotePort,
  78. options.host || '',
  79. options.port || ''
  80. );
  81. options = assign(false, options);
  82. Object.keys(socketOptions).forEach(key => {
  83. options[key] = socketOptions[key];
  84. });
  85. }
  86. this.connection = new SMTPConnection(options);
  87. this.connection.once('error', err => {
  88. this.emit('error', err);
  89. if (returned) {
  90. return;
  91. }
  92. returned = true;
  93. return callback(err);
  94. });
  95. this.connection.once('end', () => {
  96. this.close();
  97. if (returned) {
  98. return;
  99. }
  100. returned = true;
  101. let timer = setTimeout(() => {
  102. if (returned) {
  103. return;
  104. }
  105. // still have not returned, this means we have an unexpected connection close
  106. let err = new Error('Unexpected socket close');
  107. if (this.connection && this.connection._socket && this.connection._socket.upgrading) {
  108. // starttls connection errors
  109. err.code = 'ETLS';
  110. }
  111. callback(err);
  112. }, 1000);
  113. try {
  114. timer.unref();
  115. } catch (E) {
  116. // Ignore. Happens on envs with non-node timer implementation
  117. }
  118. });
  119. this.connection.connect(() => {
  120. if (returned) {
  121. return;
  122. }
  123. if (this.auth) {
  124. this.connection.login(this.auth, err => {
  125. if (returned) {
  126. return;
  127. }
  128. returned = true;
  129. if (err) {
  130. this.connection.close();
  131. this.emit('error', err);
  132. return callback(err);
  133. }
  134. this._connected = true;
  135. callback(null, true);
  136. });
  137. } else {
  138. returned = true;
  139. this._connected = true;
  140. return callback(null, true);
  141. }
  142. });
  143. });
  144. }
  145. /**
  146. * Sends an e-mail to be sent using the selected settings
  147. *
  148. * @param {Object} mail Mail object
  149. * @param {Function} callback Callback function
  150. */
  151. send(mail, callback) {
  152. if (!this._connected) {
  153. return this.connect(err => {
  154. if (err) {
  155. return callback(err);
  156. }
  157. return this.send(mail, callback);
  158. });
  159. }
  160. let envelope = mail.message.getEnvelope();
  161. let messageId = mail.message.messageId();
  162. let recipients = [].concat(envelope.to || []);
  163. if (recipients.length > 3) {
  164. recipients.push('...and ' + recipients.splice(2).length + ' more');
  165. }
  166. this.logger.info(
  167. {
  168. tnx: 'send',
  169. messageId,
  170. cid: this.id
  171. },
  172. 'Sending message %s using #%s to <%s>',
  173. messageId,
  174. this.id,
  175. recipients.join(', ')
  176. );
  177. if (mail.data.dsn) {
  178. envelope.dsn = mail.data.dsn;
  179. }
  180. this.connection.send(envelope, mail.message.createReadStream(), (err, info) => {
  181. this.messages++;
  182. if (err) {
  183. this.connection.close();
  184. this.emit('error', err);
  185. return callback(err);
  186. }
  187. info.envelope = {
  188. from: envelope.from,
  189. to: envelope.to
  190. };
  191. info.messageId = messageId;
  192. setImmediate(() => {
  193. let err;
  194. if (this.messages >= this.options.maxMessages) {
  195. err = new Error('Resource exhausted');
  196. err.code = 'EMAXLIMIT';
  197. this.connection.close();
  198. this.emit('error', err);
  199. } else {
  200. this.pool._checkRateLimit(() => {
  201. this.available = true;
  202. this.emit('available');
  203. });
  204. }
  205. });
  206. callback(null, info);
  207. });
  208. }
  209. /**
  210. * Closes the connection
  211. */
  212. close() {
  213. this._connected = false;
  214. if (this.auth && this.auth.oauth2) {
  215. this.auth.oauth2.removeAllListeners();
  216. }
  217. if (this.connection) {
  218. this.connection.close();
  219. }
  220. this.emit('close');
  221. }
  222. }
  223. module.exports = PoolResource;