binlog_dump.js
2.85 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
'use strict';
const Command = require('./command');
const Packets = require('../packets');
const eventParsers = [];
class BinlogEventHeader {
constructor(packet) {
this.timestamp = packet.readInt32();
this.eventType = packet.readInt8();
this.serverId = packet.readInt32();
this.eventSize = packet.readInt32();
this.logPos = packet.readInt32();
this.flags = packet.readInt16();
}
}
class BinlogDump extends Command {
constructor(opts) {
super();
// this.onResult = callback;
this.opts = opts;
}
start(packet, connection) {
const newPacket = new Packets.BinlogDump(this.opts);
connection.writePacket(newPacket.toPacket(1));
return BinlogDump.prototype.binlogData;
}
binlogData(packet) {
// ok - continue consuming events
// error - error
// eof - end of binlog
if (packet.isEOF()) {
this.emit('eof');
return null;
}
// binlog event header
packet.readInt8();
const header = new BinlogEventHeader(packet);
const EventParser = eventParsers[header.eventType];
let event;
if (EventParser) {
event = new EventParser(packet);
} else {
event = {
name: 'UNKNOWN'
};
}
event.header = header;
this.emit('event', event);
return BinlogDump.prototype.binlogData;
}
}
class RotateEvent {
constructor(packet) {
this.pposition = packet.readInt32();
// TODO: read uint64 here
packet.readInt32(); // positionDword2
this.nextBinlog = packet.readString();
this.name = 'RotateEvent';
}
}
class FormatDescriptionEvent {
constructor(packet) {
this.binlogVersion = packet.readInt16();
this.serverVersion = packet.readString(50).replace(/\u0000.*/, ''); // eslint-disable-line no-control-regex
this.createTimestamp = packet.readInt32();
this.eventHeaderLength = packet.readInt8(); // should be 19
this.eventsLength = packet.readBuffer();
this.name = 'FormatDescriptionEvent';
}
}
class QueryEvent {
constructor(packet) {
const parseStatusVars = require('../packets/binlog_query_statusvars.js');
this.slaveProxyId = packet.readInt32();
this.executionTime = packet.readInt32();
const schemaLength = packet.readInt8();
this.errorCode = packet.readInt16();
const statusVarsLength = packet.readInt16();
const statusVars = packet.readBuffer(statusVarsLength);
this.schema = packet.readString(schemaLength);
packet.readInt8(); // should be zero
this.statusVars = parseStatusVars(statusVars);
this.query = packet.readString();
this.name = 'QueryEvent';
}
}
class XidEvent {
constructor(packet) {
this.binlogVersion = packet.readInt16();
this.xid = packet.readInt64();
this.name = 'XidEvent';
}
}
eventParsers[2] = QueryEvent;
eventParsers[4] = RotateEvent;
eventParsers[15] = FormatDescriptionEvent;
eventParsers[16] = XidEvent;
module.exports = BinlogDump;