A bit of cleanup for readability and standardization. Add a few basic methods to stomp. Don't throw exceptions, instead emit an error event (cleaner and more node like..).

Author: benjaminws

lib/frame.js

@@ -3,7 +3,6 @@ var sys = require('sys'),
     exceptions = require('./stomp-exceptions');
 
 Frame = module.exports = function(logger) {
-    this.me = Math.floor(Math.random()*10000000+1);
     this.sock = null;
     this.command = null;
     this.headers = null;
@@ -44,20 +43,20 @@ Frame.prototype.build_frame = function(args, want_receipt) {
     this.body = args['body'];
 
     if (want_receipt) {
-        receipt_stamp = Math.floor(Math.random()*10000000+1);
+        receipt_stamp = Math.floor(Math.random()*99999999999).toString();
         if (this.session != null) {
             this.headers['receipt'] = receipt_stamp + "-" + this.session;
         }
         else {
-            this.headers['receipt'] = receipt_stamp + "-";
+            this.headers['receipt'] = receipt_stamp;
         }
     }
     return this;
 };
 
 Frame.prototype.as_string = function() {
     var self = this,
-        header_strs = Array(),
+        header_strs = [],
         frame = null,
         command = this.command,
         headers = this.headers,
@@ -83,7 +82,7 @@ Frame.prototype.send_frame = function(frame) {
 
 Frame.prototype.parse_frame = function(data) {
     var self = this,
-        args = [],
+        args = {},
         headers_str = null;
 
     if (!this.utils.really_defined(data))
@@ -100,9 +99,6 @@ Frame.prototype.parse_frame = function(data) {
     if ('content-length' in this.headers)
         this.headers['bytes_message'] = true;
 
-    if (this.command == 'ERROR')
-        throw new BrokerErrorResponse(this.body);
-
     args['command'] = this.command;
     args['headers'] = this.headers;
     args['body'] = this.body;
@@ -113,8 +109,8 @@ Frame.prototype.parse_frame = function(data) {
 };
 
 Frame.prototype.parse_headers = function(headers_str) {
-    var these_headers = new Array(),
-        one_header = new Array(),
+    var these_headers = [],
+        one_header = [],
         header_key = null,
         header_val = null,
         headers_split = headers_str.split('\n');

lib/stomp-exceptions.js

@@ -1,12 +1,3 @@
-BrokerErrorResponse = exports.BrokerErrorResponse = function(value) {
-    this.value = value;
-    this.message = "Broker returned error: " + this.value;
-};
-
-BrokerErrorResponse.prototype.toString = function() {
-    return this.messge + this.value;
-};
-
 QueueEmpty = exports.QueueEmpty = function() {
     this.name = "QueueEmpty";
     this.message = "Queue is Empty";

lib/stomp.js

@@ -4,7 +4,7 @@ var net = require('net'),
     utils = require('./stomp-utils'),
     exceptions = require('./stomp-exceptions');
 
-function Stomp(port, host, debug, conf) {
+function Stomp(port, host, debug) {
     this.socket = null;
     this.port = port || 61613;
     this.host = host || "127.0.0.1";
@@ -14,7 +14,7 @@ function Stomp(port, host, debug, conf) {
     this.frame_object = new Frame(this.stomp_log);
     this.utils = new StompUtils();
     this.connected_frame = null;
-    this._subscribed_to = {};
+    this._subscribed_to = [];
 }
 
 Stomp.prototype = new process.EventEmitter();
@@ -64,6 +64,13 @@ Stomp.prototype.connect = function(headers) {
     this.socket = socket;
 };
 
+Stomp.prototype.disconnect = function() {
+
+    this.socket.end();
+    if (this.socket.readyState == 'readOnly')
+        this.socket.destroy();
+
+}
 
 Stomp.prototype.subscribe = function(headers, callback) {
 
@@ -83,6 +90,13 @@ Stomp.prototype.unsubscribe = function(headers) {
 
 };
 
+Stomp.prototype.ack = function(message_id) {
+
+    this._send_command('ACK', {'message-id': message_id});
+    this.stomp_log.debug('acknowledged message: ' + message_id);
+
+};
+
 Stomp.prototype.send = function(headers) {
 
     destination = headers['destination'];
@@ -94,9 +108,13 @@ Stomp.prototype.send = function(headers) {
 Stomp.prototype._send_command = function(command, headers, extra) {
 
     if (!this.utils.really_defined(headers))
-        headers = {};
+        headers = [];
+
+    frame_conf = {
+        'command': command,
+        'headers': headers
+    };
 
-    frame_conf = {'command': command, 'headers': headers};
     this_frame = this.connected_frame.build_frame(frame_conf);
     reply = this.connected_frame.send_frame(this_frame);
 

stomp-client.js

@@ -0,0 +1,33 @@
+#!/usr/bin/env node
+
+var stomp = require('./lib/stomp');
+
+//{'login': 'bah', 'password': 'bah'}
+var client = new stomp.Stomp(61613, 'localhost', true);
+
+var headers = {
+    'destination': '/queue/test',
+    'ack': 'auto'
+};
+
+var messages = [];
+
+client.connect();
+
+client.on('connected', function() {
+    client.subscribe(headers);
+    console.log('Connected');
+});
+
+client.on('message', function(message) {
+    messages.push(message);
+});
+
+client.on('error', function(error_frame) {
+    console.log(error_frame.body);
+});
+
+process.on('SIGINT', function() {
+    console.log(messages.length);
+    client.disconnect();
+});

stomp-consumer.js

@@ -0,0 +1,35 @@
+#!/usr/bin/env node
+
+var stomp = require('./lib/stomp');
+
+var num = process.argv[2];
+
+//{'login': 'bah', 'password': 'bah'}
+var client = new stomp.Stomp(61613, 'localhost', true);
+
+var queue = '/queue/test';
+
+client.connect();
+
+client.on('connected', function() {
+    if (!num) num = 10;
+
+    for (var i = 0; i < num; i++) {
+        console.log(i);
+        client.send({
+            'destination': queue,
+            'body': 'Testing' + i,
+            'persistent': 'true'
+        });
+    }
+    client.disconnect();
+});
+
+client.on('error', function(error_frame) {
+    console.log(error_frame.body);
+    client.disconnect();
+});
+
+process.on('SIGINT', function() {
+    client.disconnect();
+});