added callback to subscribe function that will automatically be called when messages are placed on a given queue

Author: mbonano

lib/stomp.js

@@ -141,7 +141,7 @@ function _setupListeners(stomp) {
         buffer += chunk;
         var frames = buffer.split('\0\n');
 
-    	// Temporary fix : NULL,LF is not a guranteed standard, the LF is optional, so lets deal with it.  (Rauls)
+        // Temporary fix : NULL,LF is not a guranteed standard, the LF is optional, so lets deal with it.  (Rauls)
         if (frames.length == 1) {
             frames = buffer.split('\0');
         }
@@ -269,8 +269,17 @@ Stomp.prototype.handle_new_frame = function(this_frame) {
 
     switch (this_frame.command) {
         case "MESSAGE":
-            if (utils.really_defined(this_frame.headers['message-id']))
+            if (utils.really_defined(this_frame.headers['message-id'])) {
+                // if a subscription to the destination queue exists, fire callback
+                if (this_frame.headers !== null && this_frame.headers.destination !== null && self._subscribed_to[this_frame.headers.destination] !== null) {
+                    var subscription = self._subscribed_to[this_frame.headers.destination];
+                    if (subscription.enabled && subscription.callback !== null && typeof(subscription.callback) == 'function') {
+                        subscription.callback(this_frame.body, this_frame.headers);
+                    }
+                }
                 self.emit('message', this_frame);
+            }
+                
             break;
         case "CONNECTED":
             log.debug('Connected to STOMP');
@@ -299,12 +308,12 @@ Stomp.prototype.disconnect = function() {
  * Subscribe to destination (queue or topic)
  * @param {Object} headers
  */
-Stomp.prototype.subscribe = function(headers) {
+Stomp.prototype.subscribe = function(headers, callback) {
     var self = this;
     destination = headers['destination'];
     headers['session'] = self.session;
     send_command(this, 'SUBSCRIBE', headers);
-    self._subscribed_to[destination] = true;
+    self._subscribed_to[destination] = { enabled: true, callback: callback };
     self.log.debug('subscribed to: ' + destination + ' with headers ' + sys.inspect(headers));
 };
 
@@ -317,7 +326,7 @@ Stomp.prototype.unsubscribe = function(headers) {
     destination = headers['destination'];
     headers['session'] = self.session;
     send_command(this, 'UNSUBSCRIBE', headers);
-    self._subscribed_to[destination] = false;
+    self._subscribed_to[destination].enabled = false;
     self.log.debug('no longer subscribed to: ' + destination);
 };