Beginnings of Bind support

Author: sfackler

src/lib.rs

@@ -18,11 +18,7 @@ pub struct PostgresConnection {
 
 impl Drop for PostgresConnection {
     fn drop(&self) {
-        do io_error::cond.trap(|_| { }).inside {
-            do self.stream.with_mut_ref |s| {
-                s.write_message(&Terminate);
-            }
-        }
+        self.write_message(&Terminate);
     }
 }
 
@@ -111,7 +107,8 @@ impl PostgresConnection {
         PostgresStatement {
             conn: self,
             name: stmt_name,
-            num_params: num_params
+            num_params: num_params,
+            next_portal_id: Cell::new(0)
         }
     }
 
@@ -126,9 +123,44 @@ impl PostgresConnection {
 pub struct PostgresStatement<'self> {
     priv conn: &'self PostgresConnection,
     priv name: ~str,
-    priv num_params: uint
+    priv num_params: uint,
+    priv next_portal_id: Cell<uint>
+}
+
+#[unsafe_destructor]
+impl<'self> Drop for PostgresStatement<'self> {
+    fn drop(&self) {
+        self.conn.write_message(&Close('S' as u8, self.name.as_slice()));
+        self.conn.write_message(&Sync);
+        self.conn.read_message(); // CloseComplete or ErrorResponse
+        self.conn.wait_for_ready();
+    }
 }
 
 impl<'self> PostgresStatement<'self> {
+    pub fn num_params(&self) -> uint {
+        self.num_params
+    }
+
+    pub fn query(&self) {
+        let id = self.next_portal_id.take();
+        let portal_name = ifmt!("{:s}_portal_{}", self.name.as_slice(), id);
+        self.next_portal_id.put_back(id + 1);
 
+        let formats = [];
+        let values = [];
+        let result_formats = [];
+
+        self.conn.write_message(&Bind(portal_name, self.name.as_slice(),
+                                      formats, values, result_formats));
+        self.conn.write_message(&Sync);
+
+        match self.conn.read_message() {
+            BindComplete => (),
+            ErrorResponse(ref data) => fail!("Error: %?", data),
+            resp => fail!("Bad response: %?", resp)
+        }
+
+        self.conn.wait_for_ready();
+    }
 }

src/message.rs

@@ -12,6 +12,8 @@ pub static PROTOCOL_VERSION: i32 = 0x0003_0000;
 pub enum BackendMessage {
     AuthenticationOk,
     BackendKeyData(i32, i32),
+    BindComplete,
+    CloseComplete,
     ErrorResponse(HashMap<u8, ~str>),
     NoData,
     ParameterDescription(~[i32]),
@@ -32,6 +34,10 @@ pub struct RowDescriptionEntry {
 }
 
 pub enum FrontendMessage<'self> {
+    /// portal, stmt, formats, values, result formats
+    Bind(&'self str, &'self str, &'self [i16], &'self [Option<&'self [u8]>],
+         &'self [i16]),
+    Close(u8, &'self str),
     Describe(u8, &'self str),
     /// name, query, parameter types
     Parse(&'self str, &'self str, &'self [i32]),
@@ -63,23 +69,56 @@ impl<W: Writer> WriteMessage for W {
         let mut ident = None;
 
         match *message {
-            Describe(variant, ref name) => {
+            Bind(portal, stmt, formats, values, result_formats) => {
+                ident = Some('B');
+                buf.write_string(portal);
+                buf.write_string(stmt);
+
+                buf.write_be_i16_(formats.len() as i16);
+                for format in formats.iter() {
+                    buf.write_be_i16_(*format);
+                }
+
+                buf.write_be_i16_(values.len() as i16);
+                for value in values.iter() {
+                    match *value {
+                        None => {
+                            buf.write_be_i16_(-1);
+                        }
+                        Some(value) => {
+                            buf.write_be_i32_(value.len() as i32);
+                            buf.write(value);
+                        }
+                    }
+                }
+
+                buf.write_be_i16_(result_formats.len() as i16);
+                for format in result_formats.iter() {
+                    buf.write_be_i16_(*format);
+                }
+            }
+            Close(variant, name) => {
+                ident = Some('C');
+                buf.write_u8_(variant);
+                buf.write_string(name);
+            }
+            Describe(variant, name) => {
                 ident = Some('D');
                 buf.write_u8_(variant);
-                buf.write_string(*name);
+                buf.write_string(name);
             }
-            Parse(ref name, ref query, ref param_types) => {
+            Parse(name, query, param_types) => {
                 ident = Some('P');
-                buf.write_string(*name);
-                buf.write_string(*query);
+                buf.write_string(name);
+                buf.write_string(query);
                 buf.write_be_i16_(param_types.len() as i16);
                 for ty in param_types.iter() {
                     buf.write_be_i32_(*ty);
                 }
             }
-            Query(ref query) => {
+            Query(query) => {
                 ident = Some('Q');
-                buf.write_string(*query);
+                buf.write_string(query);
             }
             StartupMessage(ref params) => {
                 buf.write_be_i32_(PROTOCOL_VERSION);
@@ -144,6 +183,8 @@ impl<R: Reader> ReadMessage for R {
 
         let ret = match ident as char {
             '1' => ParseComplete,
+            '2' => BindComplete,
+            '3' => CloseComplete,
             'E' => read_error_message(&mut buf),
             'K' => BackendKeyData(buf.read_be_i32_(), buf.read_be_i32_()),
             'n' => NoData,

src/test.rs

@@ -8,4 +8,5 @@ fn test_connect() {
                                            "postgres");
 
     let stmt = conn.prepare("CREATE TABLE foo (id BIGINT PRIMARY KEY)");
+    stmt.query();
 }