From 7929d07f7418a6ff775ece8c2d30f42e3b6a448e Mon Sep 17 00:00:00 2001
From: Steven Fackler <sfackler@gmail.com>
Date: Sat, 25 Oct 2014 20:26:45 -0700
Subject: [PATCH] Fix a desynchronization issue

---
 src/lib.rs    | 16 ++++++++++++----
 tests/test.rs | 18 ++++++++++++++++++
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/src/lib.rs b/src/lib.rs
index 524392c4..8cafd48f 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1583,7 +1583,7 @@ impl<'a> PostgresCopyInStatement<'a> {
 
     /// Executes the prepared statement.
     ///
-    /// Each iterator retuned by the `rows` iterator will be interpreted as
+    /// Each iterator returned by the `rows` iterator will be interpreted as
     /// providing a single result row.
     ///
     /// Returns the number of rows copied.
@@ -1637,14 +1637,22 @@ impl<'a> PostgresCopyInStatement<'a> {
             loop {
                 match (row.next(), types.next()) {
                     (Some(val), Some(ty)) => {
-                        match try!(val.to_sql(ty)) {
-                            None => {
+                        match val.to_sql(ty) {
+                            Ok(None) => {
                                 let _ = buf.write_be_i32(-1);
                             }
-                            Some(val) => {
+                            Ok(Some(val)) => {
                                 let _ = buf.write_be_i32(val.len() as i32);
                                 let _ = buf.write(val[]);
                             }
+                            Err(err) => {
+                                // FIXME this is not the right way to handle this
+                                try_pg_desync!(conn, conn.stream.write_message(
+                                    &CopyFail {
+                                        message: err.to_string()[],
+                                    }));
+                                break 'l;
+                            }
                         }
                     }
                     (Some(_), None) | (None, Some(_)) => {
diff --git a/tests/test.rs b/tests/test.rs
index d324807d..f8216b52 100644
--- a/tests/test.rs
+++ b/tests/test.rs
@@ -752,6 +752,24 @@ fn test_copy_in_bad_column_count() {
     or_fail!(conn.execute("SELECT 1", []));
 }
 
+#[test]
+fn test_copy_in_bad_type() {
+    let conn = or_fail!(PostgresConnection::connect("postgres://postgres@localhost", &NoSsl));
+    or_fail!(conn.execute("CREATE TEMPORARY TABLE foo (id INT, name VARCHAR)", []));
+
+    let stmt = or_fail!(conn.prepare_copy_in("foo", ["id", "name"]));
+    let data: &[&[&ToSql]] = &[&[&0i32, &"Steven".to_string()], &[&1i32, &2i32]];
+
+    let res = stmt.execute(data.iter().map(|r| r.iter().map(|&e| e)));
+    match res {
+        Err(PgDbError(ref err)) if err.message[].contains("Unexpected type PgVarchar") => {}
+        Err(err) => fail!("unexpected error {}", err),
+        _ => fail!("Expected error"),
+    }
+
+    or_fail!(conn.execute("SELECT 1", []));
+}
+
 #[test]
 fn test_batch_execute_copy_from_err() {
     let conn = or_fail!(PostgresConnection::connect("postgres://postgres@localhost", &NoSsl));