cleaner plugin: removeUpTo

lib/MessageDb.js

@@ -86,7 +86,7 @@ MessageDb.prototype.authenticate = function(m) {
 MessageDb.parseKey = function(key) {
   var ret = {};
   var spl = key.split('-');
-  
+
   ret.to = spl[1];
   ret.ts = +spl[2];
 
@@ -124,14 +124,39 @@ MessageDb.prototype.getMessages = function(to, lower_ts, upper_ts, cb) {
     });
 };
 
-MessageDb.prototype.removeUpTo = function(ts) {
-  preconditions.checkArgument(ts);
-  preconditions.checkArgument(typeof ts === 'number');
-  var opts = {};
-  db.createKeyStream(opts)
-    .on('data', function(key) {
-      console.log('key=', key)
+MessageDb.prototype.getAll = function(cb) {
+  var list = [];
+  db.createReadStream()
+    .on('data', function(data) {
+      list.push(MessageDb.fromStorage(data));
+    })
+    .on('end', function() {
+      return cb(null, list);
     });
 };
 
+MessageDb.prototype.removeUpTo = function(ts, cb) {
+  preconditions.checkArgument(ts);
+  preconditions.checkArgument(typeof ts === 'number');
+  var opts = {};
+  var dels = [];
+  db.createKeyStream(opts)
+    .on('data', function(key) {
+      var parsed = MessageDb.parseKey(key);
+      if (parsed.ts < ts) {
+        dels.push({
+          type: 'del',
+          key: key
+        });
+      }
+    })
+    .on('end', function() {
+      db.batch(dels, function(err) {
+        if (err) return cb(err);
+        else cb(null, dels.length);
+      })
+    });
+
+};
+
 module.exports = soop(MessageDb);

test/test.MessageDb.js

@@ -108,9 +108,22 @@ describe('MessageDb', function() {
       done();
     });
   });
-  it('should be able #removeUpTo', function() {
+  it('should be able #removeUpTo', function(done) {
     var mdb = sharedMDB;
-    mdb.removeUpTo(microtime.now());
+    var upper_ts = microtime.now();
+    mdb.addMessage(message, function(err) {
+      expect(err).to.not.exist;
+      mdb.removeUpTo(upper_ts, function(err, n) {
+        expect(err).to.not.exist;
+        n.should.equal(4);
+        mdb.getAll(function(error, all) {
+          expect(error).to.not.exist;
+          all.length.should.equal(1);
+          done();
+        });
+
+      });
+    });
   });
   it('should be able to close instance', function() {
     var mdb = new MessageDb(opts);