From 20fa1b6f5bebb13526270c0327f42378dbce6ae1 Mon Sep 17 00:00:00 2001
From: Jeremy Darling <jeremy.darling@gmail.com>
Date: Wed, 16 Jan 2013 19:42:58 -0600
Subject: [PATCH] Staging changes for onAny and Monitor

---
 bin/MongoMQ.js      |  2 +-
 lib/MongoMQ.js      | 18 +++++++++++++++-
 lib/QueueMonitor.js | 52 +++++++++++++++++++++++++++++++++------------
 3 files changed, 56 insertions(+), 16 deletions(-)

diff --git a/bin/MongoMQ.js b/bin/MongoMQ.js
index 0fa8b12..64977cf 100644
--- a/bin/MongoMQ.js
+++ b/bin/MongoMQ.js
@@ -1,7 +1,7 @@
 var MongoMQ = require('../lib/MongoMQ');
 var repl = require('repl');
 
-var l = process.argv.length, tmp, names, reCmdLineStrip=/^(\-|\\|\/)*/i, opts = {},
+var l = process.argv.length, tmp, names, reCmdLineStrip=/^(\-|\\|\/)*/i, opts = {autoStart: false},
   alias = {
     db: 'databaseName',
     collection: 'queueCollection',
diff --git a/lib/MongoMQ.js b/lib/MongoMQ.js
index 47f1c43..5fdc56c 100644
--- a/lib/MongoMQ.js
+++ b/lib/MongoMQ.js
@@ -7,7 +7,8 @@ var QueueMonitor = require('./QueueMonitor');
 
 var defaults = {
   autoStart: true,
-  queueCollection: 'queue'
+  queueCollection: 'queue',
+  databaseName: 'mongomq'
 };
 
 var MongoMQ = module.exports = function(options, callback){
@@ -267,6 +268,21 @@ MongoMQ.prototype.stopListeners = function(callback){
 };
 MongoMQ.prototype.stopListeners.description = 'Stops the queue listeners that have not yet been stopped.';
 
+
+MongoMQ.prototype.onAny = function(options, callback){
+  var self = this;
+  if(typeof(options)==='function'){
+    callback = options;
+    options = {};
+  }
+  options=options||{};
+  options.listenerType = options.listenerType||'streams';
+  options.hereOnOut = typeof(options.hereOnOut)!=='undefined'?options.hereOnOut:true;
+  options.passive = true;
+  self.on('*', options, callback);
+};
+MongoMQ.prototype.onAny.description = 'Listens for any message on the message queue in a passive mode.';
+
 MongoMQ.prototype.on = MongoMQ.prototype.addListener = function(event, options, callback){
   var self = this;
   var list = self.monitors[event]=self.monitors[event]||[], i, l=list.length, index=false;
diff --git a/lib/QueueMonitor.js b/lib/QueueMonitor.js
index c2aa134..8fdccb8 100644
--- a/lib/QueueMonitor.js
+++ b/lib/QueueMonitor.js
@@ -35,13 +35,32 @@ var QueueMonitor = module.exports = function(mq, event, options, callback){
   self.start();
 };
 
+QueueMonitor.prototype.buildSelector = function(cursorFixId){
+  var self = this, event = self.event, dt = new Date(), afterDt = new Date(dt-self.mq.serverTimeOffset), $or, i, l;
+  if(event!=='*'){
+    $or = [{handled: false, event: event}, {handled: {$exists: false}, event: event}, {fix: cursorFixId}];
+  }else{
+    $or = [{handled: false}, {handled: {$exists: false}, fix: {$exists: false}}, {fix: cursorFixId}];
+  }
+  if(self.options.hereOnOut){
+    l = $or.length;
+    for(i=0; i<l; i++){
+      if($or[i].fix!==cursorFixId){
+        $or[i].globalTime = {$gt: afterDt};
+      }
+    }
+  }
+  return {$or: $or};
+};
+QueueMonitor.prototype.buildSelector.description = 'Used internally by start_usingNextObject and start_usingStreams to build the message selector.';
+
 QueueMonitor.prototype.start_usingNextObject = function(startupCallback){
   // call out to setup the tailed cursor for us
   var self = this, cursorFixId = self.cursorFixId = UUID.v4(), mq = self.mq, callback = self.callback, event = self.event;
   // this is a work around so we don't get runaway handles when there is no records for us to process
   mq.collection(mq.options.queueCollection, function(err, collection){
     collection.insert({fix: cursorFixId}, {w: 1}, function(){
-      mq.tailedCursor(mq.options.queueCollection, {$or: [{handled: false, event: event}, {handled: {$exists: false}, event: event}, {fix: cursorFixId}]}, {$natural: 1}, function(err, cursor){
+      mq.tailedCursor(mq.options.queueCollection, self.buildSelector(cursorFixId), {$natural: 1}, function(err, cursor){
         if(err){
           self.errorCallback(err);
           startupCallback(err);
@@ -63,7 +82,7 @@ QueueMonitor.prototype.start_usingStreams = function(startupCallback){
   mq.collection(mq.options.queueCollection, function(err, collection){
     if(collection){
       collection.insert({fix: cursorFixId}, {w: 1}, function(){
-        mq.tailedCursorStream(mq.options.queueCollection, {$or: [{handled: false, event: event}, {handled: {$exists: false}, event: event}, {fix: cursorFixId}]}, {$natural: 1}, function(err, stream){
+        mq.tailedCursorStream(mq.options.queueCollection, self.buildSelector(cursorFixId), {$natural: 1}, function(err, stream){
           if(!stream){
             self.start_usingStreams(startupCallback);
           }else{
@@ -104,7 +123,7 @@ QueueMonitor.prototype.start = function(callback){
       case('streams'):
         self.start_usingStreams(started);
         break;
-      case('nextObject'):
+      case('nextobject'):
       default:
         self.start_usingNextObject(started);
     }
@@ -169,18 +188,23 @@ QueueMonitor.prototype.handleResponse = function(err, record, doNext){
             var dt = new Date();
             set.$set.pickedTime = new Date(dt-mq.serverTimeOffset)
           }
-          collection.findAndModify(record, {$natural: -1}, set, {w: ((mq.db||{}).serverConfig||{}).replicaSet?self.options.writeConcern||mq.db.serverConfig.servers.length:1}, function(err, data){
-            if(data||err){
-              if(err){
-                self.errorCallback(err);
+          if(options.passive){
+            mq.emitter.emit('recieved', data);
+            callback(err, (data||{}).data, next);
+          }else{
+            collection.findAndModify(record, {$natural: -1}, set, {w: ((mq.db||{}).serverConfig||{}).replicaSet?self.options.writeConcern||mq.db.serverConfig.servers.length:1}, function(err, data){
+              if(data||err){
+                if(err){
+                  self.errorCallback(err);
+                }
+                mq.emitter.emit('recieved', data);
+                callback(err, (data||{}).data, next);
+              }else{
+                // We didn't get the lock on the record
+                next();
               }
-              mq.emitter.emit('recieved', data);
-              callback(err, (data||{}).data, next);
-            }else{
-              // We didn't get the lock on the record
-              next();
-            }
-          });
+            });
+          }
         }
       });
     }