sql-events-listener

mysql-events

A node.js package that watches a MySQL database and runs callbacks on matched events.

This package is based on the original ZongJi and the original mysql-events modules. Please make sure that you meet the requirements described at ZongJi, like MySQL binlog etc.

Check @kuroski's mysql-events-ui for a mysql-events UI implementation.

Install

npm install @rodrigogs/mysql-events

MySQL configuration:

1. Enabling Binary Logging. Use SHOW BINARY LOGS query to determine whether Binlog has been enabled or not.

From MySQL 8.0, binary logging is enabled by default. Reference

1. Run the following query:

ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'password';
FLUSH PRIVILEGES;

Troubleshooting:

Authentication protocol

• Message: MySQL 8.0 - Client does not support authentication protocol requested by server; consider upgrading MySQL client.
• Solution: Perform the second step in MySQL configuration section.

Quick Start

const mysql = require('mysql');
const MySQLEvents = require('@rodrigogs/mysql-events');
 
const program = async () => {
  const connection = mysql.createConnection({
    host: 'localhost',
    user: 'root',
    password: 'root',
  });
 
  const instance = new MySQLEvents(connection, {
    startAtEnd: true,
    excludedSchemas: {
      mysql: true,
    },
  });
 
  await instance.start();
 
  instance.addTrigger({
    name: 'TEST',
    expression: '*',
    statement: MySQLEvents.STATEMENTS.ALL,
    onEvent: (event) => { // You will receive the events here
      console.log(event);
    },
  });
  
  instance.on(MySQLEvents.EVENTS.CONNECTION_ERROR, console.error);
  instance.on(MySQLEvents.EVENTS.ZONGJI_ERROR, console.error);
};
 
program()
  .then(() => console.log('Waiting for database events...'))
  .catch(console.error);

Check the examples

Usage

### #constructor(connection, options)

• Instantiate and create a database connection using a DSN

  const dsn = {
    host: 'localhost',
    user: 'username',
    password: 'password',
  };
   
  const myInstance = new MySQLEvents(dsn, { /* ZongJi options */ });

• Instantiate and create a database connection using a preexisting connection

  const connection = mysql.createConnection({
    host: 'localhost',
    user: 'username',
    password: 'password',
  });
   
  const myInstance = new MySQLEvents(connection, { /* ZongJi options */ });

• Options(the second argument) is for ZongJi options

  const myInstance = new MySQLEvents({ /* connection */ }, {
    serverId: 3,
    startAtEnd: true,
  });

  See more about ZongJi options

### #start()

• start function ensures that MySQL is connected and ZongJi is running before resolving its promise

  myInstance.start()
    .then(() => console.log('I\'m running!'))
    .catch(err => console.error('Something bad happened', err));

### #stop()

• stop function terminates MySQL connection and stops ZongJi before resolving its promise

  myInstance.stop()
    .then(() => console.log('I\'m stopped!'))
    .catch(err => console.error('Something bad happened', err));

### #pause()

• pause function pauses MySQL connection until #resume() is called, this it useful when you're receiving more data than you can handle at the time

  myInstance.pause();

### #resume()

• resume function resumes a paused MySQL connection, so it starts to generate binlog events again

  myInstance.resume();

### #addTrigger({ name, expression, statement, onEvent })

• Adds a trigger for the given expression/statement and calls the onEvent function when the event happens

  instance.addTrigger({
    name: 'MY_TRIGGER',
    expression: 'MY_SCHEMA.MY_TABLE.MY_COLUMN',
    statement: MySQLEvents.STATEMENTS.INSERT,
    onEvent: async (event) => {
      // Here you will get the events for the given expression/statement.
      // This could be an async function.
      await doSomething(event);
    },
  });

• The name argument must be unique for each expression/statement, it will be user later if you want to remove a trigger

  instance.addTrigger({
    name: 'MY_TRIGGER',
    expression: 'MY_SCHEMA.*',
    statement: MySQLEvents.STATEMENTS.ALL,
    ...
  });
   
  instance.removeTrigger({
    name: 'MY_TRIGGER',
    expression: 'MY_SCHEMA.*',
    statement: MySQLEvents.STATEMENTS.ALL,
  });

• The expression argument is very dynamic, you can replace any step by * to make it wait for any schema, table or column events

  instance.addTrigger({
    name: 'Name updates from table USERS at SCHEMA2',
    expression: 'SCHEMA2.USERS.name',
    ...
  });

  instance.addTrigger({
    name: 'All database events',
    expression: '*',
    ...
  });

  instance.addTrigger({
    name: 'All events from SCHEMA2',
    expression: 'SCHEMA2.*',
    ...
  });

  instance.addTrigger({
    name: 'All database events for table USERS',
    expression: '*.USERS',
    ...
  });

• The statement argument indicates in which database operation an event should be triggered

  instance.addTrigger({
    ...
    statement: MySQLEvents.STATEMENTS.ALL,
    ...
  });

  Allowed statements
• The onEvent argument is a function where the trigger events should be threated

  instance.addTrigger({
    ...
    onEvent: (event) => {
      console.log(event); // { type, schema, table, affectedRows: [], affectedColumns: [], timestamp, }
    },
    ...
  });

### #removeTrigger({ name, expression, statement })

• Removes a trigger from the current instance

  instance.removeTrigger({
    name: 'My previous created trigger',
    expression: '',
    statement: MySQLEvents.STATEMENTS.INSERT,
  });

### Instance events

• MySQLEvents class emits some events related to its MySQL connection and ZongJi instance

  instance.on(MySQLEvents.EVENTS.CONNECTION_ERROR, (err) => console.log('Connection error', err));
  instance.on(MySQLEvents.EVENTS.ZONGJI_ERROR, (err) => console.log('ZongJi error', err));

Available events

Tigger event object

It has the following structure:

{
  type: 'INSERT | UPDATE | DELETE',
  schema: 'SCHEMA_NAME',
  table: 'TABLE_NAME',
  affectedRows: [{
    before: {
      column1: 'A',
      column2: 'B',
      column3: 'C',
      ...
    },
    after: {
      column1: 'D',
      column2: 'E',
      column3: 'F',
      ...
    },
  }],
  affectedColumns: [
    'column1',
    'column2',
    'column3',
  ],
  timestamp: 1530645380029,
  nextPosition: 1343,
  binlogName: 'bin.001',
}

Make sure the database user has the privilege to read the binlog on database that you want to watch on.

LICENSE

BSD-3-Clause © Rodrigo Gomes da Silva