Fix a desynchronization issue

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(_)) => {

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));