pull request to update to latest version of quid-proton 0.7

README.md

@@ -1,7 +1,7 @@
 node-qpid
 ============
 
-A Node.js native wrapper around Apache Qpid, specifically the AMQP 1.0 Proton C API
+A Node.js native wrapper around Apache Qpid, specifically the AMQP 1.0 Proton C API.
 
 ## Usage
 
@@ -43,11 +43,42 @@ The module has been preliminarily tested against:
 
 ## Installation
 
-It expects you to already have version 0.3 of the [qpid-proton library](http://qpid.apache.org/proton/) installed on your (Linux) system.
+This module expects that you have the following installed on your system:
+* [qpid-proton library](http://qpid.apache.org/proton/) version 0.7
+* nodejs-dev
+* necessary build tools
 
+Building qpid-proton is fairly involved, so to ease that burden, grabbing and installing 
+the binaries from debian sid is a faster (and probably less error prone) approach.  There
+is a PPA for qpid/released, however the libraries there are not compiled against OpenSSL 
+and SSL support is needed for connecting to Service Bus using AMQP.
+
+For Ubuntu 14.04, you can set up all the necessary requirements as follows:
+
+```bash
+# wget http://ftp.us.debian.org/debian/pool/main/q/qpid-proton/libqpid-proton2-dev_0.7-1_amd64.deb
+# wget http://ftp.us.debian.org/debian/pool/main/q/qpid-proton/libqpid-proton2_0.7-1_amd64.deb
+# sudo dpkg -i libqpid-proton2-dev_0.7-1_amd64.deb libqpid-proton2_0.7-1_amd64.deb
+# sudo apt-get install build-essential git
+# sudo apt-get install nodejs-legacy nodejs-dev nodejs npm
+```
+
+Until this module is committed back to main branch and published as an npm module, to build and install it, do the following:
+
+```bash
+# mkdir tmp
+# cd tmp
+# git clone https://github.com/jmspring/node-qpid.git
+# cd node-qpid
+# sudo npm install -g .
+```
+
+At this point, you are ready to use the module.
 
 ## Issues
 
+Grabbing binaries directly from debian sid for Ubuntu is a little silly, but for now it works.
+
 It's still rough around the edges and not ready for prime time, but pull requests are welcomed!
 
 ## Acknowledgements

binding.gyp

@@ -2,10 +2,50 @@
   "targets": [
     {
       "target_name": "cproton",
-      "sources": [ "src/cproton.cc", "src/messenger.cc" ],
-      "link_settings" : {
-        "libraries": [ '-lqpid-proton' ]
-      }
+      "type": "loadable_module",
+      "sources": [ "src/cproton.cc", "src/sending.cc", "src/messenger.cc", "src/protondata.cc" ],
+
+      'conditions': [
+        ['OS=="linux"', {
+          "link_settings" : {
+            "libraries": [ 
+              '-L/usr/local/qpid-proton/lib',
+              '-lqpid-proton' 
+            ],
+          },
+          "include_dirs": [ 
+            '/usr/include/nodejs/src',
+            '/usr/include/nodejs/deps/uv/include/',
+            '/usr/include/nodejs/deps/v8/include/',
+            '/usr/local/qpid-proton/include'
+          ],
+          "cflags": [
+            "-fPIC"
+          ]
+        }],
+        ['OS=="mac"', {
+          'defines!': [
+            'PLATFORM="mac"',
+          ],
+          'defines': [
+            # we need to use node's preferred "darwin" rather than gyp's preferred "mac"
+            'PLATFORM="darwin"',
+          ],
+          "link_settings" : {
+            "libraries": [ 
+              '-lqpid-proton',
+              '-L/usr/local/qpid-proton/lib64'
+            ],
+          },
+          "include_dirs": [ 
+            '/usr/local/qpid-proton/include', 
+            '/usr/local/node/include/node',
+          ],
+          "cflags": [
+            "-fPIC"
+          ]
+        }]
+      ]
     }
   ]
 }

examples/recv.js

@@ -22,4 +22,4 @@ m.on('message', function(message) {
   console.log(message.body);
 });
 
-m.subscribe(optimist.argv.address).receive();
+m.subscribe(optimist.argv.address, {}).receive();

examples/send.js

@@ -1,4 +1,7 @@
 var Messenger = require('..').proton.Messenger;
+var crypto = require('crypto');
+var async = require('async');
+
 var optimist = require('optimist')
     .options('a', {
         alias : 'address',
@@ -20,8 +23,56 @@ m.on('error', function(err) {
   console.log("Error: " + err.message);
 });
 
-m.send({address: optimist.argv.address, body: message}, function(err) {
-  if (err) {
-    console.log("Error sending message: " + err.message);
+var messages = {};
+var size = 128;
+var count = 10000;
+for(var i = 0; i < count; i++) {
+  messages[i] = {
+    id: i,
+    content: new Buffer(crypto.randomBytes(size)).toString('base64')
+  };
+}
+console.log("Message size -- " + messages[0].content.length);
+
+var processed = 0;
+function watcher() {
+  if(processed == count) {
+    console.log("Total time for %d messages:  %d, errors: %d", count, end - start, errors);
+  } else {
+    console.log("%d of %d processed, errors: %d", processed, count, errors);
+    setTimeout(watcher, 5000);
   }
-});
+}
+
+setTimeout(watcher, 5000);
+
+var sent = count;
+var end;
+var start = new Date().getTime();
+var errors = 0;
+for(var i = 0; i < count; i++) {
+  function send_n(n) {
+    m.send({ address: optimist.argv.address, body: messages[n].content }, function(err, val) {
+      if(!err) {
+        end = new Date().getTime();
+      } else {
+        if(err == "aborted") {
+          m.send(val, function(err, val) {
+            if(err) {
+              errors++;
+              console.log("ERR -- " + n + " : " + err);
+            } else {
+              end = new Date().getTime();
+            }
+          });
+        } else {
+          errors++;
+          console.log("ERR -- " + err);
+        }
+      }
+      processed++;
+    });
+  }
+  send_n(i);
+}
+console.log("yup");

package.json

@@ -9,10 +9,12 @@
     "url": "https://github.com/pofallon/node-qpid.git"
   },
   "devDependencies": {
-    "optimist": "*"
+    "optimist": "*",
+    "nconf": "*",
+    "async": "*"
   },
   "engine": ">= 0.6.13 && < 0.11.0",
-  "version": "0.0.1",
+  "version": "0.0.3",
   "files": [
     "index.js",
     "binding.gyp",

src/async.h

@@ -16,7 +16,7 @@ template <class Item, class Parent> class Async {
 
 protected:
     uv_async_t watcher;
-    NODE_CPROTON_MUTEX_t
+    NODE_CPROTON_MUTEX_t(mutex);
     std::vector<Item*> data;
     Callback callback;
 public:
@@ -26,7 +26,7 @@ template <class Item, class Parent> class Async {
     Async(Parent* parent_, Callback cb_)
         : callback(cb_), parent(parent_) {
         watcher.data = this;
-        NODE_CPROTON_MUTEX_INIT
+        NODE_CPROTON_MUTEX_INIT(mutex)
         uv_async_init(uv_default_loop(), &watcher, listener);
     }
 
@@ -84,7 +84,7 @@ template <class Item, class Parent> class Async {
     }
 
     ~Async() {
-        NODE_CPROTON_MUTEX_DESTROY
+        NODE_CPROTON_MUTEX_DESTROY(mutex)
     }
 };
 

src/macros.h

@@ -48,6 +48,14 @@
     }                                                                          \
     String::Utf8Value var(args[i]->ToString());
 
+#define REQUIRE_ARGUMENT_ARRAY(i, var)                                         \
+    if (args.Length() <= (i) || !args[i]->IsArray()) {                         \
+        return ThrowException(Exception::TypeError(                            \
+            String::New("Argument " #i " must be an array"))                   \
+        );                                                                     \
+    }                                                                          \
+    Local<Array> var = Local<v8::Array>::Cast(args[i])
+
 /* Would really like to know how to write this... :-(
 #define OPTIONAL_ARGUMENT_STRING(i, var)					\
 */
@@ -63,6 +71,13 @@
         var = Local<Function>::Cast(args[i]);                                  \
     }
 
+#define OPTIONAL_ARGUMENT_OBJECT(i, var)						                           \
+    if (args.Length() <= (i) || !args[i]->IsObject()) {				                 \
+        return ThrowException(Exception::TypeError(				                     \
+            String::New("Argument " #i " must be an object"))			             \
+        );									                                                   \
+    }										                                                       \
+    Local<Object> var = args[i]->ToObject();
 
 #define OPTIONAL_ARGUMENT_INTEGER(i, var, default)                             \
     int var;                                                                   \