mysql_binlog_connector_rust/command/
command_util.rs

1use crate::{
2    binlog_client::BinlogClient,
3    binlog_error::BinlogError,
4    constants::MysqlRespCode,
5    event::checksum_type::ChecksumType,
6    network::{
7        error_packet::ErrorPacket, packet_channel::PacketChannel,
8        result_set_row_packet::ResultSetRowPacket,
9    },
10};
11
12use super::{
13    dump_binlog_command::DumpBinlogCommand, dump_binlog_gtid_command::DumpBinlogGtidCommand,
14    gtid_set::GtidSet, query_command::QueryCommand,
15};
16
17pub struct CommandUtil {}
18
19impl CommandUtil {
20    pub async fn execute_query(
21        channel: &mut PacketChannel,
22        sql: &str,
23    ) -> Result<Vec<ResultSetRowPacket>, BinlogError> {
24        Self::execute_sql(channel, sql).await?;
25        // read to EOF
26        while channel.read().await?[0] != MysqlRespCode::EOF {}
27        // get result sets
28        let mut result_sets = Vec::new();
29
30        let mut buf = channel.read().await?;
31        while buf[0] != MysqlRespCode::EOF {
32            Self::check_error_packet(&buf)?;
33            let result_set = ResultSetRowPacket::new(&buf)?;
34            result_sets.push(result_set);
35            buf = channel.read().await?;
36        }
37
38        Ok(result_sets)
39    }
40
41    pub async fn execute_sql(channel: &mut PacketChannel, sql: &str) -> Result<(), BinlogError> {
42        let mut command = QueryCommand {
43            sql: sql.to_string(),
44        };
45
46        // send the query command, sequence for non-authenticate commands are always 0
47        channel.write(&command.to_bytes()?, 0).await?;
48
49        // read the response packet
50        let buf = channel.read().await?;
51        Self::check_error_packet(&buf)
52    }
53
54    pub async fn fetch_binlog_info(
55        channel: &mut PacketChannel,
56    ) -> Result<(String, u32, String), BinlogError> {
57        let result_sets = Self::execute_query(channel, "show master status").await?;
58        if result_sets.is_empty() {
59            return Err(BinlogError::ConnectError(
60                "failed to fetch binlog filename and position".into(),
61            ));
62        }
63        let binlog_filename = result_sets[0].values[0].clone();
64        let binlog_position = result_sets[0].values[1].clone().parse::<u32>()?;
65        let gtid_set = result_sets[0].values[4].clone();
66        Ok((binlog_filename, binlog_position, gtid_set))
67    }
68
69    pub async fn fetch_binlog_checksum(
70        channel: &mut PacketChannel,
71    ) -> Result<ChecksumType, BinlogError> {
72        let result_set_rows =
73            Self::execute_query(channel, "select @@global.binlog_checksum").await?;
74        let mut checksum_name = "";
75        if !result_set_rows.is_empty() {
76            checksum_name = result_set_rows[0].values[0].as_str();
77        }
78        Ok(ChecksumType::from_name(checksum_name))
79    }
80
81    pub async fn setup_binlog_connection(channel: &mut PacketChannel) -> Result<(), BinlogError> {
82        let mut command = QueryCommand {
83            sql: "set @master_binlog_checksum= @@global.binlog_checksum".to_string(),
84        };
85        channel.write(&command.to_bytes()?, 0).await?;
86
87        let buf = channel.read().await?;
88        Self::check_error_packet(&buf)?;
89        Ok(())
90    }
91
92    pub async fn dump_binlog(
93        channel: &mut PacketChannel,
94        client: &BinlogClient,
95    ) -> Result<(), BinlogError> {
96        let buf = if client.gtid_enabled {
97            let mut command = DumpBinlogGtidCommand {
98                server_id: client.server_id,
99                gtid_set: GtidSet::new(&client.gtid_set)?,
100            };
101            command.to_bytes()?
102        } else {
103            let mut command = DumpBinlogCommand {
104                binlog_filename: client.binlog_filename.clone(),
105                binlog_position: client.binlog_position,
106                server_id: client.server_id,
107            };
108            command.to_bytes()?
109        };
110        channel.write(&buf, 0).await?;
111        Ok(())
112    }
113
114    pub fn parse_result(buf: &Vec<u8>) -> Result<(), BinlogError> {
115        match buf[0] {
116            MysqlRespCode::OK => Ok(()),
117
118            MysqlRespCode::ERROR => Self::check_error_packet(buf),
119
120            _ => Err(BinlogError::ConnectError("connect mysql failed".into())),
121        }
122    }
123
124    pub fn check_error_packet(buf: &Vec<u8>) -> Result<(), BinlogError> {
125        if buf[0] == MysqlRespCode::ERROR {
126            let error_packet = ErrorPacket::new(buf)?;
127            return Err(BinlogError::ConnectError(format!(
128                "connect mysql failed: {}",
129                error_packet.error_message
130            )));
131        }
132        Ok(())
133    }
134}
mysql_binlog_connector_rust/command/command_util.rs

mysql_binlog_connector_rust/command/
command_util.rs
