Buffer failed writes until the socket is writable.

lib/blizzard/session.js

@@ -29,6 +29,8 @@ function BlizzardSession(socket, instigator) {
     this.requestMethods = {};
     this.streams = {};
 
+    this.output = [];
+
     // This ID represents a request-response pair.
     Object.defineProperty(this, "sequence", {
         get: function () {
@@ -86,6 +88,14 @@ BlizzardSession.ERROR_INVALID = -32600;
 BlizzardSession.ERROR_METHOD = -32601;
 BlizzardSession.ERROR_INTERNAL = -32603;
 
+/**
+ * Array of output Buffers.
+ *
+ * @property output
+ * @type {Array}
+ * @private
+ */
+
 /**
  * Bind our events to handler methods.
  *
@@ -119,6 +129,9 @@ BlizzardSession.prototype.bindEvents = function () {
     this.socket.on("end", this.emit.bind(this, "end"));
     this.on("end", this.end.bind(this));
 
+    // Handle socket drain.
+    this.socket.on("drain", this._socketDrain.bind(this));
+
     // Handle RPC calls.
     /**
      * @event rpc.**
@@ -193,6 +206,59 @@ BlizzardSession.prototype.incomingBridge = function (emitter, event, emitterEven
     this.on("request." + event, emitter.emit.bind(emitter, emitterEvent || event));
 };
 
+/**
+ * Drain any buffered data to our socket.
+ *
+ * @method _socketDrain
+ * @private
+ * @param {Buffer} [data] Data to buffer if socket becomes unwritable.
+ * @return {Boolean} True if socket remains writable, false otherwise.
+ */
+BlizzardSession.prototype._socketDrain = function (data) {
+    if (this.socket &&
+        this.socket.writable) {
+        while (this.output.length) {
+            if (!this.socket.writable) {
+                if (data) {
+                    this.output.push(data);
+                }
+                return false;
+            }
+            this.socket.write(this.output.shift());
+        }
+        return true;
+    }
+    return false;
+};
+
+/**
+ * Write to our socket. If the socket is not writable,
+ * the data is buffered until the next _socketWrite
+ * or socket drain event.
+ *
+ * @method _socketWrite
+ * @private
+ * @param {Buffer} data Data to write.
+ * @return {Boolean} True if write was successful, false if buffered.
+ */
+BlizzardSession.prototype._socketWrite = function (data) {
+    if (data.length === 0) {
+        return true;
+    }
+
+    if (this.socket &&
+        this.socket.writable) {
+        if (this._socketDrain(data)) {
+            return this.socket.write(data);
+        } else {
+            return false;
+        }
+    } else {
+        this.output.push(data);
+        return false;
+    }
+};
+
 /**
  * Returns true if the socket is writable.
  *
@@ -221,7 +287,7 @@ BlizzardSession.prototype.xferRawZeroLength = function (type, id) {
     header.writeUInt32BE(id, 2);
     header.writeUInt32BE(0, 6);
 
-    this.socket.write(header);
+    this._socketWrite(header);
 };
 
 /**
@@ -245,7 +311,7 @@ BlizzardSession.prototype.xferRaw = function (type, id, buffer) {
     packet.writeUInt32BE(buffer.length, 6);
     buffer.copy(packet, 10);
 
-    this.socket.write(packet);
+    this._socketWrite(packet);
 };