✨ Add a ttl option for auto expiry of deleted messages

mongodb-queue.js

@@ -21,6 +21,10 @@ function now() {
     return (new Date()).toISOString()
 }
 
+function nowAsDate() {
+    return new Date()
+}
+
 function nowPlusSecs(secs) {
     return (new Date(Date.now() + secs * 1000)).toISOString()
 }
@@ -43,6 +47,7 @@ function Queue(mongoDbClient, name, opts) {
     this.col = mongoDbClient.collection(name)
     this.visibility = opts.visibility || 30
     this.delay = opts.delay || 0
+    this.ttl = opts.ttl || null
 
     if ( opts.deadQueue ) {
         this.deadQueue = opts.deadQueue
@@ -57,7 +62,13 @@ Queue.prototype.createIndexes = function(callback) {
         if (err) return callback(err)
         self.col.createIndex({ ack : 1 }, { unique : true, sparse : true }, function(err) {
             if (err) return callback(err)
-            callback(null, indexname)
+            if (!self.ttl) {
+                return callback(null, indexname)
+            }
+            self.col.createIndex({ deleted : 1}, { expireAfterSeconds: self.ttl, background: true}, function(err) {
+                if (err) return callback(err)
+                callback(null, indexname)
+            });
         })
     })
 }
@@ -193,7 +204,7 @@ Queue.prototype.ack = function(ack, callback) {
     }
     var update = {
         $set : {
-            deleted : now(),
+            deleted : nowAsDate(),
         }
     }
     self.col.findOneAndUpdate(query, update, { returnOriginal : false }, function(err, msg, blah) {

test/indexes.js

@@ -6,18 +6,67 @@ var mongoDbQueue = require('../')
 
 setup(function(db) {
 
-    test('visibility: check message is back in queue after 3s', function(t) {
-        t.plan(2)
+    test('indexes: check indexes are created', function(t) {
+        t.plan(7)
 
-        var queue = mongoDbQueue(db, 'visibility', { visibility : 3 })
+        var queue = mongoDbQueue(db, 'indexes')
 
         queue.createIndexes(function(err, indexName) {
             t.ok(!err, 'There was no error when running .ensureIndexes()')
             t.ok(indexName, 'receive indexName we created')
-            t.end()
+
+            var collection = db.collection('indexes');
+            collection.indexInformation({ full : true }, function(err, indexInfo) {
+                t.ok(!err, 'There was no error getting index info')
+
+                /*
+                indexInfo [
+                    {"v":1,"name":"_id_","key":{"_id":1},"ns":"mongodb-queue.indexes"},
+                    {"v":1,"name":"deleted_1_visible_1","key":{"deleted":1,"visible":1},"ns":"mongodb-queue.indexes"},
+                    {"v":1,"name":"ack_1","key":{"ack":1},"unique":true,"ns":"mongodb-queue.indexes","sparse":true}]
+                */
+                t.ok(indexInfo.length === 3, '3 indexes were created')
+                t.ok(indexInfo[0].name === '_id_', 'id index was created')
+                t.ok(indexInfo[1].name === 'deleted_1_visible_1', 'deleted_visible index was created')
+                t.ok(indexInfo[2].name === 'ack_1', 'ack index was created')
+
+                t.end()
+            })
         })
     })
 
+    test('ttl: check index with ttl is created', function(t) {
+        t.plan(9)
+
+        var queue = mongoDbQueue(db, 'ttl', { ttl : 60 })
+
+        queue.createIndexes(function(err, indexName) {
+            t.ok(!err, 'There was no error when running .ensureIndexes()')
+            t.ok(indexName, 'receive indexName we created')
+
+            var collection = db.collection('ttl');
+            collection.indexInformation({ full : true }, function(err, indexInfo) {
+                t.ok(!err, 'There was no error getting index info')
+
+                /*
+                indexInfo [
+                    {"v":1,"name":"_id_","key":{"_id":1},"ns":"mongodb-queue.ttl"},
+                    {"v":1,"name":"deleted_1_visible_1","key":{"deleted":1,"visible":1},"ns":"mongodb-queue.ttl"},
+                    {"v":1,"name":"ack_1","key":{"ack":1},"unique":true,"ns":"mongodb-queue.ttl","sparse":true},
+                    {"v":1,"name":"deleted_1","key":{"deleted":1},"ns":"mongodb-queue.ttl","expireAfterSeconds":60,"background":true}]
+                */
+                t.ok(indexInfo.length === 4, '4 indexes were created')
+                t.ok(indexInfo[0].name === '_id_', 'id index was created')
+                t.ok(indexInfo[1].name === 'deleted_1_visible_1', 'deleted_visible index was created')
+                t.ok(indexInfo[2].name === 'ack_1', 'ack index was created')
+                t.ok(indexInfo[3].name === 'deleted_1', 'ttl index was created')
+                t.ok(indexInfo[3].expireAfterSeconds === 60, 'expireAfterSeconds set')
+
+                t.end()
+            })
+      })
+    })
+
     test('db.close()', function(t) {
         t.pass('db.close()')
         db.close()

test/setup.js

@@ -9,7 +9,7 @@ module.exports = function(callback) {
         // let's empty out some collections to make sure there are no messages
         var collections = [
             'default', 'delay', 'multi', 'visibility', 'clean', 'ping',
-            'stats1', 'stats2',
+            'stats1', 'stats2', 'indexes', 'ttl',
             'queue', 'dead-queue', 'queue-2', 'dead-queue-2'
         ]
         collections.forEach(function(col) {