mysql-binlog-emitter

A nodejs mysql/mariadb slave replication event emitter.

This is a high level implementation of mysql's binary log protocol, if your are only interested in changes occurring in your database, check out mysql-event-emitter.

* It is tested against mariadb 10.1.31 and binlog protocol version 4 *

Contents

• Install
• Setup
• Usage
• Options
• Packet
• Examples
• Credits

Install

npm install mysql-binlog-emitter

Setup

Mysql Server

Enable binary log replication in /etc/mysql/my.cnf

[mysqld]
server-id        = 1
log_bin          = /var/log/mysql/mysql-bin
log_bin_index    = /var/log/mysql/mysql-bin.index
binlog-format    = row                              # needed for row events
# Optional
expire_logs_days = 10
max_binlog_size  = 100M

Mysql User

Give your user the rights to read binary logs

GRANT REPLICATION SLAVE, REPLICATION CLIENT, SELECT ON *.* TO '[USER]'@'[HOST]'

Reset Binary Log

After changing the mysql configuration you may have to run RESET MASTER once.
To do that your user needs the RELOAD right

GRANT RELOAD ON *.* TO '[USER]'@'[HOST]'

Usage

const MyBinlogEmitter = require('mysql-binlog-emitter');
const Event = MyBinlogEmitter.Events;

var mbe =  MyBinlogEmitter({
  "mysql": {
    "user": "[USER]",
    "password": "[PWD]"
  }
});

mbe.on(Event.ANY, function(type, ...data){
  console.log(Event.eventName(type));
}):

mbe.start();

Options

• For a single instance no binlog option is needed
• Any mysql connection and pool option is supported.

Defaults

{  
  // mysql pool options
  "mysql": {
    "[socket]": "[SOCKET]",
    "[host]":   "127.0.0.1",
    "user":     "[USER]",
    "password": "[PWD]",
    "[database]": null        // (is not required)
  }
  // binlog options
  "binlog": {
    "slaveId": 1,             // Needs to be counted up, if more than one instance is running  
    "lastPos": 0,             // Skip all events < lastPos
    "lastTime": 0,            // Timstamp; Skip all events < lastTime 
    "recoverTimeout": 240,    // Time in ms between reconnection attempts. (Eg. on a mysql server restart)
    "hasChecksum": null,      // Auto detected; Boolean; Whether the server uses a checksum table
    "version": 0,             // Auto detected; (Only version 4 is supported right now)
  }
}

Events

No | Name | Data | . --- | ----------------- | ----------------- | --- 100 | ANY | EventType, Data | Wildcard event; Emitts all events incl. BinlogEvents. 101 | CONNECTED | | 102 | DISCONNECTED | | 103 | RECONNECTING | | 104 | RECOVERING | | 110 | BINLOG | Packet | Emits all BinlogEvents, except skipped one's 111 | SKIP | Packet | Emits a packet w/out data, when a packet is skipped 120 | TIMEOUT | | 121 | END | | 400 | ERROR | Error | Any Error 401 | ERROR_SQL | Error | SQL Error 402 | ERROR_COM | Error | SQL Com Error 403 | ERROR_PARSE | Error, Packet | Binlog Parse Error 404 | ERROR_PARSE_DATA | Error, Packet | Binlog Parse Data Error 405 | ERROR_RECOVER | Error | Recover Error; (Will not stop the emitter from recovering)

BinlogEvents

As described in MySQL Internals Manual > Binlog Event Types

No | Name --- | ---- 0 | UNKNOWN_EVENT 1 | START_EVENT_V3 2 | QUERY_EVENT 3 | STOP_EVENT 4 | ROTATE_EVENT 5 | INTVAR_EVENT 6 | LOAD_EVENT 7 | SLAVE_EVENT 8 | CREATE_FILE_EVENT 9 | APPEND_BLOCK_EVENT 10 | EXEC_LOAD_EVENT 11 | DELETE_FILE_EVENT 12 | NEW_LOAD_EVENT 13 | RAND_EVENT 14 | USER_VAR_EVENT 15 | FORMAT_DESCRIPTION_EVENT 16 | XID_EVENT 17 | BEGIN_LOAD_QUERY_EVENT 18 | EXECUTE_LOAD_QUERY_EVENT 19 | TABLE_MAP_EVENT 21 | WRITE_ROWS_EVENTv0 22 | UPDATE_ROWS_EVENTv0 23 | DELETE_ROWS_EVENTv0 24 | WRITE_ROWS_EVENTv1 25 | UPDATE_ROWS_EVENTv1 26 | DELETE_ROWS_EVENTv1 27 | INCIDENT_EVENT 28 | HEARTBEAT_EVENT 29 | IGNORABLE_EVENT 30 | ROWS_QUERY_EVENT 31 | WRITE_ROWS_EVENTv2 32 | UPDATE_ROWS_EVENTv2 33 | DELETE_ROWS_EVENTv2 34 | GTID_EVENT 35 | ANONYMOUS_GTID_EVENT 36 | PREVIOUS_GTIDS_EVENT

Additional wildcard events

No | Name | Description --- | ---- | --- 37 | ROWS_EVENT | Emitted on WRITE/UPDATE/DELETE-ROWS_EVENT 38 | WRITE_ROWS_EVENT | Emitted on WRITE_ROWS_EVENT/v0/v1/v2; Changes the packet.eventType 39 | UPDATE_ROWS_EVENT | Emitted on UPDATE_ROWS_EVENT/v0/v1/v2; Changes the packet.eventType 40 | DELETE_ROWS_EVENT | Emitted on DELETE_ROWS_EVENT/v0/v1/v2; Changes the packet.eventType

Packet

Packets consist of an event header packet and an event body packet.data.

BinlogPacket {                        // Packet Head
  timestamp: [UNIX Timestamp w/ ms],  // Log Time
  eventType: [Integer],               // BinlogEvent Type
  serverId: [Integer],                // Slave ID
  eventSize: [Integer],
  flags: [Integer],
  logPos: [Integer],                  // Log Position
  version: [Integer],                 // Protocol Version
  skipped: [Boolean],                 // Whether the packet was skipped
  data: [Object]                      // Packet Body
}

(If packet.data is null the parser for it is not implemented yet)
(Row events do not contain any data yet.)

Examples

const MyBinlogEmitter = require('mysql-binlog-emitter');
const Event = MyBinlogEmitter.Events;
const BinlogEvent = MyBinlogEmitter.BinlogEvents;

var mbe =  MyEmitter({
  "mysql": {
    "user": "[USER]",
    "password": "[PWD]"
  }
});

mbe.on(BinlogEvent.WRITE_ROWS_EVENTv1, function(packet){
  
  console.log(Event.eventName(packet.eventType)); // Event.eventName includes BinlogEvents
  
  console.log(BinlogEvent.eventName(packet.eventType)); // BinlogEvent.eventName includes BinlogEvents only
  
  console.log(packet); 

}):

mye.start(function(err){
  console.log('started');
});

// get last log position
var pos = mye.pos;

// get last log time
var time = mye.time;

// gracefully restart
// packets will be skipped until last pos | time
mye.restart(function(err){
  console.log('restarted');
});

mye.stop(function(err){
  console.log('stopped');
});

Credits

mysql
Faking a slave: Subscribing to mysql row-based-replication changes
Dive into MySQL replication protocol and go-mysql
ZongJi