Add the .clean() operation

Author: chilts

README.md

@@ -52,9 +52,19 @@ queue.ping(msg.ack, function(err, id) {
 Ack a message (and remove it from the queue):
 
 ```js
-queue.ack(msg.ack, function(err) {
-    // This msg removed from queue for this message id.
-    // 'id' is returned, useful for logging.
+queue.ack(msg.ack, function(err, id) {
+    // This msg removed from queue for this ack.
+    // The 'id' of the message is returned, useful for logging.
+})
+```
+
+By default, all old messages - even processed ones - are left in MongoDB. This is so that
+you can go and analyse them if you want. However, you can call the following function
+to remove processed messages:
+
+```js
+queue.clean(function(err) {
+    // All processed (ie. acked) messages have been deleted
 })
 ```
 
@@ -333,10 +343,17 @@ or deleted. We always use MongoDB's excellent `collection.findAndModify()` so th
 each message is updated atomically inside MongoDB and we never have to fetch something,
 change it and store it back.
 
+## Note on MongoDB Version ##
+
+When using MongoDB v2.6 and the v1.3.23 version of the mongodb driver from npm, I was getting
+a weird error similar to "key $exists must not start with '$'". Yes, very strange. Anyway, the fix
+is to install a later version of the driver. I have tried this with v1.4.9 and it seems ok.
+
 ## Releases ##
 
-### 0.8.0 (not yet released) ###
+### 0.8.0 (2014-08-28) ###
 
+* [NEW] Added .clean() method to remove old (processed) messages
 * [NEW] Add 'delay' option to queue.add() so individual messages can be delayed separately
 * [TEST] Test individual 'delay' option for each message
 

mongodb-queue.js

@@ -177,6 +177,16 @@ Queue.prototype.ack = function(ack, callback) {
     })
 }
 
+Queue.prototype.clean = function(callback) {
+    var self = this
+
+    var query = {
+        deleted : { $exists : true },
+    }
+
+    self.col.remove(query, callback)
+}
+
 Queue.prototype.total = function(callback) {
     var self = this
 

package.json

@@ -15,9 +15,9 @@
   },
   "dependencies": {},
   "devDependencies": {
-    "tape": "^2.10.2",
-    "mongodb": "^1.3.23",
-    "async": "^0.2.10"
+    "tape": "^2.14.0",
+    "mongodb": "^1.4.9",
+    "async": "^0.9.0"
   },
   "repository": {
     "type": "git",

test/clean.js

@@ -0,0 +1,172 @@
+var async = require('async')
+var test = require('tape')
+
+var setup = require('./setup.js')
+var mongoDbQueue = require('../')
+
+setup(function(db) {
+
+    test('clean: check deleted messages are deleted', function(t) {
+        var queue = mongoDbQueue(db, 'clean', { visibility : 3 })
+        var msg
+
+        async.series(
+            [
+                function(next) {
+                    queue.size(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 0, 'There is currently nothing on the queue')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.total(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 0, 'There is currently nothing in the queue at all')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.clean(function(err) {
+                        t.ok(!err, 'There is no error.')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.size(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 0, 'There is currently nothing on the queue')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.total(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 0, 'There is currently nothing in the queue at all')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.add('Hello, World!', function(err) {
+                        t.ok(!err, 'There is no error when adding a message.')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.clean(function(err) {
+                        t.ok(!err, 'There is no error.')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.size(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 1, 'Queue size is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.total(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 1, 'Queue total is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.get(function(err, newMsg) {
+                        msg = newMsg
+                        t.ok(msg.id, 'Got a msg.id (sanity check)')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.size(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 0, 'Queue size is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.total(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 1, 'Queue total is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.clean(function(err) {
+                        t.ok(!err, 'There is no error.')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.size(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 0, 'Queue size is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.total(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 1, 'Queue total is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.ack(msg.ack, function(err, id) {
+                        t.ok(!err, 'No error when acking the message')
+                        t.ok(id, 'Received an id when acking this message')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.size(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 0, 'Queue size is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.total(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 1, 'Queue total is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.clean(function(err) {
+                        t.ok(!err, 'There is no error.')
+                        next()
+                    })
+                },
+                function(next) {
+                    queue.size(function(err, size) {
+                        t.ok(!err, 'There is no error.')
+                        t.equal(size, 0, 'Queue size is correct')
+                        next()
+                    })
+                },
+                function(next) {
+                  queue.total(function(err, size) {
+                    t.ok(!err, 'There is no error.')
+                    t.equal(size, 0, 'Queue total is correct')
+                    next()
+                  })
+                },
+            ],
+            function(err) {
+                if (err) t.fail(err)
+                t.pass('Finished test ok')
+                t.end()
+            }
+        )
+    })
+
+    test('db.close()', function(t) {
+        t.pass('db.close()')
+        db.close()
+        t.end()
+    })
+
+})

test/setup.js

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