rabbitmq_plugin plugin

← Firebird streaming / CDC

The rabbitmq_plugin plugin is designed to register events in the Firebird database and send them to the RabbitMQ queue server.

Each event is a JSON document.

The event contains the following information:

event type (INSERT, UPDATE, DELETE, SET SEQUENCE, EXECUTE SQL);
for INSERT, UPDATE, DELETE events:
- the name of the table for which this event occurred;
- the number of the transaction in which this event occurred;
- old and new values of the table record fields;
- an array of changed fields for the UPDATE event;
- new values of BLOB fields;
for the SET SEQUENCE event:
- sequence (generator) name;
- sequence value;
for the EXECUTE SQL event:
- the number of the transaction in which this event occurred;
- SQL query text.

Only new field values are present in the INSERT event. The UPDATE event contains both old and new field values. Only the old field values are present in the DELETE event.

The EXECUTE SQL event occurs only for DDL queries.

Note: for BLOB fields, the BlobId is always stored. If the configuration parameter is dumpBlobs, then in addition to this for events INSERT and UPDATE will publish new BLOB field values. For old field values, this is not possible because old field BLOB values are not present in the replication log files.

The algorithm of the rabbitmq_plugin plugin

Each event is packaged in JSON format and sent to the RabbitMQ queue server exchange specified in the exchange_name parameter. Depending on the exchanger type, the event is either sent to all queues associated with the exchanger, or to the queue bound by the routing_key routing key. If the queue_name parameter is specified, then the queue with the given name is declared and bound by the routing_key key to the exchanger exchange_name.

the fb_streaming service analyzes the replication log directory set for the task for unprocessed files;
for unprocessed replication files, the following events are analyzed:
- start of transaction. Allocates a structure in memory to store savepoints;
- create a savepoint. A structure in memory is allocated for storing documents;
- release the savepoint. Documents saved in this savepoint are copied to a higher-level savepoint;
- rollback of the savepoint (the savepoint is destroyed along with all documents);
- commit the transaction. Documents created in the INSERT, UPDATE, DELETE and EXECUTE SQL events are sent to the RabbitMQ exchanger;
- transaction rollback. All savepoints and documents stored in them are destroyed;
- DDL query execution (EXECUTE SQL). A new document is created and saved to the list of documents for the savepoint. The statement text, event type, and transaction number are stored in the document;
- setting a new sequence value (SET SEQUENCE). A new document is created and immediately sent to the RabbitMQ exchange. The sequence name, event type, and new sequence value are stored in the document;
- INSERT, UPDATE and DELETE events. The table name is checked in the include_tables and exclude_tables filters, and if the table name satisfies the filtering criteria, then a new document is created and saved to the list of documents for the savepoint. The document saves the table name, event type, transaction number, old and new field values.
after processing the file, its number is fixed in the control file.

Other replication events are ignored.

Setting up the rabbitmq_plugin plugin

First of all, you need to set up asynchronous replication for your database, for this you need to add the following lines to the replication.conf file:

database = d:\fbdata\4.0\horses.fdb
{
    journal_directory = d:\fbdata\4.0\replication\horses\journal
    journal_archive_directory = d:\fbdata\4.0\replication\horses\archive
    journal_archive_command = "copy $(pathname) $(archivepathname) && copy $(pathname) d:\fbdata\4.0\replication\horses\archive_rabbit
}

Please note: that there is a duplication of log archive files so that logical replication and the task of sending events to RabbitMQ work at the same time. This is necessary because log archive files are deleted after processing and cannot be used by another task.

Now you need to enable replication in the database:

ALTER DATABASE INCLUDE ALL TO PUBLICATION;
ALTER DATABASE ENABLE PUBLICATION;

Next, configure the fb_streaming.conf configuration so that fb_streaming automatically sends events to the RabbitMQ exchanger.

Before configuring, open the “RabbitMQ Management” tool in your browser. If you installed it on the same computer, then by default you need to type in the browser http://localhost:15672, login “guest” password “guest”. This tool allows you to manage exchanges, queues and bindings, as well as view queues.

task = d:\fbdata\4.0\replication\horses\archive_rabbit
{
    database = inet://localhost:3054/d:\fbdata\4.0\horses.fdb
    username = SYSDBA
    password = masterkey
    deleteProcessedFile = true
    plugin = rabbitmq_plugin
    dumpBlobs = true
    register_ddl_events = true
    register_sequence_events = true
    # include_tables =
    # exclude_tables =
    rabbit_uri = amqp://guest:guest@localhost
    exchange_name = horses
    exchange_type = direct
    queue_name = horses_events
    routing_key = 123
}

On Linux, this configuration will look like this:

task = /mnt/d/fbdata/4.0/replication/horses/archive
{
    database = inet://192.168.1.48:3054/horses
    username = SYSDBA
    password = masterkey
    deleteProcessedFile = true
    plugin = rabbitmq_plugin
    dumpBlobs = true
    register_ddl_events = true
    register_sequence_events = true
    # include_tables =
    # exclude_tables =
    rabbit_uri = amqp://test:[email protected]
    exchange_name = horses
    exchange_type = direct
    queue_name = horses_events
    routing_key = 123
}

There are a number of additional settings for this plugin:

rabbit_uri - URI to connect to the RabbitMQ server;
exchange_name - exchanger name. If the exchanger does not yet exist, then it will be created;
exchange_type - exchanger type. Valid values: fanout, topic, direct (default fanout);
queue_name - queue name. If specified, the queue is declared and bound to the exchanger by the key routing_key;
routing_key - routing key;
dumpBlobs - whether to publish new values of BLOB fields (false by default);
register_ddl_events - whether to register DDL events (true by default);
register_sequence_events - whether to register sequence value setting events (true by default);
include_tables - a regular expression that defines the names of tables for which it is necessary to track events;
exclude_tables - a regular expression that defines the names of tables for which events should not be tracked;
charset_for_none_fields - character set used for converting data from string fields with the character set type NONE. Typically, data in string fields with the NONE character set is stored “as is,” meaning in the original encoding used by the application when saving it. Firebird does not re-encode such strings. The JSON format requires all strings to be in UTF-8 encoding. To correctly convert strings from NONE fields to UTF8, the system needs to know the original encoding of the data. This is precisely what the charset_for_none_fields parameter specifies.

Now you can install and start the service:

fb_streaming install

fb_streaming start

On Linux:

sudo systemctl enable fb_streaming

sudo systemctl start fb_streaming

While the service is running, the horses_events queue will receive messages.

The content of the messages will be something like this:

{
  "event": "EXECUTE SQL",
  "sql": "CREATE SEQUENCE SEQ1",
  "tnx": 6590
}
{
  "event": "EXECUTE SQL",
  "sql": "CREATE TABLE TABLE1 (\r\n  ID INT NOT NULL,\r\n  S VARCHAR(10),\r\n  PRIMARY KEY(ID)\r\n)",
  "tnx": 6591
}
{
  "event": "EXECUTE SQL",
  "sql": "ALTER TABLE TABLE1\r\nENABLE PUBLICATION",
  "tnx": 6594
}
{
  "event": "SET SEQUENCE",
  "sequence": "SEQ1",
  "value": 1
}
{
  "event": "INSERT",
  "table": "TABLE1",
  "tnx": 6597,
  "record": {
    "ID": 1,
    "S": "Hello"
  }
}
{
  "event": "UPDATE",
  "table": "COLOR",
  "tnx": 11771,
  "changedFields": [ "NAME_DE" ],
  "oldRecord": {
     "NAME_EN": "dun",
     "NAME": "dun",
     "CODE_COLOR": 14,
     "CODE_SENDER": 1,
     "NAME_DE": "",
     "SHORTNAME_EN": "dun",
     "SHORTNAME": "d."
  },
  "record": {
     "NAME_EN": "dun",
     "NAME": "dun",
     "CODE_COLOR": 14,
     "CODE_SENDER": 1,
     "NAME_DE": "g",
     "SHORTNAME_EN": "dun",
     "SHORTNAME": "d."
  }
}
{
  "event": "INSERT",
  "table": "CLIP",
  "tnx": 11821,
  "record": {
     "AVALUE": 44,
     "CODE_CLIP": 1,
     "CODE_CLIPTYPE": 1,
     "CODE_RECORD": 345,
     "REMARK": null
  }
}
{
  "event": "DELETE",
  "table": "CLIP",
  "tnx": 11849,
  "record": {
     "AVALUE": 44,
     "CODE_CLIP": 1,
     "CODE_CLIPTYPE": 1,
     "CODE_RECORD": 345,
     "REMARK": null
  }
}
{
  "event": "UPDATE",
  "table": "BREED",
  "tnx": 11891,
  "changedFields": [ "MARK" ],
  "oldRecord": {
     "NAME": "Orlov trotter",
     "CODE_DEPARTURE": 15,
     "CODE_BREED": 55,
     "CODE_SENDER": 1,
     "NAME_EN": "Orlov trotter",
     "SHORTNAME_EN": "orl. trot.",
     "SHORTNAME": "orl.trot.",
     "MARK": ""
  },
  "record": {
     "NAME": "Orlov trotter",
     "CODE_DEPARTURE": 15,
     "CODE_BREED": 55,
     "CODE_SENDER": 1,
     "NAME_EN": "Orlov trotter",
     "SHORTNAME_EN": "orl. trot.",
     "SHORTNAME": "orl.trot.",
     "MARK": "5"
  }
}

Field descriptions:

event - event type;
table - the name of the table for which the event occurred;
tnx - transaction number in which the event occurred;
record - a new record in the INSERT and UPDATE events, an old one in the DELETE event;
oldRecord - old record in the UPDATE event;
changedFields - list of column names that were changed in the UPDATE event;
newBlobs - new values of BLOB fields;
sql - SQL query text for DDL operators;
sequence - name of the sequence;
value - new value of the sequence.