Refactor Manager - Connection responsibilities to be clearer

lib/connection.js

@@ -3,100 +3,46 @@ const net = require('net');
 const fs = require('fs');
 const tls = require('tls');
 
-const ECONNREFUSED_REGEXP = /ECONNREFUSED/;
-
 /**
  * Represents a connection to Logstash.
  * @constructor
  * @param {object} options
- * @param {function} onErrorHook
+ * @param {object} manager
  */
 class Connection {
-  constructor(options, onErrorHook) {
-    this.onErrorHook = onErrorHook;
+  constructor(options, manager) {
+    this.manager = manager;
     this.host = options.host || '127.0.0.1';
     this.port = options.port || 28777;
-
-    // Connection state flags
-    this.connecting = false;
-    this.terminating = false;
-    this.connected = false;
-
-    // Connection retry attributes
-    this.tryReconnect = true;
-    this.retries = -1;
-    this.max_connect_retries =
-      ('number' === typeof options.max_connect_retries) ?
-        options.max_connect_retries : 4;
-    this.timeout_connect_retries =
-      ('number' === typeof options.timeout_connect_retries) ?
-        options.timeout_connect_retries : 100;
   }
 
   socketOnError(error) {
-    this.connecting = false;
-    this.connected = false;
-
-    if (typeof (this.socket) !== 'undefined' && this.socket != null) {
-      this.socket.destroy();
-    }
-
-    if (!ECONNREFUSED_REGEXP.test(error.message)) {
-      this.tryReconnect = false;
-      this.onErrorHook(error);
-    }
+    this.manager.emit('connection:error', error);
   }
   socketOnTimeout() {
     if (this.socket.readyState !== 'open') {
       this.socket.destroy();
     }
   }
   socketOnConnect() {
-    this.retries = 0;
-    this.connecting = false;
-    this.connected = true;
     this.socket.setKeepAlive(true, 60 * 1000);
-    this.actionOnConnect();
+    this.manager.emit('connection:connect');
   }
 
   socketOnClose(error) {
-    this.connected = false;
-
-    if (this.terminating) {
-      return;
-    }
-
-    if (this.max_connect_retries < 0 ||
-      this.retries < this.max_connect_retries) {
-      if (!this.connecting) {
-        setTimeout(function() {
-          this.connect();
-        }.bind(this), this.timeout_connect_retries);
-      }
-    } else {
-      this.onErrorHook(
-          new Error('Max retries reached, transport in silent mode, OFFLINE'));
-    }
+    this.manager.emit('connection:close', error);
   }
-  bindCommonEventListeners(socket) {
+
+  addEventListeners(socket) {
     socket.on('error', this.socketOnError.bind(this));
     socket.on('timeout', this.socketOnTimeout.bind(this));
     socket.on('close', this.socketOnClose.bind(this));
   }
 
-  connect(onConnection) {
-    this.retries++;
-    this.connecting = true;
-    this.terminating = false;
-    this.actionOnConnect = onConnection || function() { };
-  }
-
   close() {
-    this.terminating = true;
     if (this.socket) {
       this.socket.end();
       this.socket.destroy();
-      this.connected = false;
     }
   }
 
@@ -110,18 +56,17 @@ class Connection {
 }
 
 class PlainConnection extends Connection {
-  connect(onConnection) {
-    super.connect(onConnection);
+  connect() {
     this.socket = new net.Socket();
     this.socket.connect(this.port, this.host);
-    super.bindCommonEventListeners(this.socket);
+    super.addEventListeners(this.socket);
     this.socket.on('connect', super.socketOnConnect.bind(this));
   }
 }
 
 class SecureConnection extends Connection {
-  constructor(options, onErrorHook) {
-    super(options, onErrorHook);
+  constructor(options, manager) {
+    super(options, manager);
     // SSL Settings
     this.ssl_enable = options.ssl_enable || false;
     this.secureContextOptions = this.ssl_enable ?
@@ -154,12 +99,11 @@ class SecureConnection extends Connection {
     return secureContextOptions;
   }
 
-  connect(onConnection) {
-    super.connect(onConnection);
+  connect() {
     this.socket = tls.connect(this.port,
         this.host,
         this.secureContextOptions);
-    super.bindCommonEventListeners(this.socket);
+    super.addEventListeners(this.socket);
     this.socket.on('secureConnect', super.socketOnConnect.bind(this));
   }
 }

lib/manager.js

@@ -1,23 +1,82 @@
 /* eslint-disable require-jsdoc */
 const Connection = require('./connection');
+const EventEmitter = require('events');
+const ECONNREFUSED_REGEXP = /ECONNREFUSED/;
 
-module.exports = class Manager {
-  constructor(options, onError) {
+module.exports = class Manager extends EventEmitter {
+  constructor(options) {
+    super();
     this.host = options.host || '127.0.0.1';
     this.port = options.port || 28777;
     this.connectionStarted = false;
+    this.closing = false;
     this.logQueue = [];
 
+    // Connection retry attributes
+    this.tryReconnect = true;
+    this.retries = -1;
+    this.max_connect_retries =
+      ('number' === typeof options.max_connect_retries) ?
+        options.max_connect_retries : 4;
+    this.timeout_connect_retries =
+      ('number' === typeof options.timeout_connect_retries) ?
+        options.timeout_connect_retries : 100;
+
+    this.on('connection:connect', this.onConnection.bind(this));
+    this.on('connection:close', this.onConnectionClose.bind(this));
+    this.on('connection:error', this.onConnectionError.bind(this));
+
     if (options.ssl_enable) {
-      this.connection = new Connection.SecureConnection(options, onError);
+      this.connection = new Connection.SecureConnection(options, this);
     } else {
-      this.connection = new Connection.PlainConnection(options, onError);
+      this.connection = new Connection.PlainConnection(options, this);
+    }
+  }
+
+  onConnection() {
+    this.retries = 0;
+    this.flush();
+  }
+
+  onConnectionClose() {
+    if (this.tryReconnect === true) {
+      this.connection.connect(this.flush.bind(this));
+    }
+  }
+
+  isRetryableError(error) {
+    // TODO: Due bug in the orginal implementation
+    //       all the errors will get retried
+    return true; // !ECONNREFUSED_REGEXP.test(error.code);
+  }
+
+  tryToReconnect(error) {
+    if (this.isRetryableError(error) === true &&
+      this.closing === false) {
+      if (this.max_connect_retries < 0 ||
+      this.retries < this.max_connect_retries) {
+        return true;
+      } else {
+        return false;
+      }
+    }
+  }
+
+  onConnectionError(error) {
+    this.retries++;
+    this.tryReconnect = this.tryToReconnect(error);
+
+    if (this.tryReconnect === false) {
+      this.emit('error',
+          new Error('Max retries reached, transport in silent mode, OFFLINE'));
+      this.closing = true;
+      this.connection.close();
     }
   }
 
   start() {
     if (!this.connectionStarted) {
-      this.connection.connect(this.flush.bind(this));
+      this.connection.connect();
       this.connectionStarted = true;
     }
   }
@@ -29,6 +88,7 @@ module.exports = class Manager {
 
   close() {
     this.flush();
+    this.closing = true;
     this.connection.close();
   }
 

lib/winston-logstash-latest.js

@@ -12,7 +12,8 @@ module.exports = class LogstashTransport extends Transport {
 
     this.name = 'logstash';
 
-    this.manager = new Manager(options, this.onError.bind(this));
+    this.manager = new Manager(options);
+    this.manager.on('error', this.onError.bind(this));
     this.manager.start();
   }
 
@@ -27,9 +28,12 @@ module.exports = class LogstashTransport extends Transport {
     });
 
     // Perform the writing to the remote service
-
     this.manager.log(JSON.stringify(info), () => {
       callback();
     });
   }
+
+  close() {
+    this.manager.close();
+  }
 };

lib/winston-logstash.js

@@ -32,7 +32,8 @@ class Logstash extends winston.Transport {
     this.meta_defaults = Object.assign({}, options.meta);
     this.transform = options.transform || defaultTransform;
 
-    this.manager = new Manager(options, this.onError.bind(this));
+    this.manager = new Manager(options);
+    this.manager.on('error', this.onError.bind(this));
     this.manager.start();
   }
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "winston-logstash",
-  "version": "1.0.0",
+  "version": "1.0.1",
   "description": "A Logstash transport for winston",
   "main": "./lib/winston-logstash",
   "homepage": "https://github.com/jaakkos/winston-logstash",