src: add 'unix-connect' protocol

- Starting with unix-dgram@0.1.0 there is support for using `connect` in unix
  DGRAM sockets. It allows to have some kind of congestion control via the
  `congestion` and `writable` events.
- This patch adds the `unix-connect` protocol that uses this functionality. While
  in congestion it enquees the messages. Once in writable mode it flushes the
  queue and keeps sending messages as normal.
- It avoids problems like the ones described in
  bnoordhuis/node-unix-dgram#4.

README.md

@@ -41,7 +41,7 @@ In addition to the options accepted by the syslog (compliant with [RFC 3164][1]
 
 * __host:__ The host running syslogd, defaults to localhost.
 * __port:__ The port on the host that syslog is running on, defaults to syslogd's default port.
-* __protocol:__ The network protocol to log over (e.g. `tcp4`, `udp4`, etc).
+* __protocol:__ The network protocol to log over (e.g. `tcp4`, `udp4`, `unix`, `unix-connect`, etc).
 * __pid:__ PID of the process that log messages are coming from (Default `process.pid`).
 * __facility:__ Syslog facility to use (Default: `local0`).
 * __localhost:__ Host to indicate that log messages are coming from (Default: `localhost`).

lib/winston-syslog.js

@@ -39,6 +39,7 @@ var Syslog = exports.Syslog = function (options) {
   // Setup connection state
   //
   this.connected = false;
+  this.congested = false;
   this.retries = 0;
   this.queue = [];
   this.inFlight = 0;
@@ -51,9 +52,9 @@ var Syslog = exports.Syslog = function (options) {
   this.path     = options.path     || null;
   this.app_id   = options.app_id   || null;
   this.protocol = options.protocol || 'udp4';
-  this.isDgram  = /^udp|unix/.test(this.protocol);
+  this.isDgram  = /^udp|^unix/.test(this.protocol);
 
-  if (!/^udp|unix|tcp/.test(this.protocol)) {
+  if (!/^udp|unix|unix-connect|tcp/.test(this.protocol)) {
     throw new Error('Invalid syslog protocol: ' + this.protocol);
   }
 
@@ -154,12 +155,30 @@ Syslog.prototype.log = function (level, msg, meta, callback) {
         self.inFlight++;
         self.socket.send(buffer, 0, buffer.length, self.port, self.host, onError);
       }
-      else {
+      else if (self.protocol === 'unix') {
         self.socket.send(buffer, 0, buffer.length, self.path, onError);
       }
+      else {
+        if (self.congested) {
+          self.queue.push(syslogMsg)
+        } else {
+          function on_congestion() {
+            onError(new Error('Congestion Error'));
+          }
+
+          self.socket.once('congestion', on_congestion);
+          self.socket.once('error', onError);
+          self.socket.send(buffer, function() {
+            self.socket.removeListener('congestion', on_congestion);
+            self.socket.removeListener('error', onError);
+            onError();
+          });
+        }
+      }
     }
     else {
       self.socket.write(syslogMsg, 'utf8', onError);
+
     }
   });
 
@@ -208,9 +227,12 @@ Syslog.prototype.connect = function (callback) {
     if (self.protocol.match(/^udp/)) {
       this.socket = new dgram.Socket(this.protocol);
     }
-    else {
+    else if (self.protocol === 'unix') {
       this.socket = new unix.createSocket('unix_dgram');
     }
+    else {
+      return this._unix_dgram_connect(callback);
+    }
 
     return callback(null);
   }
@@ -281,3 +303,41 @@ Syslog.prototype.connect = function (callback) {
 
   this.socket.connect(this.port, this.host);
 };
+
+Syslog.prototype._unix_dgram_connect = function(cb) {
+  var self = this;
+
+  function flush_queue() {
+    var sent_msgs = 0;
+    self.queue.forEach(function(msg) {
+      var buffer = new Buffer(msg);
+      if (!self.congested) {
+        self.socket.send(buffer, function() {
+          ++ sent_msgs;
+        });
+      }
+    });
+
+    self.queue.splice(0, sent_msgs);
+  }
+
+  this.socket = new unix.createSocket('unix_dgram');
+  this.socket.on('error', function(err) {
+    self.emit('error', err);
+  });
+
+  this.socket.on('connect', function() {
+    this.on('congestion', function() {
+      self.congested = true;
+    });
+
+    this.on('writable', function() {
+      self.congested = false;
+      flush_queue();
+    });
+
+    cb();
+  });
+
+  this.socket.connect(this.path);
+};

package.json

@@ -13,7 +13,7 @@
   "keywords": ["logging", "sysadmin", "tools", "winston", "syslog"],
   "dependencies": {
     "glossy": "0.x.x",
-    "unix-dgram": ">= 0.0.1"
+    "unix-dgram": "~0.1.1"
   },
   "devDependencies": {
     "winston": "0.5.x",