Update to odbc 0.15

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "r2d2_odbc"
-version = "0.3.0"
+version = "0.4.0"
 authors = ["Konstantin Salikhov <koka58@yandex.ru>"]
 license = "MIT"
 description = "ODBC support for the r2d2 connection pool"
@@ -22,7 +22,8 @@ path = "tests/test.rs"
 [dependencies]
 lazy_static = "1.0"
 r2d2 = "0.8"
-odbc = "0.14"
+odbc = "0.15"
+odbc-safe = "0.4.2"
 
 [badges]
 travis-ci = { repository = "Koka/r2d2-odbc", branch = "master" }

src/lib.rs

@@ -1,6 +1,7 @@
 //! ODBC support for the `r2d2` connection pool.
 extern crate r2d2;
 extern crate odbc;
+extern crate odbc_safe as safe;
 
 #[macro_use]
 extern crate lazy_static;
@@ -14,12 +15,18 @@ pub struct ODBCConnectionManager {
     connection_string: String
 }
 
-pub struct ODBCConnection<'a>(Connection<'a>);
+#[derive(Debug)]
+pub struct ODBCConnectionManagerTx {
+    connection_string: String
+}
+
+pub struct ODBCConnection<'a, AC: safe::AutocommitMode>(Connection<'a, AC>);
 
-unsafe impl Send for ODBCConnection<'static> {}
+unsafe impl Send for ODBCConnection<'static, safe::AutocommitOn> {}
+unsafe impl Send for ODBCConnection<'static, safe::AutocommitOff> {}
 
-impl <'a> ODBCConnection<'a> {
-    pub fn raw(&self) -> &Connection<'a> {
+impl <'a, AC: safe::AutocommitMode> ODBCConnection<'a, AC> {
+    pub fn raw(&self) -> &Connection<'a, AC> {
         &self.0
     }
 }
@@ -31,7 +38,7 @@ unsafe impl Sync for ODBCEnv {}
 unsafe impl Send for ODBCEnv {}
 
 #[derive(Debug)]
-pub struct ODBCError(Box<Error>);
+pub struct ODBCError(Box<dyn Error>);
 
 lazy_static! {
     static ref ENV: ODBCEnv = ODBCEnv(create_environment_v3().unwrap());
@@ -51,7 +58,7 @@ impl fmt::Display for ODBCError {
 
 impl From<DiagnosticRecord> for ODBCError {
     fn from(err: DiagnosticRecord) -> Self {
-        println!("OLOLO {}", err);
+        println!("ODBC ERROR {}", err);
         ODBCError(Box::new(err))
     }
 }
@@ -72,6 +79,16 @@ impl ODBCConnectionManager {
     }
 }
 
+impl ODBCConnectionManagerTx {
+    /// Creates a new `ODBCConnectionManagerTx`.
+    pub fn new<S: Into<String>>(connection_string: S) -> ODBCConnectionManagerTx
+    {
+        ODBCConnectionManagerTx {
+            connection_string: connection_string.into()
+        }
+    }
+}
+
 /// An `r2d2::ManageConnection` for ODBC connections.
 ///
 /// ## Example
@@ -111,7 +128,7 @@ impl ODBCConnectionManager {
 /// }
 /// ```
 impl r2d2::ManageConnection for ODBCConnectionManager {
-    type Connection = ODBCConnection<'static>;
+    type Connection = ODBCConnection<'static, safe::AutocommitOn>;
     type Error = ODBCError;
 
     fn connect(&self) -> std::result::Result<Self::Connection, Self::Error> {
@@ -129,3 +146,28 @@ impl r2d2::ManageConnection for ODBCConnectionManager {
         false
     }
 }
+
+impl r2d2::ManageConnection for ODBCConnectionManagerTx {
+    type Connection = ODBCConnection<'static, safe::AutocommitOff>;
+    type Error = ODBCError;
+
+    fn connect(&self) -> std::result::Result<Self::Connection, Self::Error> {
+        let env = &ENV.0;
+        let conn = env.connect_with_connection_string(&self.connection_string)?;
+        let conn_result = conn.disable_autocommit();
+        match conn_result {
+            Ok(conn) => Ok(ODBCConnection(conn)),
+            _ => Err(ODBCError("Unable to use transactions".into()))
+        }
+    }
+
+    fn is_valid(&self, _conn: &mut Self::Connection) -> std::result::Result<(), Self::Error> {
+        //TODO
+        Ok(())
+    }
+
+    fn has_broken(&self, _conn: &mut Self::Connection) -> bool {
+        //TODO
+        false
+    }
+}

tests/test.rs

@@ -4,6 +4,7 @@ extern crate r2d2_odbc;
 
 use r2d2::Pool;
 use r2d2_odbc::ODBCConnectionManager;
+use r2d2_odbc::ODBCConnectionManagerTx;
 
 use std::sync::mpsc;
 use std::thread;
@@ -35,5 +36,35 @@ fn test_smoke() {
     t1.join().unwrap();
     t2.join().unwrap();
 
+    pool.get().unwrap();
+}
+
+#[test]
+fn test_smoke_tx() {
+    let manager = ODBCConnectionManagerTx::new("DSN=PostgreSQL");
+    let pool = Pool::builder().max_size(2).build(manager).unwrap();
+
+    let (s1, r1) = mpsc::channel();
+    let (s2, r2) = mpsc::channel();
+
+    let pool1 = pool.clone();
+    let t1 = thread::spawn(move || {
+        let conn = pool1.get().unwrap();
+        s1.send(()).unwrap();
+        r2.recv().unwrap();
+        drop(conn);
+    });
+
+    let pool2 = pool.clone();
+    let t2 = thread::spawn(move || {
+        let conn = pool2.get().unwrap();
+        s2.send(()).unwrap();
+        r1.recv().unwrap();
+        drop(conn);
+    });
+
+    t1.join().unwrap();
+    t2.join().unwrap();
+
     pool.get().unwrap();
 }