Oh hey, let's start working on this again! Functioning producer.

Author: benjaminws

lib/frame.js

@@ -3,56 +3,67 @@ 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;
     this.body = null;
+    this.reply = null;
     this.session = null;
     this.stomp_log = logger;
     this.rqueue = new StompQueue();
     this.iqueue = new IntermediateQueue();
+    this.utils = new StompUtils();
 };
 
 Frame.prototype.stomp_connect = function(client) {
-    this.sock = client;
-    var args = {};
-    var headers = {};
-    var next_frame = null;
+    var self = this,
+        args = {},
+        headers = {}
+        next_frame = null;
 
+    this.sock = client;
     args['command'] = 'CONNECT';
     args['headers'] = headers;
-    frame_to_send = this.build_frame(args);
-    this.send_frame(frame_to_send);
+    frame_to_send = this.build_frame(args, true);
+    parsed_frame = this.send_frame(frame_to_send);
     this.stomp_log.debug('Connected to STOMP');
-    next_frame = this.get_reply();
-    this.session = next_frame.headers;
+    this.stomp_log.debug('headers: ' + sys.inspect(this.headers));
+    if ('session' in this.headers)
+        this.session = this.headers['session'];
+
     return this;
 };
 
 Frame.prototype.build_frame = function(args, want_receipt) {
+    var self = this,
+        receipt_stamp = null;
+
     this.command = args['command'];
     this.headers = args['headers'];
     this.body = args['body'];
-    var receipt_stamp = null;
 
     if (want_receipt) {
         receipt_stamp = Math.floor(Math.random()*10000000+1);
-        this.headers['receipt'] = this.session['session'] + '-' + receipt_stamp;
-        this.stomp_log.debug(want_receipt);
+        if (this.session != null) {
+            this.headers['receipt'] = receipt_stamp + "-" + this.session;
+        }
+        else {
+            this.headers['receipt'] = receipt_stamp + "-";
+        }
     }
-
     return this;
 };
 
 Frame.prototype.as_string = function() {
-    var header_strs = Array(),
+    var self = this,
+        header_strs = Array(),
         frame = null,
         command = this.command,
         headers = this.headers,
         body = this.body;
 
     for (var header in headers) {
-        this.stomp_log.debug(header);
         header_strs.push(header + ':' + headers[header] + '\n');
     }
 
@@ -62,6 +73,7 @@ Frame.prototype.as_string = function() {
 };
 
 Frame.prototype.send_frame = function(frame) {
+    self = this;
     this.sock.write(frame.as_string());
 
     if ('receipt' in frame.headers)
@@ -70,11 +82,12 @@ Frame.prototype.send_frame = function(frame) {
 };
 
 Frame.prototype.parse_frame = function(data) {
-    var args = [],
+    var self = this,
+        args = [],
         headers_str = null;
 
-    if (data == null || data == undefined)
-        return;
+    if (!this.utils.really_defined(data))
+        return null;
 
     this.command = this.parse_command(data);
     var _data = data.slice(this.command.length + 1, data.length);
@@ -95,20 +108,23 @@ Frame.prototype.parse_frame = function(data) {
     args['body'] = this.body;
 
     this_frame = new Frame(this.sock);
-    return this_frame.build_frame(args);
+    return_frame = this_frame.build_frame(args);
+    return return_frame;
 };
 
 Frame.prototype.parse_headers = function(headers_str) {
-    var these_headers = Array(),
-        one_header = Array(),
+    var these_headers = new Array(),
+        one_header = new Array(),
+        header_key = null,
+        header_val = null,
         headers_split = headers_str.split('\n');
 
+
     for (var i = 0; i < headers_split.length; i++) {
         one_header = headers_split[i].split(':');
-
         if (one_header.length > 1) {
-            var header_key = one_header.shift();
-            var header_val = one_header.join(':');
+            header_key = one_header.shift();
+            header_val = one_header.join(':');
             these_headers[header_key] = header_val;
         }
         else {
@@ -126,32 +142,36 @@ Frame.prototype.parse_command = function(data) {
 };
 
 Frame.prototype.get_reply = function() {
+    var self = this;
+
     while (true) {
         try {
-            return this.rqueue.get()
+            return self.rqueue.get()
         }
         catch (error) {
             if (error.name == "QueueEmpty") {
-                frame = this.parse_frame();
-                if (frame == null || frame == undefined)
+                this_frame = self.reply;
+                if (!self.utils.really_defined(this_frame))
                     return null;
-                if (frame.command == "MESSAGE")
-                    this.iqueue.put(frame);
+                if (this_frame.command == "MESSAGE")
+                    return self.iqueue.put(this_frame);
                 else
-                    this.rqueue.put(frame);
+                    return self.rqueue.put(this_frame);
             }
         }
     }
 };
 
 Frame.prototype.get_message = function() {
+    var self = this;
+
     while (true) {
-        frame = this.rqueue.get()
-        if (frame == null || frame == undefined)
+        this_frame = this.rqueue.get()
+        if (!this.utils.really_defined(this_frame))
             return null;
-        if (frame.command == "MESSAGE")
+        if (this_frame.command == "MESSAGE")
             return frame;
         else
-           this.rqueue.put(frame);
+           this.rqueue.put(this_frame);
     }
 };

lib/stomp-utils.js

@@ -9,6 +9,21 @@ StompLogging.prototype.debug = function(message) {
         console.log("debug: " + message);
 };
 
+StompUtils = exports.StompUtils = function() {
+    this.available_utils = [];
+};
+
+StompUtils.prototype.really_defined = function(var_to_test) {
+    return !(var_to_test == null || var_to_test == undefined);
+};
+
+// Extend associative array
+StompUtils.prototype.extend = function(destination, source) {
+    for (var property in source)
+        destination[property] = source[property];
+    return destination;
+};
+
 /**
  * Queueing implementation, blatantly ripped off from:
  * http://safalra.com/web-design/javascript/queues/Queue.js
@@ -68,6 +83,7 @@ IntermediateQueue = exports.IntermediateQueue = function() {
 IntermediateQueue.prototype.put = function(frame) {
     if (!"destination" in frame.headers)
         return;
+    console.log(frame.headers)
     this._queue.put(frame);
 }
 

lib/stomp.js

@@ -1,37 +1,110 @@
 var net = require('net'),
+    sys = require('sys'),
     frame = require('./frame'),
     utils = require('./stomp-utils'),
     exceptions = require('./stomp-exceptions');
 
-Stomp = module.exports = function(port, host, debug) {
-    this.port = port;
-    this.host = host;
+function Stomp(port, host, debug, conf) {
+    this.socket = null;
+    this.port = port || 61613;
+    this.host = host || "127.0.0.1";
     this.socket_connected = false;
     this.debug = debug;
     this.stomp_log = new StompLogging(debug);
-    this.frame = new Frame(this.stomp_log);
+    this.frame_object = new Frame(this.stomp_log);
+    this.utils = new StompUtils();
     this.connected_frame = null;
-};
+    this._subscribed_to = {};
+}
+
+Stomp.prototype = new process.EventEmitter();
 
-Stomp.prototype.connect = function() {
+Stomp.prototype.connect = function(headers) {
+    var self = this;
     this.stomp_log.debug('Connecting to ' + this.host + ':' + this.port);
-    var client = net.createConnection(this.port, this.host);
-    var _stomp = this;
+    socket = net.createConnection(this.port, this.host);
+    socket.setTimeout(0);
+    socket.setNoDelay(true);
 
-    client.addListener('connect', function () {
-        _stomp.stomp_log.debug('Connected to socket');
-        _stomp.connected_frame = _stomp.frame.stomp_connect(client);
+    socket.addListener('connect', function () {
+        self.stomp_log.debug('Connected to socket');
+        self.connected_frame = self.frame_object.stomp_connect(self.socket);
     });
-    client.addListener('data', function (data) {
-        _stomp.connected_frame.parse_frame(data);
+
+    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');
+                break;
+            case "RECEIPT":
+                self.emit('receipt');
+                break;
+            case "ERROR":
+                self.emit('error');
+                break;
+        }
     });
-    client.addListener('end', function () {
-        _stomp.stomp_log.debug('goodbye');
+    socket.addListener('end', function () {
+        self.stomp_log.debug('goodbye');
     });
-    client.addListener('error', function (error) {
+    socket.addListener('error', function (error) {
         console.log('error: ' + error.stack);
     });
-    client.addListener('close', function (error) {
-        _stomp.stomp_log.debug('disconnected');
+    socket.addListener('close', function (error) {
+        self.stomp_log.debug('disconnected');
     });
+
+    this.socket = socket;
+};
+
+
+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);
+
+};
+
+Stomp.prototype.unsubscribe = function(headers) {
+
+    destination = headers['destination'];
+    this._send_command('UNSUBSCRIBE', headers);
+    this._subscribed_to[destination] = null;
+    this.stomp_log.debug('no longer subscribed to: ' + destination);
+
 };
+
+Stomp.prototype.send = function(headers) {
+
+    destination = headers['destination'];
+    headers['want_receipt'] = true;
+
+    return this._send_command('SEND', headers)
+};
+
+Stomp.prototype._send_command = function(command, headers, extra) {
+
+    if (!this.utils.really_defined(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);
+
+    if ('want_receipt' in headers)
+        return reply;
+
+    return this_frame;
+
+};
+
+module.exports.Stomp = Stomp;

stomp-client.js

@@ -1,7 +1,23 @@
 #!/usr/bin/env node
 
-var Stomp = require('./lib/stomp');
+var stomp = require('./lib/stomp');
 
-var client = new Stomp(61613, 'localhost', true);
+var client = new stomp.Stomp(61613, 'localhost', true, {'login': 'bah', 'password': 'bah'});
+
+var queue = '/queue/test';
+
+var headers = {'destination': queue, 'ack': 'client'};
 
 client.connect();
+
+client.on('connected', function() {
+    client.subscribe(headers);
+
+    for (var i = 0; i < 10; i++) {
+        console.log(i);
+        client.send({'destination': queue,
+                     'body': 'Testing' + i,
+                     'persistent': 'true'
+        });
+    }
+});