Merge pull request #242 from isocolsky/multiple_emailservices

Multiple emailservices

lib/emailservice.js

@@ -204,6 +204,14 @@ EmailService.prototype.sendEmail = function(notification, cb) {
   self._getRecipientsList(notification, emailType, function(err, recipientsList) {
     if (_.isEmpty(recipientsList)) return cb();
 
+    // TODO: Optimize so one process does not have to wait until all others are done
+    // Instead set a flag somewhere in the db to indicate that this process is free
+    // to serve another request.
+    self.lock.runLocked('email-' + notification.id, cb, function(cb) {
+      self.storage.fetchEmailByNotification(notification.id, function(err, email) {
+        if (err) return cb(err);
+        if (email) return cb();
+
         async.waterfall([
 
           function(next) {
@@ -231,6 +239,7 @@ EmailService.prototype.sendEmail = function(notification, cb) {
                 to: recipient.emailAddress,
                 subject: content.subject,
                 body: content.body,
+                notificationId: notification.id,
               });
               self.storage.storeEmail(email, function(err) {
                 return next(err, email);
@@ -258,6 +267,8 @@ EmailService.prototype.sendEmail = function(notification, cb) {
           return cb(err);
         });
       });
+    });
+  });
 };
 
 module.exports = EmailService;

lib/model/email.js

@@ -24,6 +24,7 @@ Email.create = function(opts) {
   x.status = 'pending';
   x.attempts = 0;
   x.lastAttemptOn = null;
+  x.notificationId = opts.notificationId;
   return x;
 };
 
@@ -41,6 +42,7 @@ Email.fromObj = function(obj) {
   x.status = obj.status;
   x.attempts = obj.attempts;
   x.lastAttemptOn = obj.lastAttemptOn;
+  x.notificationId = obj.notificationId;
   return x;
 };
 

lib/storage.js

@@ -52,6 +52,9 @@ Storage.prototype._createIndexes = function() {
   this.db.collection(collections.ADDRESSES).createIndex({
     address: 1,
   });
+  this.db.collection(collections.EMAIL_QUEUE).createIndex({
+    notificationId: 1,
+  });
 };
 
 Storage.prototype.connect = function(opts, cb) {
@@ -407,6 +410,16 @@ Storage.prototype.fetchUnsentEmails = function(cb) {
   });
 };
 
+Storage.prototype.fetchEmailByNotification = function(notificationId, cb) {
+  this.db.collection(collections.EMAIL_QUEUE).findOne({
+    notificationId: notificationId,
+  }, function(err, result) {
+    if (err) return cb(err);
+    if (!result) return cb();
+
+    return cb(null, Model.Email.fromObj(result));
+  });
+};
 
 Storage.prototype._dump = function(cb, fn) {
   fn = fn || console.log;

test/integration/server.js

@@ -496,6 +496,36 @@ describe('Wallet service', function() {
         });
       });
     });
+
+    it('should support multiple emailservice instances running concurrently', function(done) {
+      var emailService2 = new EmailService();
+      emailService2.start({
+        lock: emailService.lock, // Use same locker service
+        messageBroker: server.messageBroker,
+        storage: storage,
+        mailer: mailerStub,
+        emailOpts: {
+          from: 'bws2@dummy.net',
+          subjectPrefix: '[test wallet 2]',
+        },
+      }, function(err) {
+        helpers.stubUtxos(server, wallet, 1, function() {
+          var txOpts = helpers.createProposalOpts('18PzpUFkFZE8zKWUPvfykkTxmB9oMR8qP7', 0.8, 'some message', TestData.copayers[0].privKey_1H_0);
+          server.createTx(txOpts, function(err, tx) {
+            should.not.exist(err);
+            setTimeout(function() {
+              var calls = mailerStub.sendMail.getCalls();
+              calls.length.should.equal(2);
+              server.storage.fetchUnsentEmails(function(err, unsent) {
+                should.not.exist(err);
+                unsent.should.be.empty;
+                done();
+              });
+            }, 100);
+          });
+        });
+      });
+    });
   });