A couple of bug fixes, but it all works. Move the event emiter logic to a seperate method. Add some more concise examples and update the README. Oh, and hey look, a license

Author: benjaminws

README

@@ -0,0 +1,99 @@
+stomp-js
+========
+
+## Overview
+
+An exercise with node.js to implement the STOMP protocol.
+The design & implementation is heavily inspired by [python-stomp](http://bitbucket.org/benjaminws/python-stomp/), with an evented twist.
+
+This is merely the first iteration; a complete spike.  I will be rewriting it from the ground up with BDD.
+
+## Examples
+
+### Consumer
+
+    #!/usr/bin/env node
+
+    var stomp = require('./lib/stomp');
+
+    var stomp_args = {
+        port: 61613,
+        host: 'localhost',
+        debug: true
+    }
+    // Could also add..
+    //{login: 'bah', password: 'bah'}
+
+    var client = new stomp.Stomp(stomp_args);
+
+    var headers = {
+        destination: '/queue/test_stomp',
+        ack: 'client'
+    };
+
+    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('\nConsumed ' + messages.length + ' messages');
+        client.disconnect();
+    });
+
+### Producer
+
+    #!/usr/bin/env node
+
+    var stomp = require('./lib/stomp');
+
+    var num = process.argv[2];
+
+    var stomp_args = {
+        port: 61613,
+        host: 'localhost',
+        debug: true
+    }
+    // Could also add..
+    //{login: 'bah', password: 'bah'}
+
+    var client = new stomp.Stomp(stomp_args);
+
+    var queue = '/queue/test_stomp';
+
+    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();
+    });

README.md

@@ -2,7 +2,7 @@ var sys = require('sys'),
     utils = require('./stomp-utils'),
     exceptions = require('./stomp-exceptions');
 
-Frame = module.exports = function(logger) {
+function Frame(logger) {
     this.sock = null;
     this.command = null;
     this.headers = null;
@@ -27,7 +27,6 @@ Frame.prototype.stomp_connect = function(client) {
     frame_to_send = this.build_frame(args, true);
     parsed_frame = this.send_frame(frame_to_send);
     this.stomp_log.debug('Connected to STOMP');
-    this.stomp_log.debug('headers: ' + sys.inspect(this.headers));
     if ('session' in this.headers)
         this.session = this.headers['session'];
 
@@ -63,11 +62,10 @@ Frame.prototype.as_string = function() {
         body = this.body;
 
     for (var header in headers) {
-        header_strs.push(header + ':' + headers[header] + '\n');
+        header_strs.push(header + ': ' + headers[header]);
     }
 
     frame = command + '\n' + header_strs.join('\n') + '\n\n' + body + '\x00';
-
     return frame;
 };
 
@@ -103,8 +101,9 @@ Frame.prototype.parse_frame = function(data) {
     args['headers'] = this.headers;
     args['body'] = this.body;
 
-    this_frame = new Frame(this.sock);
-    return_frame = this_frame.build_frame(args);
+    var this_frame = new Frame(this.sock);
+    var return_frame = this_frame.build_frame(args);
+
     return return_frame;
 };
 
@@ -166,8 +165,10 @@ Frame.prototype.get_message = function() {
         if (!this.utils.really_defined(this_frame))
             return null;
         if (this_frame.command == "MESSAGE")
-            return frame;
+            return this_frame;
         else
            this.rqueue.put(this_frame);
     }
 };
+
+module.exports.Frame = Frame;

lib/frame.js

@@ -54,7 +54,7 @@ StompQueue.prototype.get = function() {
 
         if (++this.queue_space * 2 >= this.queue.length) {
             this.queue = this.queue.slice(this.queue_space);
-            console.log(this.queue);
+            console.log("Queue: " + this.queue);
             this.queue_space = 0;
         }
     }
@@ -83,7 +83,7 @@ IntermediateQueue = exports.IntermediateQueue = function() {
 IntermediateQueue.prototype.put = function(frame) {
     if (!"destination" in frame.headers)
         return;
-    console.log(frame.headers)
+    console.log("Queue headers: " + frame.headers)
     this._queue.put(frame);
 }
 

lib/stomp-utils.js

@@ -4,66 +4,78 @@ var net = require('net'),
     utils = require('./stomp-utils'),
     exceptions = require('./stomp-exceptions');
 
-function Stomp(port, host, debug) {
+function Stomp(args) {
     this.socket = null;
-    this.port = port || 61613;
-    this.host = host || "127.0.0.1";
+    this.port = args['port'] || 61613;
+    this.host = args['host'] || "127.0.0.1";
     this.socket_connected = false;
-    this.debug = debug;
-    this.stomp_log = new StompLogging(debug);
-    this.frame_object = new Frame(this.stomp_log);
+    this.debug = args['debug'];
+    this.stomp_log = new StompLogging(this.debug);
+    this.frame_object = new frame.Frame(this.stomp_log);
     this.utils = new StompUtils();
     this.connected_frame = null;
-    this._subscribed_to = [];
-}
+    this._subscribed_to = {};
+};
 
 Stomp.prototype = new process.EventEmitter();
 
 Stomp.prototype.connect = function(headers) {
     var self = this;
     this.stomp_log.debug('Connecting to ' + this.host + ':' + this.port);
     socket = net.createConnection(this.port, this.host);
+    socket.setEncoding('ascii');
     socket.setTimeout(0);
     socket.setNoDelay(true);
 
-    socket.addListener('connect', function () {
+    socket.addListener('connect', function() {
         self.stomp_log.debug('Connected to socket');
         self.connected_frame = self.frame_object.stomp_connect(self.socket);
     });
 
-    socket.addListener('data', function (data) {
-        var reply_frame = self.connected_frame.parse_frame(data)
-        if ('session' in reply_frame.headers)
-            self.connected_frame.session = reply_frame.headers['session'];
-
-        switch (reply_frame.command) {
-            case "CONNECTED":
-                self.emit('connected');
-                break;
-            case "MESSAGE":
-                self.emit('message', reply_frame);
-                break;
-            case "RECEIPT":
-                self.emit('receipt', reply_frame.headers['receipt']);
-                break;
-            case "ERROR":
-                self.emit('error', reply_frame);
-                break;
+    socket.addListener('drain', function() {
+
+    });
+
+    socket.addListener('data', function(data) {
+        var frames = data.split('\0\n');
+        var reply_frame = null;
+        var frame = null;
+        while (frame = frames.shift()) {
+            reply_frame = self.connected_frame.parse_frame(frame + '\0\n')
+            self.handle_new_frame(reply_frame);
         }
     });
-    socket.addListener('end', function () {
+    socket.addListener('end', function() {
         self.stomp_log.debug('goodbye');
     });
-    socket.addListener('error', function (error) {
+    socket.addListener('error', function(error) {
         console.log('error: ' + error.stack);
     });
-    socket.addListener('close', function (error) {
+    socket.addListener('close', function(error) {
         self.stomp_log.debug('disconnected');
     });
 
     this.socket = socket;
 };
 
+Stomp.prototype.handle_new_frame = function(this_frame) {
+
+    switch (this_frame.command) {
+        case "MESSAGE":
+            this.emit('message', this_frame);
+            break;
+        case "CONNECTED":
+            this.emit('connected');
+            break;
+        case "RECEIPT":
+            this.emit('receipt', this_frame.headers['receipt']);
+            break;
+        case "ERROR":
+            this.emit('error', this_frame);
+            break;
+    }
+};
+
 Stomp.prototype.disconnect = function() {
 
     this.socket.end();
@@ -77,7 +89,7 @@ Stomp.prototype.subscribe = function(headers, callback) {
     destination = headers['destination'];
     this._send_command('SUBSCRIBE', headers);
     this._subscribed_to[destination] = callback;
-    this.stomp_log.debug('subscribed to: ' + destination);
+    this.stomp_log.debug('subscribed to: ' + destination + ' with headers ' + sys.inspect(headers));
 
 };
 
@@ -100,23 +112,25 @@ Stomp.prototype.ack = function(message_id) {
 Stomp.prototype.send = function(headers) {
 
     destination = headers['destination'];
+    body = headers['body'] || null;
     headers['want_receipt'] = true;
 
-    return this._send_command('SEND', headers)
+    return this._send_command('SEND', headers, body)
 };
 
-Stomp.prototype._send_command = function(command, headers, extra) {
+Stomp.prototype._send_command = function(command, headers, body) {
 
     if (!this.utils.really_defined(headers))
-        headers = [];
+        headers = {};
 
     frame_conf = {
         'command': command,
-        'headers': headers
+        'headers': headers,
+        'body': body
     };
 
-    this_frame = this.connected_frame.build_frame(frame_conf);
-    reply = this.connected_frame.send_frame(this_frame);
+    var this_frame = this.connected_frame.build_frame(frame_conf);
+    var reply = this.connected_frame.send_frame(this_frame);
 
     if ('want_receipt' in headers)
         return reply;

lib/stomp.js

@@ -2,12 +2,19 @@
 
 var stomp = require('./lib/stomp');
 
-//{'login': 'bah', 'password': 'bah'}
-var client = new stomp.Stomp(61613, 'localhost', true);
+var stomp_args = {
+    port: 61613,
+    host: 'localhost',
+    debug: true
+}
+// Could also add..
+//{login: 'bah', password: 'bah'}
+
+var client = new stomp.Stomp(stomp_args);
 
 var headers = {
-    'destination': '/queue/test',
-    'ack': 'auto'
+    destination: '/queue/test_stomp',
+    ack: 'client'
 };
 
 var messages = [];
@@ -20,14 +27,20 @@ client.on('connected', function() {
 });
 
 client.on('message', function(message) {
+    if (!client.utils.really_defined(message.headers['message-id'])) {
+        console.log(message);
+        return;
+    }
+    client.ack(message.headers['message-id']);
     messages.push(message);
 });
 
 client.on('error', function(error_frame) {
     console.log(error_frame.body);
+    client.disconnect();
 });
 
 process.on('SIGINT', function() {
-    console.log(messages.length);
+    console.log('\nConsumed ' + messages.length + ' messages');
     client.disconnect();
 });