Minor tidy up

squaremo · squaremo · commit 3b4eeb45846e · 2014-03-21T15:30:11.000Z
diff --git a/lib/sockets.js b/lib/sockets.js
@@ -22,12 +22,21 @@ function bufferify(chunk, encoding) {
     : chunk;
 }
 
+// To avoid errors being swallowed by the promise machinery, I provide
+// an error continuation that will emit the error outside the dynamic
+// extent of the promise.
+function errorLater(obj) {
+  return function(err) {
+    delay(function() {
+      obj.emit('error', err);
+    });
+  }
+}
+
 function Context(url) {
   EventEmitter.call(this);
   var self = this;
-  var onError = function(e) {
-    setImmediate(self.emit.bind(self, 'error', e));
-  }
+  var onError = errorLater(this);
   var onClose = this.emit.bind(this, 'close');
   var c = this._connection = amqp.connect(url);
   c.then(function(conn) {
@@ -101,28 +110,32 @@ function Socket(setup, options) {
   patch(self, ready,
         ['close', 'write', 'end', 'connect',
          'setsockopt', 'ack', 'publish']);
-  // Apply any options we've been given, in case they have immediate
-  // effects rather than just being consulted (e.g., prefetch).
-  ready.then(function() {
-    for (var opt in options) {
-      self.setsockopt(opt, options[opt]);
-    }
-  });
 
   function closeAndInvalidate(event, err) {
     this.readable = this.writable = false;
-    this.emit(event, err);
+    setImmediate(this.emit.bind(this, event, err));
   }
 
   var close = closeAndInvalidate.bind(this, 'close');
   var error = closeAndInvalidate.bind(this, 'error');
 
+  // Relay these events from the channel
   setup.then(function(ch) {
     ch.on('close', close);
     ch.on('error', error);
     ch.on('drain', self.emit.bind(self, 'drain'));
     ch.on('readable', self.emit.bind(self, 'readable'));
   });
+
+  // Apply any options we've been given, in case they have immediate
+  // effects rather than just being consulted (e.g., prefetch).
+  ready.then(function() {
+    for (var opt in options) {
+      self.setsockopt(opt, options[opt]);
+    }
+  }).then(null, errorLater(this));
+  // ^ pick up any setup or setsockopt error and reraise it as an
+  // error event
 }
 
 function close() {
@@ -141,7 +154,7 @@ function setsockopt(opt, value) {
   case 'expiration':
   case 'persistent':
   case 'topic':
-    this.options[opt] = value;
+    this.options[opt] = value; break;
   }
 }
 
diff --git a/test/tests.js b/test/tests.js
@@ -97,9 +97,11 @@ suite.endAwaitsConnects = testWithContext(function(done, CTX) {
   var sock = CTX.socket('REP');
   sock.on('error', done);
   sock.on('close', done);
-  sock.connect('testRepEnd');
-  sock.connect('testRepEnd2');
-  sock.close();
+  sock.connect('testRepEnd', function() {
+    sock.connect('testRepEnd2', function() {
+      sock.close();
+    });
+  });
 });
 
 suite.endWithWrite = testWithContext(function(done, CTX) {
@@ -122,7 +124,8 @@ suite.endWithWrite = testWithContext(function(done, CTX) {
 suite.testManySockets = testWithContext(function(done, CTX) {
   var WINDOW = 10;
   var socks = [];
-  var types = ['PUB', 'SUB', 'PUSH', 'PULL', 'REQ', 'REP'];
+  var types = ['PUB', 'SUB', 'PUSH', 'PULL',
+               'REQ', 'REP', 'WORKER'];
   var total = WINDOW * 10;
   var ended = 0;
   function latch() {
@@ -132,9 +135,9 @@ suite.testManySockets = testWithContext(function(done, CTX) {
   for (var i = 0; i < total; i++) {
     var t = types.shift(); types.push(t);
     var s = CTX.socket(t);
-    s.connect('testManySockets');
     s.on('close', latch);
     socks.push(s);
+    s.connect('testManySockets');
     if (i > WINDOW) {
       socks.shift().close();
     }
@@ -145,14 +148,15 @@ suite.testManySockets = testWithContext(function(done, CTX) {
 suite.simplestPushPull = testWithContext(function(done, CTX) {
     var push = CTX.socket('PUSH');
     var pull = CTX.socket('PULL');
+    var msg = randomString();
     pull.setEncoding('utf8');
-    pull.on('data', function(msg) {
-        assert.equal('foo', msg);
+    pull.on('data', function(m) {
+        assert.equal(msg, m);
         done();
     });
 
     push.connect('testPushPull', function() {
-      push.write('foo');
+      push.write(msg);
       pull.connect('testPushPull');
     });
 });
