start callback

Author: stz443

lib/MyEmitter.js

@@ -8,8 +8,7 @@ const mysql = require('mysql');
 const EventEmitter = require('events');
 
 
-class MyEmitter extends EventEmitter
-{
+class MyEmitter extends EventEmitter {
 
   /**
    * @param Object conf
@@ -20,29 +19,25 @@ class MyEmitter extends EventEmitter
    * @param Object [conf.binlog.lastTime=0]   - Start emitting events after this time 
    * @param Object [conf.binlog.recoverTimeout=240]
    */
-  constructor(conf)
-  {
+  constructor(conf) {
     super();
 
 
     // Create mysql pool
     this.pool = mysql.createPool(Object.assign({}, (conf.mysql ? conf.mysql : {})));
 
     // Check connection
-    this.pool.getConnection(function(err, conn)
-    {
+    this.pool.getConnection(function(err, conn) {
       if (err) throw err;
-      conn.ping(function(err)
-      {
+      conn.ping(function(err) {
         if (err) throw err;
         conn.release();
       });
     });
 
 
     // Conf
-    conf.binlog = Object.assign(
-      {
+    conf.binlog = Object.assign({
         slaveId: 1,
         lastPos: 0,
         lastTime: 0,
@@ -60,8 +55,7 @@ class MyEmitter extends EventEmitter
       },
       packet: // BinlogPacket
       {
-        last:
-        {
+        last: {
           pos: conf.binlog.lastPos,
           time: conf.binlog.lastTime,
         },
@@ -82,8 +76,7 @@ class MyEmitter extends EventEmitter
     this._binlogRowsEvents = false;
     this._wildcardEvents = false;
 
-    this.on('newListener', function(event, listener)
-    {
+    this.on('newListener', function(event, listener) {
       if (
         typeof event == 'number' &&
         event < Event.ANY
@@ -97,8 +90,7 @@ class MyEmitter extends EventEmitter
         event == BinlogEvent.DELETE_ROWS_EVENT
       ) this._binlogRowsEvents = true;
 
-      if (event == Event.ANY)
-      {
+      if (event == Event.ANY) {
         this._binlogNativeEvents = true;
         this._binlogRowsEvents = true;
         this._wildcardEvents = true;
@@ -113,57 +105,46 @@ class MyEmitter extends EventEmitter
 
 
   /* Shortcuts */
-  get pos()
-  {
+  get pos() {
     return this._conf.packet.last.pos;
   }
-  get time()
-  {
+  get time() {
     return this._conf.packet.last.time;
   }
 
   /**
    * @param Function [cb]
    */
-  start(cb)
-  {
-    this._loadHasChecksum(cb ? cb : (err) => { throw err; });
+  start(cb) {
+    this._loadHasChecksum(cb ? cb : (err) => { if (err) throw err; });
   }
 
-  _loadHasChecksum(cb)
-  {
-    if (this._conf.packet.hasChecksum === null)
-    {
+  _loadHasChecksum(cb) {
+    if (this._conf.packet.hasChecksum === null) {
       this.pool.query(
         'SELECT `VARIABLE_VALUE` FROM `information_schema`.`GLOBAL_VARIABLES` WHERE `VARIABLE_NAME` LIKE "MASTER_VERIFY_CHECKSUM"',
-        function(err, res)
-        {
+        function(err, res) {
           if (err) return cb(err);
 
           this._conf.packet.hasChecksum = (res.length > 0 && res[0].VARIABLE_VALUE == 'ON');
 
           this._connect(cb);
 
         }.bind(this));
-    }
-    else
-    {
+    } else {
       this._connect(cb);
     }
   }
 
-  _connect(cb)
-  {
+  _connect(cb) {
     // Dont connect twice
-    if (this._conn)
-    {
+    if (this._conn) {
       cb(new Error('Already connected'));
       return;
     }
 
     // Get connection
-    this.pool.getConnection(function(err, conn)
-    {
+    this.pool.getConnection(function(err, conn) {
       if (err) return cb(err);
 
       this._conn = conn;
@@ -178,11 +159,9 @@ class MyEmitter extends EventEmitter
       rg.on('unhandledError', this.emit.bind(this, Event.ERROR_COM));
       rg.on('timeout', this.emit.bind(this, Event.ERROR_COM));
 
-      rg.on('end', function()
-      {
+      rg.on('end', function() {
         // Binlog
-        this._bl = new Binlog(
-          {
+        this._bl = new Binlog({
             binlog: this._conf.binlog,
             packet: this._conf.packet
           },
@@ -196,8 +175,7 @@ class MyEmitter extends EventEmitter
         this._bl.on('timeout', this.emit.bind(this, Event.TIMEOUT));
         this._bl.on('end', this.emit.bind(this, Event.END));
 
-        this._bl.on('handshake', function()
-        {
+        this._bl.on('handshake', function() {
           this.emit(Event.CONNECTED);
 
           cb();
@@ -213,18 +191,14 @@ class MyEmitter extends EventEmitter
     }.bind(this));
   }
 
-  _binlog_cb(err, packet)
-  {
-    if (err)
-    {
+  _binlog_cb(err, packet) {
+    if (err) {
       this.emit(Event.ERROR_COM, err);
-    }
-    else if (
+    } else if (
       packet.error ||
       packet.data_error ||
       packet.data_row_errors
-    )
-    {
+    ) {
       let err1 = packet.error;
       let err2 = packet.data_error;
       let err3 = packet.data_row_errors;
@@ -233,41 +207,33 @@ class MyEmitter extends EventEmitter
       delete packet.data_error;
       delete packet.data_row_errors;
 
-      if (err1)
-      {
+      if (err1) {
         this.emit(Event.ERROR_PARSE, err1, packet);
       }
 
-      if (err2)
-      {
+      if (err2) {
         this.emit(Event.ERROR_PARSE_DATA, err2, packet);
       }
 
-      if (err3)
-      {
-        for (let i = 0, l = err3.length; i < l; i++)
-        {
+      if (err3) {
+        for (let i = 0, l = err3.length; i < l; i++) {
           this.emit(Event.ERROR_PARSE_DATA, err3[i], packet);
         }
       }
-    }
-    else
-    {
+    } else {
       packet.skipped ?
         this.emit(Event.SKIP, packet) :
         this.emit(Event.BINLOG, packet);
     }
   }
 
   /** Emit aditional events */
-  emit(type, ...data)
-  {
+  emit(type, ...data) {
     // Super
     super.emit.apply(this, arguments);
 
     // Wildcard Events
-    if (this._wildcardEvents)
-    {
+    if (this._wildcardEvents) {
       super.emit.call(this, Event.ANY, type, ...data);
     }
 
@@ -280,8 +246,7 @@ class MyEmitter extends EventEmitter
       ) &&
       data[0] && // has packet object
       data[0].eventType // has event type
-    )
-    {
+    ) {
       var pck = data[0];
 
       // native rows event
@@ -302,10 +267,8 @@ class MyEmitter extends EventEmitter
           pck.eventType == BinlogEvent.DELETE_ROWS_EVENTv1 ||
           pck.eventType == BinlogEvent.DELETE_ROWS_EVENTv2
         )
-      )
-      {
-        switch (pck.eventType)
-        {
+      ) {
+        switch (pck.eventType) {
           case BinlogEvent.WRITE_ROWS_EVENTv0:
           case BinlogEvent.WRITE_ROWS_EVENTv1:
           case BinlogEvent.WRITE_ROWS_EVENTv2:
@@ -334,17 +297,15 @@ class MyEmitter extends EventEmitter
       }
 
       // native binlog event
-      if (this._binlogNativeEvents)
-      {
+      if (this._binlogNativeEvents) {
         this.emit(pck.eventType, pck);
       }
 
       return;
     }
 
     // Event.ERROR
-    switch (type)
-    {
+    switch (type) {
       case Event.ERROR_SQL:
       case Event.ERROR_COM:
       case Event.ERROR_PARSE:
@@ -356,8 +317,7 @@ class MyEmitter extends EventEmitter
     }
 
     // recover
-    switch (type)
-    {
+    switch (type) {
       case Event.ERROR_SQL:
       case Event.ERROR_COM:
       case Event.ERROR_RECOVER:
@@ -370,24 +330,18 @@ class MyEmitter extends EventEmitter
   }
 
 
-  _disconnect(cb)
-  {
-    try
-    {
+  _disconnect(cb) {
+    try {
       this._bl.removeAllListeners(); // stop recovering
       this._conn.removeAllListeners();
 
       this._bl.end();
 
       this._conn._protocol.quit(); // throws a squence error
       this._conn.destroy(); // maybe the connection is broke
-    }
-    catch (erro)
-    {
+    } catch (erro) {
       // nothing to be done here
-    }
-    finally
-    {
+    } finally {
       this._bl = null;
       this._conn = null;
 
@@ -396,31 +350,24 @@ class MyEmitter extends EventEmitter
     }
   }
 
-  _reconnect(cb)
-  {
+  _reconnect(cb) {
     this.emit(Event.RECONNECTING);
 
-    this._disconnect(function(err)
-    {
+    this._disconnect(function(err) {
       if (err) return cb(err);
 
       this._connect(cb);
 
     }.bind(this));
   }
 
-  _recover()
-  {
+  _recover() {
     this.emit(Event.RECOVERING);
 
-    this._disconnect(function()
-    {
-      setTimeout(function()
-        {
-          this._connect(function(err)
-          {
-            if (err)
-            {
+    this._disconnect(function() {
+      setTimeout(function() {
+          this._connect(function(err) {
+            if (err) {
               this.emit(Event.ERROR_RECOVER, err);
 
               if (!this._conn || !this._bl) this._recover();
@@ -437,19 +384,16 @@ class MyEmitter extends EventEmitter
   /**
    * @param Function [cb]
    */
-  stop(cb)
-  {
+  stop(cb) {
     this._disconnect(cb ? cb : (err) => { throw err; });
   }
 
 
   /**
    * @param Function [cb]
    */
-  restart(cb)
-  {
-    this.stop(function(err)
-    {
+  restart(cb) {
+    this.stop(function(err) {
       if (err) { if (cb) cb(err); return; }
       this.start(cb);
     }.bind(this));