WIP: Prepared statements

src/mariadb/connection/establish.rs

@@ -139,7 +139,7 @@ mod test {
 
         conn.select_db("test").await?;
 
-        let mut prepared = conn.prepare("SELECT username FROM users WHERE username = ?;").await?;
+        let mut prepared = conn.prepare("PREPARE password_stmt FROM 'SELECT username FROM users WHERE password = ?';").await?;
 
         println!("{:?}", prepared);
 
@@ -148,24 +148,33 @@ mod test {
         }
 
         let exec = ComStmtExec {
-            stmt_id: -1,
-            flags: StmtExecFlag::ReadOnly,
-            params: Some(vec![Some(Bytes::from_static(b"daniel"))]),
+            stmt_id: prepared.ok.stmt_id,
+            flags: StmtExecFlag::NoCursor,
+            params: Some(vec![Some(Bytes::from_static(b"random"))]),
             param_defs: prepared.param_defs,
         };
 
-        println!("{:?}", ResultSet::deserialize(DeContext::with_stream(&mut conn.context, &mut conn.stream)).await?);
+        conn.send(exec).await?;
 
-        let fetch = ComStmtFetch {
-            stmt_id: -1,
-            rows: 10,
-        };
+        let mut ctx = DeContext::with_stream(&mut conn.context, &mut conn.stream);
+        ctx.next_packet().await?;
 
-        conn.send(fetch).await?;
+        match ctx.decoder.peek_tag() {
+            0xFF => println!("{:?}", ErrPacket::deserialize(&mut ctx)?),
+            0x00 => println!("{:?}", OkPacket::deserialize(&mut ctx)?),
+            _ => println!("{:?}", ResultSet::deserialize(ctx).await?),
+        }
 
-        let buf = conn.stream.next_packet().await?;
-
-        println!("{:?}", buf);
+//        let fetch = ComStmtFetch {
+//            stmt_id: -1,
+//            rows: 10,
+//        };
+//
+//        conn.send(fetch).await?;
+//
+//        let buf = conn.stream.next_packet().await?;
+//
+//        println!("{:?}", buf);
 
 //        println!("{:?}", ResultSet::deserialize(&mut DeContext::new(&mut conn.context, &buf))?);
 

src/mariadb/connection/mod.rs

@@ -104,11 +104,9 @@ impl Connection {
     pub async fn query<'a>(&'a mut self, sql_statement: &'a str) -> Result<Option<ResultSet>, Error> {
         self.send(ComQuery { sql_statement: bytes::Bytes::from(sql_statement) }).await?;
 
-        println!("awaiting next packet");
         let mut ctx = DeContext::with_stream(&mut self.context, &mut self.stream);
         ctx.next_packet().await?;
 
-        println!("Got next packet");
         match ctx.decoder.peek_tag() {
             0xFF => Err(ErrPacket::deserialize(&mut ctx)?.into()),
             0x00 => {
@@ -155,10 +153,9 @@ impl Connection {
             statement: Bytes::from(query),
         }).await?;
 
-//        let mut ctx = DeContext::with_stream(&mut self.context, &mut self.stream);
-//        ctx.next_packet().await?;
-//        ComStmtPrepareResp::deserialize(&mut ctx)
-        Ok(ComStmtPrepareResp::default())
+        let mut ctx = DeContext::with_stream(&mut self.context, &mut self.stream);
+        ctx.next_packet().await?;
+        Ok(ComStmtPrepareResp::deserialize(ctx).await?)
     }
 }
 
@@ -187,6 +184,7 @@ impl Framed {
             // After this operation we know that packet_headers.last() *SHOULD* always return valid data,
             // so the the use of packet_headers.last().unwrap() is allowed.
             // TODO: Stitch packets together by removing the length and seq_no from in-between packet definitions.
+            println!("{:?}", self.buf);
             if let Some(packet_header) = packet_headers.last() {
                 if packet_header.length as usize == encode::U24_MAX {
                     packet_headers.push(PacketHeader::try_from(&self.buf[self.index..])?);

src/mariadb/protocol/encode.rs

@@ -210,8 +210,10 @@ impl Encoder {
             panic!("String inside string lenenc serialization is too long");
         }
 
-        self.encode_int_u24(bytes.len() as u32);
-        self.buf.extend_from_slice(bytes);
+        self.encode_int_lenenc(Some(&bytes.len()));
+        if bytes.len() > 0 {
+            self.buf.extend_from_slice(bytes);
+        }
     }
 
     // Same as the string counterpart copied to maintain consistency with the spec.

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

@@ -23,7 +23,7 @@ impl crate::mariadb::Serialize for ComStmtExec {
             (Some(params), Some(param_defs)) if params.len() > 0 => {
                 let null_bitmap_size = (params.len() + 7) / 8;
                 let mut shift_amount = 0u8;
-                let mut bitmap = vec![0u8];
+                let mut bitmap = vec![1u8];
                 let send_type = 1u8;
 
                 // Generate NULL-bitmap from params

src/mariadb/protocol/packets/result_set.rs

@@ -14,6 +14,8 @@ impl ResultSet {
     pub async fn deserialize<'a>(mut ctx: DeContext<'a>) -> Result<Self, Error> {
         let column_packet = ColumnPacket::deserialize(&mut ctx)?;
 
+        println!("{:?}", column_packet);
+
         let columns = if let Some(columns) = column_packet.columns {
             let mut column_defs = Vec::new();
             for _ in 0..columns {
@@ -25,11 +27,14 @@ impl ResultSet {
             Vec::new()
         };
 
+        println!("{:?}", columns);
+
         ctx.next_packet().await?;
 
         let eof_packet = if !ctx.ctx.capabilities.contains(Capabilities::CLIENT_DEPRECATE_EOF) {
             // If we get an eof packet we must update ctx to hold a new buffer of the next packet.
             let eof_packet = Some(EofPacket::deserialize(&mut ctx)?);
+            println!("{:?}", eof_packet);
             ctx.next_packet().await?;
             eof_packet
         } else {