You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

136 lines
4.0 KiB

1 month ago
'use strict';
const packageData = require('../../package.json');
const shared = require('../shared');
/**
* Generates a Transport object for streaming
*
* Possible options can be the following:
*
* * **buffer** if true, then returns the message as a Buffer object instead of a stream
* * **newline** either 'windows' or 'unix'
*
* @constructor
* @param {Object} optional config parameter
*/
class StreamTransport {
constructor(options) {
options = options || {};
this.options = options || {};
this.name = 'StreamTransport';
this.version = packageData.version;
this.logger = shared.getLogger(this.options, {
component: this.options.component || 'stream-transport'
});
this.winbreak = ['win', 'windows', 'dos', '\r\n'].includes((options.newline || '').toString().toLowerCase());
}
/**
* Compiles a mailcomposer message and forwards it to handler that sends it
*
* @param {Object} emailMessage MailComposer object
* @param {Function} callback Callback function to run when the sending is completed
*/
send(mail, done) {
// We probably need this in the output
mail.message.keepBcc = true;
let envelope = mail.data.envelope || mail.message.getEnvelope();
let messageId = mail.message.messageId();
let recipients = [].concat(envelope.to || []);
if (recipients.length > 3) {
recipients.push('...and ' + recipients.splice(2).length + ' more');
}
this.logger.info(
{
tnx: 'send',
messageId
},
'Sending message %s to <%s> using %s line breaks',
messageId,
recipients.join(', '),
this.winbreak ? '<CR><LF>' : '<LF>'
);
setImmediate(() => {
let stream;
try {
stream = mail.message.createReadStream();
} catch (E) {
this.logger.error(
{
err: E,
tnx: 'send',
messageId
},
'Creating send stream failed for %s. %s',
messageId,
E.message
);
return done(E);
}
if (!this.options.buffer) {
stream.once('error', err => {
this.logger.error(
{
err,
tnx: 'send',
messageId
},
'Failed creating message for %s. %s',
messageId,
err.message
);
});
return done(null, {
envelope: mail.data.envelope || mail.message.getEnvelope(),
messageId,
message: stream
});
}
let chunks = [];
let chunklen = 0;
stream.on('readable', () => {
let chunk;
while ((chunk = stream.read()) !== null) {
chunks.push(chunk);
chunklen += chunk.length;
}
});
stream.once('error', err => {
this.logger.error(
{
err,
tnx: 'send',
messageId
},
'Failed creating message for %s. %s',
messageId,
err.message
);
return done(err);
});
stream.on('end', () =>
done(null, {
envelope: mail.data.envelope || mail.message.getEnvelope(),
messageId,
message: Buffer.concat(chunks, chunklen)
})
);
});
}
}
module.exports = StreamTransport;