mysql_binlog_connector_rust/command/
dump_binlog_gtid_command.rs

1use std::io::Write;
2
3use byteorder::{LittleEndian, WriteBytesExt};
4
5use crate::binlog_error::BinlogError;
6
7use super::{command_type::CommandType, gtid_set::GtidSet};
8
9pub struct DumpBinlogGtidCommand {
10    pub server_id: u64,
11    pub gtid_set: GtidSet,
12}
13
14impl DumpBinlogGtidCommand {
15    pub fn to_bytes(&mut self) -> Result<Vec<u8>, BinlogError> {
16        let mut buf = Vec::new();
17        buf.write_u8(CommandType::BinlogDumpGtid as u8)?;
18
19        // BINLOG_DUMP_NEVER_STOP           = 0x00
20        // BINLOG_DUMP_NON_BLOCK            = 0x01
21        // BINLOG_SEND_ANNOTATE_ROWS_EVENT  = 0x02
22        // BINLOG_THROUGH_POSITION          = 0x02
23        // BINLOG_THROUGH_GTID              = 0x04
24        let binlog_flags = 4;
25        buf.write_u16::<LittleEndian>(binlog_flags)?;
26
27        buf.write_u32::<LittleEndian>(self.server_id as u32)?;
28        // binlog-filename-len
29        buf.write_u32::<LittleEndian>(0)?;
30        // binlog-filename, none
31        // binlog-pos
32        buf.write_u64::<LittleEndian>(4)?;
33
34        let mut data_size = 8; // number of uuid_sets
35        for uuid_set in self.gtid_set.map.values() {
36            data_size += 16; // uuid
37            data_size += 8; // number of intervals
38            data_size += uuid_set.intervals.len() * 16; // start to end
39        }
40        buf.write_u32::<LittleEndian>(data_size as u32)?;
41
42        buf.write_u64::<LittleEndian>(self.gtid_set.map.len() as u64)?;
43        for (uuid, uuid_set) in self.gtid_set.map.iter() {
44            let uuid_bytes = Self::hex_to_bytes(&uuid.replace('-', ""))?;
45            buf.write_all(&uuid_bytes)?;
46
47            // intervals
48            buf.write_u64::<LittleEndian>(uuid_set.intervals.len() as u64)?;
49            for interval in &uuid_set.intervals {
50                buf.write_u64::<LittleEndian>(interval.start)?;
51                buf.write_u64::<LittleEndian>(interval.end + 1)?; // right-open
52            }
53        }
54
55        Ok(buf)
56    }
57
58    fn hex_to_bytes(uuid: &str) -> Result<Vec<u8>, BinlogError> {
59        let mut bytes = Vec::with_capacity(uuid.len() / 2);
60        for i in (0..uuid.len()).step_by(2) {
61            let hex_byte = &uuid[i..i + 2];
62            bytes.push(u8::from_str_radix(hex_byte, 16)?);
63        }
64        Ok(bytes)
65    }
66}
mysql_binlog_connector_rust/command/dump_binlog_gtid_command.rs

mysql_binlog_connector_rust/command/
dump_binlog_gtid_command.rs
