Add basic serialization tests

src/mariadb/connection/mod.rs

@@ -6,18 +6,7 @@ use futures::{
     prelude::*,
 };
 use runtime::net::TcpStream;
-
-use crate::ConnectOptions;
-
-use crate::mariadb::protocol::{
-    deserialize::{DeContext, Deserialize},
-    encode::Encoder,
-    packets::{com_init_db::ComInitDb, com_ping::ComPing, com_query::ComQuery, com_quit::ComQuit, ok::OkPacket},
-    serialize::Serialize,
-    server::Message as ServerMessage,
-    types::{Capabilities, ServerStatusFlag},
-};
-use crate::mariadb::protocol::server::Message;
+use crate::{ConnectOptions, mariadb::protocol::{DeContext, Deserialize, Encoder, ComInitDb, ComPing, ComQuery, ComQuit, OkPacket, Serialize, Message, Capabilities, ServerStatusFlag}};
 
 mod establish;
 
@@ -160,7 +149,7 @@ impl Connection {
         Ok(())
     }
 
-    pub async fn next(&mut self) -> Result<Option<ServerMessage>, Error> {
+    pub async fn next(&mut self) -> Result<Option<Message>, Error> {
         let mut rbuf = BytesMut::new();
         let mut len = 0;
 
@@ -187,7 +176,7 @@ impl Connection {
 
             while len > 0 {
                 let size = rbuf.len();
-                let message = ServerMessage::deserialize(&mut DeContext::new(
+                let message = Message::deserialize(&mut DeContext::new(
                     &mut self.context,
                     &rbuf.as_ref().into(),
                 ))?;

src/mariadb/mod.rs

@@ -47,5 +47,3 @@ pub use protocol::FieldType;
 pub use protocol::FieldDetailFlag;
 pub use protocol::SessionChangeType;
 pub use protocol::StmtExecFlag;
-pub use protocol::TextProtocol;
-pub use protocol::BinaryProtocol;

src/mariadb/protocol/client.rs

@@ -1,50 +0,0 @@
-// Reference: https://mariadb.com/kb/en/library/connection
-// Packets: https://mariadb.com/kb/en/library/0-packet
-
-// TODO: Handle lengths which are greater than 3 bytes
-// Either break the backet into several smaller ones, or
-// return error
-// TODO: Handle different Capabilities for server and client
-// TODO: Handle when capability is set, but field is None
-
-use super::packets::{SetOptionOptions, ShutdownOptions};
-
-// This is an enum of text protocol packet tags.
-// Tags are the 5th byte of the packet (1st byte of packet body)
-// and are used to determine which type of query was sent.
-// The name of the enum variant represents the type of query, and
-// the value is the byte value required by the server.
-pub enum TextProtocol {
-    ComChangeUser = 0x11,
-    ComDebug = 0x0D,
-    ComInitDb = 0x02,
-    ComPing = 0x0e,
-    ComProcessKill = 0xC,
-    ComQuery = 0x03,
-    ComQuit = 0x01,
-    ComResetConnection = 0x1F,
-    ComSetOption = 0x1B,
-    ComShutdown = 0x0A,
-    ComSleep = 0x00,
-    ComStatistics = 0x09,
-}
-
-// Helper method to easily transform into u8
-impl Into<u8> for TextProtocol {
-    fn into(self) -> u8 {
-        self as u8
-    }
-}
-
-pub enum BinaryProtocol {
-    ComStmtPrepare = 0x16,
-    ComStmtClose = 0x19,
-    ComStmtExec = 0x17,
-}
-
-// Helper method to easily transform into u8
-impl Into<u8> for BinaryProtocol {
-    fn into(self) -> u8 {
-        self as u8
-    }
-}

src/mariadb/protocol/encode.rs

@@ -231,7 +231,7 @@ impl Encoder {
     }
 
     #[inline]
-    pub fn encode_binary(&mut self, bytes: &Bytes, ty: &FieldType) {
+    pub fn encode_param(&mut self, bytes: &Bytes, ty: &FieldType) {
         match ty {
             FieldType::MysqlTypeDecimal => self.encode_string_lenenc(bytes),
             FieldType::MysqlTypeTiny => self.encode_int_1(bytes),

src/mariadb/protocol/mod.rs

@@ -1,4 +1,12 @@
-pub mod client;
+// Reference: https://mariadb.com/kb/en/library/connection
+// Packets: https://mariadb.com/kb/en/library/0-packet
+
+// TODO: Handle lengths which are greater than 3 bytes
+// Either break the packet into several smaller ones, or
+// return error
+// TODO: Handle different Capabilities for server and client
+// TODO: Handle when capability is set, but field is None
+
 pub mod decode;
 pub mod deserialize;
 pub mod encode;
@@ -39,6 +47,8 @@ pub use packets::ComStmtPrepareOk;
 pub use packets::ComStmtPrepareResp;
 pub use packets::ComStmtClose;
 pub use packets::ComStmtExec;
+pub use packets::ComStmtFetch;
+pub use packets::ComStmtReset;
 
 pub use decode::Decoder;
 
@@ -59,6 +69,3 @@ pub use types::FieldType;
 pub use types::FieldDetailFlag;
 pub use types::SessionChangeType;
 pub use types::StmtExecFlag;
-
-pub use client::TextProtocol;
-pub use client::BinaryProtocol;

src/mariadb/protocol/packets/binary/com_stmt_close.rs

@@ -0,0 +1,40 @@
+use std::convert::TryInto;
+
+#[derive(Debug)]
+pub struct ComStmtClose {
+    stmt_id: i32
+}
+
+impl crate::mariadb::Serialize for ComStmtClose {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), failure::Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::BinaryProtocol::ComStmtClose.into());
+        encoder.encode_int_i32(self.stmt_id);
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder, Serialize};
+
+    #[test]
+    fn it_encodes_com_stmt_close() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComStmtClose {
+            stmt_id: 1
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x05\0\0\x00\x19\x01\0\0\0");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/binary/com_stmt_exec.rs

@@ -0,0 +1,111 @@
+use crate::mariadb::{StmtExecFlag, ColumnDefPacket, FieldDetailFlag};
+use bytes::Bytes;
+
+#[derive(Debug)]
+pub struct ComStmtExec {
+    pub stmt_id: i32,
+    pub flags: StmtExecFlag,
+    pub params: Option<Vec<Option<Bytes>>>,
+    pub param_defs: Option<Vec<ColumnDefPacket>>,
+}
+
+impl crate::mariadb::Serialize for ComStmtExec {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), failure::Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::BinaryProtocol::ComStmtExec.into());
+        encoder.encode_int_i32(self.stmt_id);
+        encoder.encode_int_u8(self.flags as u8);
+        encoder.encode_int_u8(0);
+
+        if let Some(params) = &self.params {
+            if let Some(param_defs) = &self.param_defs {
+                if params.len() != param_defs.len() {
+                    failure::bail!("Unequal number of params and param definitions supplied");
+                }
+            }
+
+            let null_bitmap_size = (params.len() + 7) / 8;
+            let mut shift_amount = 0u8;
+            let mut bitmap = vec![0u8];
+
+            // Generate NULL-bitmap from params
+            for param in params {
+               if param.is_none() {
+                   bitmap.push(bitmap.last().unwrap() & (1 << shift_amount));
+                }
+
+                shift_amount = (shift_amount + 1) % 8;
+
+                if shift_amount % 8 == 0 {
+                    bitmap.push(0u8);
+                }
+            }
+
+            // Do not send the param types
+            encoder.encode_int_u8(if self.param_defs.is_some() {
+                1u8
+            } else {
+                0u8
+            });
+
+            if let Some(params_defs) = &self.param_defs {
+                for param in params_defs {
+                    encoder.encode_int_u8(param.field_type as u8);
+                    encoder.encode_int_u8(if (param.field_details & FieldDetailFlag::UNSIGNED).is_empty() {
+                        1u8
+                    } else {
+                        0u8
+                    });
+                }
+
+                // Encode params
+                for index in 0..params.len() {
+                    if let Some(bytes) = &params[index] {
+                        encoder.encode_param(&bytes, &params_defs[index].field_type);
+                    }
+                }
+            }
+        }
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder, Serialize, FieldType, FieldDetailFlag};
+
+    #[test]
+    fn it_encodes_com_stmt_close() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComStmtExec {
+            stmt_id: 1,
+            flags: StmtExecFlag::NoCursor,
+            params: Some(vec![Some(Bytes::from_static(b"\x06daniel"))]),
+            param_defs: Some(vec![ColumnDefPacket {
+                catalog: Bytes::from_static(b"def"),
+                schema: Bytes::from_static(b"test"),
+                table_alias: Bytes::from_static(b"users"),
+                table: Bytes::from_static(b"users"),
+                column_alias: Bytes::from_static(b"username"),
+                column: Bytes::from_static(b"username"),
+                length_of_fixed_fields: Some(0x0Ci64),
+                char_set: 1,
+                max_columns: 1,
+                field_type: FieldType::MysqlTypeString,
+                field_details: FieldDetailFlag::NOT_NULL,
+                decimals: 0,
+            }]),
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        Ok(())
+    }
+}
+

src/mariadb/protocol/packets/binary/com_stmt_fetch.rs

@@ -0,0 +1,41 @@
+#[derive(Debug)]
+pub struct ComStmtFetch {
+    pub stmt_id: i32,
+    pub rows: u32,
+}
+
+impl crate::mariadb::Serialize for ComStmtFetch {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), failure::Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::BinaryProtocol::ComStmtFetch.into());
+        encoder.encode_int_i32(self.stmt_id);
+        encoder.encode_int_u32(self.rows);
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder, Serialize};
+
+    #[test]
+    fn it_encodes_com_stmt_fetch() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComStmtFetch {
+            stmt_id: 1,
+            rows: 10,
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x09\0\0\x00\x1C\x01\0\0\0\x0A\0\0\0");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/binary/com_stmt_prepare.rs

@@ -0,0 +1,41 @@
+use bytes::Bytes;
+
+#[derive(Debug)]
+pub struct ComStmtPrepare {
+    statement: Bytes
+}
+
+impl crate::mariadb::Serialize for ComStmtPrepare {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), failure::Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::BinaryProtocol::ComStmtPrepare.into());
+        encoder.encode_string_eof(&self.statement);
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder, Serialize};
+
+    #[test]
+    fn it_encodes_com_stmt_prepare() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComStmtPrepare {
+            statement: Bytes::from_static(b"SELECT * FROM users WHERE username = ?")
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], Bytes::from_static(b"\x27\0\0\x00\x16SELECT * FROM users WHERE username = ?"));
+
+        Ok(())
+    }
+}
+

src/mariadb/protocol/packets/binary/com_stmt_prepare_ok.rs

@@ -0,0 +1,79 @@
+use std::convert::TryFrom;
+
+#[derive(Debug, Default)]
+pub struct ComStmtPrepareOk {
+    pub stmt_id: i32,
+    pub columns: i16,
+    pub params: i16,
+    pub warnings: i16,
+}
+
+impl crate::mariadb::Deserialize for ComStmtPrepareOk {
+    fn deserialize(ctx: &mut crate::mariadb::DeContext) -> Result<Self, failure::Error> {
+        let decoder = &mut ctx.decoder;
+        let length = decoder.decode_length()?;
+        let seq_no = decoder.decode_int_u8();
+
+        let header = decoder.decode_int_u8();
+
+        let stmt_id = decoder.decode_int_i32();
+
+        let columns = decoder.decode_int_i16();
+        let params = decoder.decode_int_i16();
+
+        // Skip 1 unused byte;
+        decoder.skip_bytes(1);
+
+        let warnings = decoder.decode_int_i16();
+
+        Ok(ComStmtPrepareOk {
+            stmt_id,
+            columns,
+            params,
+            warnings
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{__bytes_builder, ConnectOptions, mariadb::{ConnContext, DeContext, Deserialize}};
+    use bytes::Bytes;
+
+    #[test]
+    fn it_decodes_com_stmt_prepare_ok() -> Result<(), failure::Error> {
+        #[rustfmt::skip]
+        let buf = __bytes_builder!(
+        // int<3> length
+        0u8, 0u8, 0u8,
+        // int<1> seq_no
+        0u8,
+        // int<1> 0x00 COM_STMT_PREPARE_OK header
+        0u8,
+        // int<4> statement id
+        1u8, 0u8, 0u8, 0u8,
+        // int<2> number of columns in the returned result set (or 0 if statement does not return result set)
+        10u8, 0u8,
+        // int<2> number of prepared statement parameters ('?' placeholders)
+        1u8, 0u8,
+        // string<1> -not used-
+        0u8,
+        // int<2> number of warnings
+        0u8, 0u8
+        );
+
+        let mut context = ConnContext::new();
+        let mut ctx = DeContext::new(&mut context, &buf);
+
+        let message = ComStmtPrepareOk::deserialize(&mut ctx)?;
+
+        assert_eq!(message.stmt_id, 1);
+        assert_eq!(message.columns, 10);
+        assert_eq!(message.params, 1);
+        assert_eq!(message.warnings, 0);
+
+        Ok(())
+    }
+}
+

src/mariadb/protocol/packets/binary/com_stmt_prepare_resp.rs

@@ -0,0 +1,162 @@
+use crate::mariadb::{ComStmtPrepareOk, ColumnDefPacket, Capabilities, EofPacket};
+
+#[derive(Debug, Default)]
+pub struct ComStmtPrepareResp {
+    pub ok: ComStmtPrepareOk,
+    pub param_defs: Option<Vec<ColumnDefPacket>>,
+    pub res_columns: Option<Vec<ColumnDefPacket>>,
+}
+
+impl crate::mariadb::Deserialize for ComStmtPrepareResp {
+    fn deserialize(ctx: &mut crate::mariadb::DeContext) -> Result<Self, failure::Error> {
+        let ok = ComStmtPrepareOk::deserialize(ctx)?;
+
+        let param_defs = if ok.params > 0 {
+            let param_defs = (0..ok.params).map(|_| ColumnDefPacket::deserialize(ctx))
+                .filter(Result::is_ok)
+                .map(Result::unwrap)
+                .collect::<Vec<ColumnDefPacket>>();
+
+            if !ctx.conn.capabilities.contains(Capabilities::CLIENT_DEPRECATE_EOF) {
+                EofPacket::deserialize(ctx)?;
+            }
+
+            Some(param_defs)
+        } else {
+            None
+        };
+
+        let res_columns = if ok.columns > 0 {
+            let param_defs = (0..ok.columns).map(|_| ColumnDefPacket::deserialize(ctx))
+                .filter(Result::is_ok)
+                .map(Result::unwrap)
+                .collect::<Vec<ColumnDefPacket>>();
+
+            if !ctx.conn.capabilities.contains(Capabilities::CLIENT_DEPRECATE_EOF) {
+                EofPacket::deserialize(ctx)?;
+            }
+
+            Some(param_defs)
+        } else {
+            None
+        };
+
+        Ok(ComStmtPrepareResp {
+            ok,
+            param_defs,
+            res_columns,
+        })
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::{__bytes_builder, ConnectOptions, mariadb::{ConnContext, DeContext, Deserialize}};
+
+    #[test]
+    fn it_decodes_com_stmt_prepare_resp() -> Result<(), failure::Error> {
+        #[rustfmt::skip]
+        let buf = __bytes_builder!(
+        // ---------------------------- //
+        // Statement Prepared Ok Packet //
+        // ---------------------------- //
+
+        // int<3> length
+        0u8, 0u8, 0u8,
+        // int<1> seq_no
+        0u8,
+        // int<1> 0x00 COM_STMT_PREPARE_OK header
+        0u8,
+        // int<4> statement id
+        1u8, 0u8, 0u8, 0u8,
+        // int<2> number of columns in the returned result set (or 0 if statement does not return result set)
+        1u8, 0u8,
+        // int<2> number of prepared statement parameters ('?' placeholders)
+        1u8, 0u8,
+        // string<1> -not used-
+        0u8,
+        // int<2> number of warnings
+        0u8, 0u8,
+
+        // Param column definition
+
+        // ------------------------ //
+        // Column Definition packet //
+        // ------------------------ //
+        // int<3> length
+        52u8, 0u8, 0u8,
+        // int<1> seq_no
+        3u8,
+        // string<lenenc> catalog (always 'def')
+        3u8, b"def",
+        // string<lenenc> schema
+        4u8, b"test",
+        // string<lenenc> table alias
+        5u8, b"users",
+        // string<lenenc> table
+        5u8, b"users",
+        // string<lenenc> column alias
+        8u8, b"username",
+        // string<lenenc> column
+        8u8, b"username",
+        // int<lenenc> length of fixed fields (=0xC)
+        0x0C_u8,
+        // int<2> character set number
+        8u8, 0u8,
+        // int<4> max. column size
+        0xFF_u8, 0xFF_u8, 0u8, 0u8,
+        // int<1> Field types
+        0xFC_u8,
+        // int<2> Field detail flag
+        0x11_u8, 0x10_u8,
+        // int<1> decimals
+        0u8,
+        // int<2> - unused -
+        0u8, 0u8,
+
+        // Result column definitions
+
+        // ------------------------ //
+        // Column Definition packet //
+        // ------------------------ //
+        // int<3> length
+        52u8, 0u8, 0u8,
+        // int<1> seq_no
+        3u8,
+        // string<lenenc> catalog (always 'def')
+        3u8, b"def",
+        // string<lenenc> schema
+        4u8, b"test",
+        // string<lenenc> table alias
+        5u8, b"users",
+        // string<lenenc> table
+        5u8, b"users",
+        // string<lenenc> column alias
+        8u8, b"username",
+        // string<lenenc> column
+        8u8, b"username",
+        // int<lenenc> length of fixed fields (=0xC)
+        0x0C_u8,
+        // int<2> character set number
+        8u8, 0u8,
+        // int<4> max. column size
+        0xFF_u8, 0xFF_u8, 0u8, 0u8,
+        // int<1> Field types
+        0xFC_u8,
+        // int<2> Field detail flag
+        0x11_u8, 0x10_u8,
+        // int<1> decimals
+        0u8,
+        // int<2> - unused -
+        0u8, 0u8
+        );
+
+        let mut context = ConnContext::new();
+        let mut ctx = DeContext::new(&mut context, &buf);
+
+        let message = ComStmtPrepareResp::deserialize(&mut ctx)?;
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/binary/com_stmt_reset.rs

@@ -0,0 +1,38 @@
+#[derive(Debug)]
+pub struct ComStmtReset {
+    pub stmt_id: i32
+}
+
+impl crate::mariadb::Serialize for ComStmtReset {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), failure::Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::BinaryProtocol::ComStmtReset.into());
+        encoder.encode_int_i32(self.stmt_id);
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder, Serialize};
+
+    #[test]
+    fn it_encodes_com_stmt_reset() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComStmtReset {
+            stmt_id: 1
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x05\0\0\x00\x1A\x01\0\0\0");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/binary/mod.rs

@@ -0,0 +1,31 @@
+pub mod com_stmt_prepare;
+pub mod com_stmt_prepare_ok;
+pub mod com_stmt_prepare_resp;
+pub mod com_stmt_close;
+pub mod com_stmt_exec;
+pub mod com_stmt_fetch;
+pub mod com_stmt_reset;
+
+pub use com_stmt_prepare::ComStmtPrepare;
+pub use com_stmt_prepare_ok::ComStmtPrepareOk;
+pub use com_stmt_prepare_resp::ComStmtPrepareResp;
+pub use com_stmt_close::ComStmtClose;
+pub use com_stmt_exec::ComStmtExec;
+pub use com_stmt_fetch::ComStmtFetch;
+pub use com_stmt_reset::ComStmtReset;
+
+pub enum BinaryProtocol {
+    ComStmtPrepare = 0x16,
+    ComStmtExec = 0x17,
+    ComStmtClose = 0x19,
+    ComStmtReset = 0x1A,
+    ComStmtFetch = 0x1C,
+}
+
+// Helper method to easily transform into u8
+impl Into<u8> for BinaryProtocol {
+    fn into(self) -> u8 {
+        self as u8
+    }
+}
+

src/mariadb/protocol/packets/com_debug.rs

@@ -1,17 +0,0 @@
-use crate::mariadb::{Connection, TextProtocol, Serialize};
-use failure::Error;
-
-pub struct ComDebug();
-
-impl Serialize for ComDebug {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComDebug.into());
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_init_db.rs

@@ -1,21 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use bytes::Bytes;
-use failure::Error;
-
-pub struct ComInitDb {
-    pub schema_name: Bytes,
-}
-
-impl Serialize for ComInitDb {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComInitDb.into());
-        encoder.encode_string_null(&self.schema_name);
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_ping.rs

@@ -1,17 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use failure::Error;
-
-pub struct ComPing();
-
-impl Serialize for ComPing {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComPing.into());
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_process_kill.rs

@@ -1,20 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use failure::Error;
-
-pub struct ComProcessKill {
-    pub process_id: u32,
-}
-
-impl Serialize for ComProcessKill {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComProcessKill.into());
-        encoder.encode_int_u32(self.process_id);
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_query.rs

@@ -1,21 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use bytes::Bytes;
-use failure::Error;
-
-pub struct ComQuery {
-    pub sql_statement: Bytes,
-}
-
-impl Serialize for ComQuery {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComQuery.into());
-        encoder.encode_string_eof(&self.sql_statement);
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_quit.rs

@@ -1,17 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use failure::Error;
-
-pub struct ComQuit();
-
-impl Serialize for ComQuit {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComQuit.into());
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_reset_conn.rs

@@ -1,17 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use failure::Error;
-
-pub struct ComResetConnection();
-
-impl Serialize for ComResetConnection {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComResetConnection.into());
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_set_option.rs

@@ -1,33 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use failure::Error;
-
-#[derive(Clone, Copy)]
-pub enum SetOptionOptions {
-    MySqlOptionMultiStatementsOn = 0x00,
-    MySqlOptionMultiStatementsOff = 0x01,
-}
-
-pub struct ComSetOption {
-    pub option: SetOptionOptions,
-}
-
-impl Serialize for ComSetOption {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComSetOption.into());
-        encoder.encode_int_u16(self.option.into());
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}
-
-// Helper method to easily transform into u16
-impl Into<u16> for SetOptionOptions {
-    fn into(self) -> u16 {
-        self as u16
-    }
-}

src/mariadb/protocol/packets/com_shutdown.rs

@@ -1,32 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use failure::Error;
-
-#[derive(Clone, Copy)]
-pub enum ShutdownOptions {
-    ShutdownDefault = 0x00,
-}
-
-pub struct ComShutdown {
-    pub option: ShutdownOptions,
-}
-
-impl Serialize for ComShutdown {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComShutdown.into());
-        encoder.encode_int_u8(self.option.into());
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}
-
-// Helper method to easily transform into u8
-impl Into<u8> for ShutdownOptions {
-    fn into(self) -> u8 {
-        self as u8
-    }
-}

src/mariadb/protocol/packets/com_sleep.rs

@@ -1,17 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use failure::Error;
-
-pub struct ComSleep();
-
-impl Serialize for ComSleep {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComSleep.into());
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_statistics.rs

@@ -1,17 +0,0 @@
-use crate::mariadb::{TextProtocol, Serialize, Connection};
-use failure::Error;
-
-pub struct ComStatistics();
-
-impl Serialize for ComStatistics {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(TextProtocol::ComStatistics.into());
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_stmt_close.rs

@@ -1,20 +0,0 @@
-use std::convert::TryInto;
-
-#[derive(Debug)]
-pub struct ComStmtClose {
-    stmt_id: i32
-}
-
-impl crate::mariadb::Serialize for ComStmtClose {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), failure::Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(crate::mariadb::BinaryProtocol::ComStmtClose.into());
-        encoder.encode_int_i32(self.stmt_id);
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_stmt_exec.rs

@@ -1,68 +0,0 @@
-use crate::mariadb::{StmtExecFlag, ColumnDefPacket, FieldDetailFlag};
-use bytes::Bytes;
-
-#[derive(Debug)]
-pub struct ComStmtExec {
-    pub stmt_id: i32,
-    pub flags: StmtExecFlag,
-    pub params: Option<Vec<Option<Bytes>>>,
-    pub param_defs: Option<Vec<ColumnDefPacket>>,
-}
-
-impl crate::mariadb::Serialize for ComStmtExec {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), failure::Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(crate::mariadb::BinaryProtocol::ComStmtExec.into());
-        encoder.encode_int_i32(self.stmt_id);
-        encoder.encode_int_u8(self.flags as u8);
-        encoder.encode_int_u8(0);
-
-        if let Some(params) = &self.params {
-            let null_bitmap_size = (params.len() + 7) / 8;
-            let mut shift_amount = 0u8;
-            let mut bitmap = vec![0u8];
-
-            // Generate NULL-bitmap from params
-            for param in params {
-               if param.is_none() {
-                   bitmap.push(bitmap.last().unwrap() & (1 << shift_amount));
-                }
-
-                shift_amount = (shift_amount + 1) % 8;
-
-                if shift_amount % 8 == 0 {
-                    bitmap.push(0u8);
-                }
-            }
-
-            // Do not send the param types
-            encoder.encode_int_u8(if self.param_defs.is_some() {
-                1u8
-            } else {
-                0u8
-            });
-
-            if let Some(params) = &self.param_defs {
-                for param in params {
-                    encoder.encode_int_u8(param.field_type as u8);
-                    encoder.encode_int_u8(if (param.field_details & FieldDetailFlag::UNSIGNED).is_empty() {
-                        1u8
-                    } else {
-                        0u8
-                    });
-                }
-            }
-
-            // Encode params
-            for param in params {
-
-            }
-        }
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_stmt_prepare.rs

@@ -1,20 +0,0 @@
-use bytes::Bytes;
-
-#[derive(Debug)]
-pub struct ComStmtPrepare {
-    statement: Bytes
-}
-
-impl crate::mariadb::Serialize for ComStmtPrepare {
-    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::connection::ConnContext, encoder: &mut crate::mariadb::protocol::encode::Encoder) -> Result<(), failure::Error> {
-        encoder.alloc_packet_header();
-        encoder.seq_no(0);
-
-        encoder.encode_int_u8(crate::mariadb::BinaryProtocol::ComStmtPrepare.into());
-        encoder.encode_string_eof(&self.statement);
-
-        encoder.encode_length();
-
-        Ok(())
-    }
-}

src/mariadb/protocol/packets/com_stmt_prepare_ok.rs

@@ -1,34 +0,0 @@
-use std::convert::TryFrom;
-
-#[derive(Debug)]
-pub struct ComStmtPrepareOk {
-    pub stmt_id: i32,
-    pub columns: i16,
-    pub params: i16,
-    pub warnings: i16,
-}
-
-impl crate::mariadb::Deserialize for ComStmtPrepareOk {
-    fn deserialize(ctx: &mut crate::mariadb::DeContext) -> Result<Self, failure::Error> {
-        let decoder = &mut ctx.decoder;
-        let length = decoder.decode_length()?;
-        let seq_no = decoder.decode_int_u8();
-
-        let stmt_id = decoder.decode_int_i32();
-
-        let columns = decoder.decode_int_i16();
-        let params = decoder.decode_int_i16();
-
-        // Skip 1 unused byte;
-        decoder.skip_bytes(1);
-
-        let warnings = decoder.decode_int_i16();
-
-        Ok(ComStmtPrepareOk {
-            stmt_id,
-            columns,
-            params,
-            warnings
-        })
-    }
-}

src/mariadb/protocol/packets/com_stmt_prepare_resp.rs

@@ -1,60 +0,0 @@
-use crate::mariadb::{ComStmtPrepareOk, ColumnDefPacket, Capabilities, EofPacket};
-
-#[derive(Debug)]
-pub struct ComStmtPrepareResp {
-    pub ok: ComStmtPrepareOk,
-    pub param_defs: Option<Vec<ColumnDefPacket>>,
-    pub res_columns: Option<Vec<ColumnDefPacket>>,
-}
-
-//int<1> 0x00 COM_STMT_PREPARE_OK header
-//int<4> statement id
-//int<2> number of columns in the returned result set (or 0 if statement does not return result set)
-//int<2> number of prepared statement parameters ('?' placeholders)
-//string<1> -not used-
-//int<2> number of warnings
-
-impl crate::mariadb::Deserialize for ComStmtPrepareResp {
-    fn deserialize(ctx: &mut crate::mariadb::DeContext) -> Result<Self, failure::Error> {
-        let decoder = &mut ctx.decoder;
-        let length = decoder.decode_length()?;
-
-        let ok = ComStmtPrepareOk::deserialize(ctx)?;
-
-        let param_defs = if ok.params > 0 {
-            let param_defs = (0..ok.params).map(|_| ColumnDefPacket::deserialize(ctx))
-                .filter(Result::is_ok)
-                .map(Result::unwrap)
-                .collect::<Vec<ColumnDefPacket>>();
-
-            if (ctx.conn.capabilities & Capabilities::CLIENT_DEPRECATE_EOF).is_empty() {
-                EofPacket::deserialize(ctx)?;
-            }
-
-            Some(param_defs)
-        } else {
-            None
-        };
-
-        let res_columns = if ok.columns > 0 {
-            let param_defs = (0..ok.columns).map(|_| ColumnDefPacket::deserialize(ctx))
-                .filter(Result::is_ok)
-                .map(Result::unwrap)
-                .collect::<Vec<ColumnDefPacket>>();
-
-            if (ctx.conn.capabilities & Capabilities::CLIENT_DEPRECATE_EOF).is_empty() {
-                EofPacket::deserialize(ctx)?;
-            }
-
-            Some(param_defs)
-        } else {
-            None
-        };
-
-        Ok(ComStmtPrepareResp {
-            ok,
-            param_defs,
-            res_columns
-        })
-    }
-}

src/mariadb/protocol/packets/mod.rs

@@ -1,17 +1,6 @@
 pub mod auth_switch_request;
 pub mod column;
 pub mod column_def;
-pub mod com_debug;
-pub mod com_init_db;
-pub mod com_ping;
-pub mod com_process_kill;
-pub mod com_query;
-pub mod com_quit;
-pub mod com_reset_conn;
-pub mod com_set_option;
-pub mod com_shutdown;
-pub mod com_sleep;
-pub mod com_statistics;
 pub mod eof;
 pub mod err;
 pub mod handshake_response;
@@ -21,28 +10,12 @@ pub mod packet_header;
 pub mod result_set;
 pub mod ssl_request;
 pub mod result_row;
-pub mod com_stmt_prepare;
-pub mod com_stmt_prepare_ok;
-pub mod com_stmt_prepare_resp;
-pub mod com_stmt_close;
-pub mod com_stmt_exec;
+pub mod binary;
+pub mod text;
 
 pub use auth_switch_request::AuthenticationSwitchRequestPacket;
 pub use column::ColumnPacket;
 pub use column_def::ColumnDefPacket;
-pub use com_debug::ComDebug;
-pub use com_init_db::ComInitDb;
-pub use com_ping::ComPing;
-pub use com_process_kill::ComProcessKill;
-pub use com_query::ComQuery;
-pub use com_quit::ComQuit;
-pub use com_reset_conn::ComResetConnection;
-pub use com_set_option::ComSetOption;
-pub use com_set_option::SetOptionOptions;
-pub use com_shutdown::ShutdownOptions;
-pub use com_shutdown::ComShutdown;
-pub use com_sleep::ComSleep;
-pub use com_statistics::ComStatistics;
 pub use eof::EofPacket;
 pub use err::ErrPacket;
 pub use handshake_response::HandshakeResponsePacket;
@@ -52,8 +25,25 @@ pub use packet_header::PacketHeader;
 pub use result_set::ResultSet;
 pub use result_row::ResultRow;
 pub use ssl_request::SSLRequestPacket;
-pub use com_stmt_prepare::ComStmtPrepare;
-pub use com_stmt_prepare_ok::ComStmtPrepareOk;
-pub use com_stmt_prepare_resp::ComStmtPrepareResp;
-pub use com_stmt_close::ComStmtClose;
-pub use com_stmt_exec::ComStmtExec;
+
+pub use text::ComDebug;
+pub use text::ComInitDb;
+pub use text::ComPing;
+pub use text::ComProcessKill;
+pub use text::ComQuery;
+pub use text::ComQuit;
+pub use text::ComResetConnection;
+pub use text::ComSetOption;
+pub use text::SetOptionOptions;
+pub use text::ShutdownOptions;
+pub use text::ComShutdown;
+pub use text::ComSleep;
+pub use text::ComStatistics;
+
+pub use binary::ComStmtPrepare;
+pub use binary::ComStmtPrepareOk;
+pub use binary::ComStmtPrepareResp;
+pub use binary::ComStmtClose;
+pub use binary::ComStmtExec;
+pub use binary::ComStmtFetch;
+pub use binary::ComStmtReset;

src/mariadb/protocol/packets/result_set.rs

@@ -31,7 +31,7 @@ impl Deserialize for ResultSet {
             Vec::new()
         };
 
-        let eof_packet = if !(ctx.conn.capabilities & Capabilities::CLIENT_DEPRECATE_EOF).is_empty() {
+        let eof_packet = if !ctx.conn.capabilities.contains(Capabilities::CLIENT_DEPRECATE_EOF) {
             Some(EofPacket::deserialize(ctx)?)
         } else {
             None
@@ -62,7 +62,7 @@ impl Deserialize for ResultSet {
             }
         }
 
-        if (ctx.conn.capabilities & Capabilities::CLIENT_DEPRECATE_EOF).is_empty() {
+        if ctx.conn.capabilities.contains(Capabilities::CLIENT_DEPRECATE_EOF) {
             OkPacket::deserialize(ctx)?;
         } else {
             EofPacket::deserialize(ctx)?;

src/mariadb/protocol/packets/text/com_debug.rs

@@ -0,0 +1,35 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+pub struct ComDebug();
+
+impl Serialize for ComDebug {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComDebug.into());
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_debug() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComDebug().serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x01\0\0\x00\x0D");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/text/com_init_db.rs

@@ -0,0 +1,42 @@
+use crate::mariadb::{Connection, Serialize};
+use bytes::Bytes;
+use failure::Error;
+
+pub struct ComInitDb {
+    pub schema_name: Bytes,
+}
+
+impl Serialize for ComInitDb {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComInitDb.into());
+        encoder.encode_string_null(&self.schema_name);
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_init_db() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComInitDb {
+          schema_name: Bytes::from_static(b"portal"),
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x08\0\0\x00\x02portal\0");
+
+        Ok(())
+    }
+}
+

src/mariadb/protocol/packets/text/com_ping.rs

@@ -0,0 +1,35 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+pub struct ComPing();
+
+impl Serialize for ComPing {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComPing.into());
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_ping() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComPing().serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x01\0\0\x00\x0E");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/text/com_process_kill.rs

@@ -0,0 +1,40 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+pub struct ComProcessKill {
+    pub process_id: u32,
+}
+
+impl Serialize for ComProcessKill {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComProcessKill.into());
+        encoder.encode_int_u32(self.process_id);
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_process_kill() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComProcessKill {
+            process_id: 1,
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x05\0\0\x00\x0C\x01\0\0\0");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/text/com_query.rs

@@ -0,0 +1,42 @@
+use crate::mariadb::{Connection, Serialize};
+use bytes::Bytes;
+use failure::Error;
+
+pub struct ComQuery {
+    pub sql_statement: Bytes,
+}
+
+impl Serialize for ComQuery {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComQuery.into());
+        encoder.encode_string_eof(&self.sql_statement);
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_query() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComQuery {
+            sql_statement: Bytes::from_static(b"SELECT * FROM users"),
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x14\0\0\x00\x03SELECT * FROM users");
+
+        Ok(())
+    }
+}
+

src/mariadb/protocol/packets/text/com_quit.rs

@@ -0,0 +1,36 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+pub struct ComQuit();
+
+impl Serialize for ComQuit {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComQuit.into());
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_quit() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComQuit().serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x01\0\0\x00\x01");
+
+        Ok(())
+    }
+}
+

src/mariadb/protocol/packets/text/com_reset_conn.rs

@@ -0,0 +1,35 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+pub struct ComResetConnection();
+
+impl Serialize for ComResetConnection {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComResetConnection.into());
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_reset_conn() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComResetConnection().serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x01\0\0\x00\x1F");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/text/com_set_option.rs

@@ -0,0 +1,55 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+#[derive(Clone, Copy)]
+pub enum SetOptionOptions {
+    MySqlOptionMultiStatementsOn = 0x00,
+    MySqlOptionMultiStatementsOff = 0x01,
+}
+
+pub struct ComSetOption {
+    pub option: SetOptionOptions,
+}
+
+impl Serialize for ComSetOption {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComSetOption.into());
+        encoder.encode_int_u16(self.option.into());
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+// Helper method to easily transform into u16
+impl Into<u16> for SetOptionOptions {
+    fn into(self) -> u16 {
+        self as u16
+    }
+}
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_set_option() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComSetOption {
+            option: SetOptionOptions::MySqlOptionMultiStatementsOff
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x03\0\0\x00\x1B\x01\0");
+
+        Ok(())
+    }
+}
+

src/mariadb/protocol/packets/text/com_shutdown.rs

@@ -0,0 +1,53 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+#[derive(Clone, Copy)]
+pub enum ShutdownOptions {
+    ShutdownDefault = 0x00,
+}
+
+pub struct ComShutdown {
+    pub option: ShutdownOptions,
+}
+
+impl Serialize for ComShutdown {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComShutdown.into());
+        encoder.encode_int_u8(self.option.into());
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+// Helper method to easily transform into u8
+impl Into<u8> for ShutdownOptions {
+    fn into(self) -> u8 {
+        self as u8
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_shutdown() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComShutdown {
+            option: ShutdownOptions::ShutdownDefault
+        }.serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x02\0\0\x00\x0A\x00");
+
+        Ok(())
+    }
+}
+

src/mariadb/protocol/packets/text/com_sleep.rs

@@ -0,0 +1,35 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+pub struct ComSleep();
+
+impl Serialize for ComSleep {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComSleep.into());
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_sleep() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComSleep().serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x01\0\0\x00\x00");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/text/com_statistics.rs

@@ -0,0 +1,35 @@
+use crate::mariadb::{Connection, Serialize};
+use failure::Error;
+
+pub struct ComStatistics();
+
+impl Serialize for ComStatistics {
+    fn serialize<'a, 'b>(&self, ctx: &mut crate::mariadb::ConnContext, encoder: &mut crate::mariadb::Encoder) -> Result<(), Error> {
+        encoder.alloc_packet_header();
+        encoder.seq_no(0);
+
+        encoder.encode_int_u8(super::TextProtocol::ComStatistics.into());
+
+        encoder.encode_length();
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::mariadb::{ConnContext, Encoder};
+
+    #[test]
+    fn it_encodes_com_statistics() -> Result<(), failure::Error> {
+        let mut encoder = Encoder::new(128);
+        let mut ctx = ConnContext::new();
+
+        ComStatistics().serialize(&mut ctx, &mut encoder)?;
+
+        assert_eq!(&encoder.buf[..], b"\x01\0\0\x00\x09");
+
+        Ok(())
+    }
+}

src/mariadb/protocol/packets/text/mod.rs

@@ -0,0 +1,52 @@
+pub mod com_debug;
+pub mod com_init_db;
+pub mod com_ping;
+pub mod com_process_kill;
+pub mod com_query;
+pub mod com_quit;
+pub mod com_reset_conn;
+pub mod com_set_option;
+pub mod com_shutdown;
+pub mod com_sleep;
+pub mod com_statistics;
+
+pub use com_debug::ComDebug;
+pub use com_init_db::ComInitDb;
+pub use com_ping::ComPing;
+pub use com_process_kill::ComProcessKill;
+pub use com_query::ComQuery;
+pub use com_quit::ComQuit;
+pub use com_reset_conn::ComResetConnection;
+pub use com_set_option::ComSetOption;
+pub use com_set_option::SetOptionOptions;
+pub use com_shutdown::ShutdownOptions;
+pub use com_shutdown::ComShutdown;
+pub use com_sleep::ComSleep;
+pub use com_statistics::ComStatistics;
+
+// This is an enum of text protocol packet tags.
+// Tags are the 5th byte of the packet (1st byte of packet body)
+// and are used to determine which type of query was sent.
+// The name of the enum variant represents the type of query, and
+// the value is the byte value required by the server.
+pub enum TextProtocol {
+    ComChangeUser = 0x11,
+    ComDebug = 0x0D,
+    ComInitDb = 0x02,
+    ComPing = 0x0e,
+    ComProcessKill = 0x0C,
+    ComQuery = 0x03,
+    ComQuit = 0x01,
+    ComResetConnection = 0x1F,
+    ComSetOption = 0x1B,
+    ComShutdown = 0x0A,
+    ComSleep = 0x00,
+    ComStatistics = 0x09,
+}
+
+// Helper method to easily transform into u8
+impl Into<u8> for TextProtocol {
+    fn into(self) -> u8 {
+        self as u8
+    }
+}