lib.rs

//! Rust-Postgres is a pure-Rust frontend for the popular PostgreSQL database. It
//! exposes a high level interface in the vein of JDBC or Go's `database/sql`
//! package.
//!
//! ```rust,no_run
//! extern crate postgres;
//! extern crate time;
//!
//! use time::Timespec;
//!
//! use postgres::{PostgresConnection, NoSsl};
//!
//! struct Person {
//!     id: i32,
//!     name: String,
//!     time_created: Timespec,
//!     data: Option<Vec<u8>>
//! }
//!
//! fn main() {
//!     let conn = PostgresConnection::connect("postgresql://postgres@localhost",
//!                                            &NoSsl).unwrap();
//!
//!     conn.execute("CREATE TABLE person (
//!                     id              SERIAL PRIMARY KEY,
//!                     name            VARCHAR NOT NULL,
//!                     time_created    TIMESTAMP NOT NULL,
//!                     data            BYTEA
//!                   )", []).unwrap();
//!     let me = Person {
//!         id: 0,
//!         name: "Steven".into_string(),
//!         time_created: time::get_time(),
//!         data: None
//!     };
//!     conn.execute("INSERT INTO person (name, time_created, data)
//!                     VALUES ($1, $2, $3)",
//!                  &[&me.name, &me.time_created, &me.data]).unwrap();
//!
//!     let stmt = conn.prepare("SELECT id, name, time_created, data FROM person")
//!             .unwrap();
//!     for row in stmt.query([]).unwrap() {
//!         let person = Person {
//!             id: row.get(0),
//!             name: row.get(1),
//!             time_created: row.get(2),
//!             data: row.get(3)
//!         };
//!         println!("Found person {}", person.name);
//!     }
//! }
//! ```
#![doc(html_root_url="https://sfackler.github.io/doc")]
#![feature(macro_rules, struct_variant, phase, unsafe_destructor, slicing_syntax, default_type_params, if_let)]
#![warn(missing_doc)]

extern crate collections;
extern crate openssl;
extern crate serialize;
extern crate time;
extern crate phf;
#[phase(plugin)]
extern crate phf_mac;
#[phase(plugin, link)]
extern crate log;

use collections::{Deque, RingBuf};
use url::{UserInfo, Url};
use openssl::crypto::hash::{MD5, Hasher};
use openssl::ssl::SslContext;
use serialize::hex::ToHex;
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::io::{BufferedStream, IoResult, MemWriter};
use std::io::net::ip::Port;
use std::mem;
use std::fmt;

use error::{InvalidUrl,
            MissingPassword,
            MissingUser,
            PgConnectStreamError,
            PgConnectBadResponse,
            PgInvalidColumn,
            PgStreamDesynchronized,
            PgStreamError,
            PgWrongParamCount,
            PostgresConnectError,
            PostgresDbError,
            PostgresError,
            UnsupportedAuthentication,
            PgWrongConnection,
            PgWrongTransaction,
            PgBadResponse};
use io::{MaybeSslStream, InternalStream};
use message::{AuthenticationCleartextPassword,
              AuthenticationGSS,
              AuthenticationKerberosV5,
              AuthenticationMD5Password,
              AuthenticationOk,
              AuthenticationSCMCredential,
              AuthenticationSSPI,
              BackendKeyData,
              BackendMessage,
              BindComplete,
              CommandComplete,
              CopyInResponse,
              DataRow,
              EmptyQueryResponse,
              ErrorResponse,
              NoData,
              NoticeResponse,
              NotificationResponse,
              ParameterDescription,
              ParameterStatus,
              ParseComplete,
              PortalSuspended,
              ReadyForQuery,
              RowDescription,
              RowDescriptionEntry};
use message::{Bind,
              CancelRequest,
              Close,
              CopyData,
              CopyDone,
              CopyFail,
              Describe,
              Execute,
              FrontendMessage,
              Parse,
              PasswordMessage,
              Query,
              StartupMessage,
              Sync,
              Terminate};
use message::{WriteMessage, ReadMessage};
use types::{Oid, PostgresType, ToSql, FromSql, PgUnknownType};

#[macro_escape]
mod macros;

mod io;
mod message;
mod url;
mod util;
pub mod error;
pub mod pool;
pub mod types;

static CANARY: u32 = 0xdeadbeef;

/// A typedef of the result returned by many methods.
pub type PostgresResult<T> = Result<T, PostgresError>;

/// Specifies the target server to connect to.
#[deriving(Clone)]
pub enum PostgresConnectTarget {
    /// Connect via TCP to the specified host.
    TargetTcp(String),
    /// Connect via a Unix domain socket in the specified directory.
    TargetUnix(Path)
}

/// Authentication information
#[deriving(Clone)]
pub struct PostgresUserInfo {
    /// The username
    pub user: String,
    /// An optional password
    pub password: Option<String>,
}

/// Information necessary to open a new connection to a Postgres server.
#[deriving(Clone)]
pub struct PostgresConnectParams {
    /// The target server
    pub target: PostgresConnectTarget,
    /// The target port.
    ///
    /// Defaults to 5432 if not specified.
    pub port: Option<Port>,
    /// The user to login as.
    ///
    /// `PostgresConnection::connect` requires a user but `cancel_query` does
    /// not.
    pub user: Option<PostgresUserInfo>,
    /// The database to connect to. Defaults the value of `user`.
    pub database: Option<String>,
    /// Runtime parameters to be passed to the Postgres backend.
    pub options: Vec<(String, String)>,
}

/// A trait implemented by types that can be converted into a
/// `PostgresConnectParams`.
pub trait IntoConnectParams {
    /// Converts the value of `self` into a `PostgresConnectParams`.
    fn into_connect_params(self) -> Result<PostgresConnectParams, PostgresConnectError>;
}

impl IntoConnectParams for PostgresConnectParams {
    fn into_connect_params(self) -> Result<PostgresConnectParams, PostgresConnectError> {
        Ok(self)
    }
}

impl<'a> IntoConnectParams for &'a str {
    fn into_connect_params(self) -> Result<PostgresConnectParams, PostgresConnectError> {
        match Url::parse(self) {
            Ok(url) => url.into_connect_params(),
            Err(err) => return Err(InvalidUrl(err)),
        }
    }
}

impl IntoConnectParams for Url {
    fn into_connect_params(self) -> Result<PostgresConnectParams, PostgresConnectError> {
        let Url {
            host,
            port,
            user,
            path: url::Path { path, query: options, .. },
            ..
        } = self;

        let maybe_path = try!(url::decode_component(host[]).map_err(InvalidUrl));
        let target = if maybe_path[].starts_with("/") {
            TargetUnix(Path::new(maybe_path))
        } else {
            TargetTcp(host)
        };

        let user = match user {
            Some(UserInfo { user, pass }) => Some(PostgresUserInfo { user: user, password: pass }),
            None => None,
        };

        let database = if !path.is_empty() {
            // path contains the leading /
            let (_, path) = path[].slice_shift_char();
            Some(path.into_string())
        } else {
            None
        };

        Ok(PostgresConnectParams {
            target: target,
            port: port,
            user: user,
            database: database,
            options: options,
        })
    }
}

/// Trait for types that can handle Postgres notice messages
pub trait PostgresNoticeHandler {
    /// Handle a Postgres notice message
    fn handle(&mut self, notice: PostgresDbError);
}

/// A notice handler which logs at the `info` level.
///
/// This is the default handler used by a `PostgresConnection`.
pub struct DefaultNoticeHandler;

impl PostgresNoticeHandler for DefaultNoticeHandler {
    fn handle(&mut self, notice: PostgresDbError) {
        info!("{}: {}", notice.severity, notice.message);
    }
}

/// An asynchronous notification
pub struct PostgresNotification {
    /// The process ID of the notifying backend process
    pub pid: u32,
    /// The name of the channel that the notify has been raised on
    pub channel: String,
    /// The "payload" string passed from the notifying process
    pub payload: String,
}

/// An iterator over asynchronous notifications
pub struct PostgresNotifications<'conn> {
    conn: &'conn PostgresConnection
}

impl<'conn> Iterator<PostgresNotification> for PostgresNotifications<'conn> {
    /// Returns the oldest pending notification or `None` if there are none.
    ///
    /// ## Note
    ///
    /// `next` may return `Some` notification after returning `None` if a new
    /// notification was received.
    fn next(&mut self) -> Option<PostgresNotification> {
        self.conn.conn.borrow_mut().notifications.pop_front()
    }
}

/// Contains information necessary to cancel queries for a session
pub struct PostgresCancelData {
    /// The process ID of the session
    pub process_id: u32,
    /// The secret key for the session
    pub secret_key: u32,
}

/// Attempts to cancel an in-progress query.
///
/// The backend provides no information about whether a cancellation attempt
/// was successful or not. An error will only be returned if the driver was
/// unable to connect to the database.
///
/// A `PostgresCancelData` object can be created via
/// `PostgresConnection::cancel_data`. The object can cancel any query made on
/// that connection.
///
/// Only the host and port of the connection info are used. See
/// `PostgresConnection::connect` for details of the `params` argument.
///
/// ## Example
///
/// ```rust,no_run
/// # use postgres::{PostgresConnection, NoSsl};
/// # let url = "";
/// let conn = PostgresConnection::connect(url, &NoSsl).unwrap();
/// let cancel_data = conn.cancel_data();
/// spawn(proc() {
///     conn.execute("SOME EXPENSIVE QUERY", []).unwrap();
/// });
/// # let _ =
/// postgres::cancel_query(url, &NoSsl, cancel_data);
/// ```
pub fn cancel_query<T>(params: T, ssl: &SslMode, data: PostgresCancelData)
                       -> Result<(), PostgresConnectError> where T: IntoConnectParams {
    let params = try!(params.into_connect_params());
    let mut socket = try!(io::initialize_stream(&params, ssl));

    try_pg_conn!(socket.write_message(&CancelRequest {
        code: message::CANCEL_CODE,
        process_id: data.process_id,
        secret_key: data.secret_key
    }));
    try_pg_conn!(socket.flush());

    Ok(())
}

struct InnerPostgresConnection {
    stream: BufferedStream<MaybeSslStream<InternalStream>>,
    next_stmt_id: uint,
    notice_handler: Box<PostgresNoticeHandler+Send>,
    notifications: RingBuf<PostgresNotification>,
    cancel_data: PostgresCancelData,
    unknown_types: HashMap<Oid, String>,
    desynchronized: bool,
    finished: bool,
    trans_depth: u32,
    canary: u32,
}

impl Drop for InnerPostgresConnection {
    fn drop(&mut self) {
        if !self.finished {
            let _ = self.finish_inner();
        }
    }
}

impl InnerPostgresConnection {
    fn connect<T>(params: T, ssl: &SslMode)
                  -> Result<InnerPostgresConnection, PostgresConnectError>
            where T: IntoConnectParams {
        let params = try!(params.into_connect_params());
        let stream = try!(io::initialize_stream(&params, ssl));

        let PostgresConnectParams {
            user,
            database,
            mut options,
            ..
        } = params;

        let user = try!(user.ok_or(MissingUser));

        let mut conn = InnerPostgresConnection {
            stream: BufferedStream::new(stream),
            next_stmt_id: 0,
            notice_handler: box DefaultNoticeHandler,
            notifications: RingBuf::new(),
            cancel_data: PostgresCancelData { process_id: 0, secret_key: 0 },
            unknown_types: HashMap::new(),
            desynchronized: false,
            finished: false,
            trans_depth: 0,
            canary: CANARY,
        };

        options.push(("client_encoding".into_string(), "UTF8".into_string()));
        // Postgres uses the value of TimeZone as the time zone for TIMESTAMP
        // WITH TIME ZONE values. Timespec converts to GMT internally.
        options.push(("TimeZone".into_string(), "GMT".into_string()));
        // We have to clone here since we need the user again for auth
        options.push(("user".into_string(), user.user.clone()));
        match database {
            Some(database) => options.push(("database".into_string(), database)),
            None => {}
        }

        try_pg_conn!(conn.write_messages([StartupMessage {
            version: message::PROTOCOL_VERSION,
            parameters: options[]
        }]));

        try!(conn.handle_auth(user));

        loop {
            match try_pg_conn!(conn.read_message()) {
                BackendKeyData { process_id, secret_key } => {
                    conn.cancel_data.process_id = process_id;
                    conn.cancel_data.secret_key = secret_key;
                }
                ReadyForQuery { .. } => break,
                ErrorResponse { fields } =>
                    return PostgresDbError::new_connect(fields),
                _ => return Err(PgConnectBadResponse),
            }
        }

        Ok(conn)
    }

    fn write_messages(&mut self, messages: &[FrontendMessage]) -> IoResult<()> {
        debug_assert!(!self.desynchronized);
        for message in messages.iter() {
            try_desync!(self, self.stream.write_message(message));
        }
        Ok(try_desync!(self, self.stream.flush()))
    }

    fn read_message(&mut self) -> IoResult<BackendMessage> {
        debug_assert!(!self.desynchronized);
        loop {
            match try_desync!(self, self.stream.read_message()) {
                NoticeResponse { fields } => {
                    if let Ok(err) = PostgresDbError::new_raw(fields) {
                        self.notice_handler.handle(err);
                    }
                }
                NotificationResponse { pid, channel, payload } => {
                    self.notifications.push(PostgresNotification {
                        pid: pid,
                        channel: channel,
                        payload: payload
                    })
                }
                ParameterStatus { parameter, value } => {
                    debug!("Parameter {} = {}", parameter, value)
                }
                val => return Ok(val)
            }
        }
    }

    fn handle_auth(&mut self, user: PostgresUserInfo) -> Result<(), PostgresConnectError> {
        match try_pg_conn!(self.read_message()) {
            AuthenticationOk => return Ok(()),
            AuthenticationCleartextPassword => {
                let pass = try!(user.password.ok_or(MissingPassword));
                try_pg_conn!(self.write_messages([PasswordMessage {
                        password: pass[],
                    }]));
            }
            AuthenticationMD5Password { salt } => {
                let pass = try!(user.password.ok_or(MissingPassword));
                let hasher = Hasher::new(MD5);
                hasher.update(pass.as_bytes());
                hasher.update(user.user.as_bytes());
                let output = hasher.finalize()[].to_hex();
                let hasher = Hasher::new(MD5);
                hasher.update(output.as_bytes());
                hasher.update(salt);
                let output = format!("md5{}",
                                     hasher.finalize()[].to_hex());
                try_pg_conn!(self.write_messages([PasswordMessage {
                        password: output[]
                    }]));
            }
            AuthenticationKerberosV5
            | AuthenticationSCMCredential
            | AuthenticationGSS
            | AuthenticationSSPI => return Err(UnsupportedAuthentication),
            ErrorResponse { fields } => return PostgresDbError::new_connect(fields),
            _ => {
                self.desynchronized = true;
                return Err(PgConnectBadResponse);
            }
        }

        match try_pg_conn!(self.read_message()) {
            AuthenticationOk => Ok(()),
            ErrorResponse { fields } => return PostgresDbError::new_connect(fields),
            _ => {
                self.desynchronized = true;
                return Err(PgConnectBadResponse);
            }
        }
    }

    fn set_notice_handler(&mut self, handler: Box<PostgresNoticeHandler+Send>)
                          -> Box<PostgresNoticeHandler+Send> {
        mem::replace(&mut self.notice_handler, handler)
    }

    fn raw_prepare(&mut self, query: &str)
                   -> PostgresResult<(String, Vec<PostgresType>, Vec<ResultDescription>)> {
        let stmt_name = format!("s{}", self.next_stmt_id);
        self.next_stmt_id += 1;

        try_pg!(self.write_messages([
            Parse {
                name: stmt_name[],
                query: query,
                param_types: []
            },
            Describe {
                variant: b'S',
                name: stmt_name[],
            },
            Sync]));

        match try_pg!(self.read_message()) {
            ParseComplete => {}
            ErrorResponse { fields } => {
                try!(self.wait_for_ready());
                return PostgresDbError::new(fields);
            }
            _ => bad_response!(self),
        }

        let mut param_types: Vec<PostgresType> = match try_pg!(self.read_message()) {
            ParameterDescription { types } => {
                types.iter().map(|ty| PostgresType::from_oid(*ty)).collect()
            }
            _ => bad_response!(self),
        };

        let mut result_desc: Vec<ResultDescription> = match try_pg!(self.read_message()) {
            RowDescription { descriptions } => {
                descriptions.into_iter().map(|RowDescriptionEntry { name, type_oid, .. }| {
                    ResultDescription {
                        name: name,
                        ty: PostgresType::from_oid(type_oid)
                    }
                }).collect()
            }
            NoData => vec![],
            _ => bad_response!(self)
        };

        try!(self.wait_for_ready());

        // now that the connection is ready again, get unknown type names
        try!(self.set_type_names(param_types.iter_mut()));
        try!(self.set_type_names(result_desc.iter_mut().map(|d| &mut d.ty)));

        Ok((stmt_name, param_types, result_desc))
    }

    fn prepare<'a>(&mut self, query: &str, conn: &'a PostgresConnection)
                   -> PostgresResult<PostgresStatement<'a>> {
        let (stmt_name, param_types, result_desc) = try!(self.raw_prepare(query));
        Ok(PostgresStatement {
            conn: conn,
            name: stmt_name,
            param_types: param_types,
            result_desc: result_desc,
            next_portal_id: Cell::new(0),
            finished: false,
        })
    }

    fn prepare_copy_in<'a>(&mut self, table: &str, rows: &[&str], conn: &'a PostgresConnection)
                           -> PostgresResult<PostgresCopyInStatement<'a>> {
        let mut query = MemWriter::new();
        let _ = write!(query, "SELECT ");
        let _ = util::comma_join(&mut query, rows.iter().map(|&e| e));
        let _ = write!(query, " FROM {}", table);
        let query = String::from_utf8(query.unwrap()).unwrap();
        let (stmt_name, _, result_desc) = try!(self.raw_prepare(query[]));

        let column_types = result_desc.iter().map(|desc| desc.ty.clone()).collect();
        try!(self.close_statement(stmt_name[]));

        let mut query = MemWriter::new();
        let _ = write!(query, "COPY {} (", table);
        let _ = util::comma_join(&mut query, rows.iter().map(|&e| e));
        let _ = write!(query, ") FROM STDIN WITH (FORMAT binary)");
        let query = String::from_utf8(query.unwrap()).unwrap();
        let (stmt_name, _, _) = try!(self.raw_prepare(query[]));

        Ok(PostgresCopyInStatement {
            conn: conn,
            name: stmt_name,
            column_types: column_types,
            finished: false,
        })
    }

    fn close_statement(&mut self, stmt_name: &str) -> PostgresResult<()> {
        try_pg!(self.write_messages([
            Close {
                variant: b'S',
                name: stmt_name,
            },
            Sync]));
        loop {
            match try_pg!(self.read_message()) {
                ReadyForQuery { .. } => break,
                ErrorResponse { fields } => {
                    try!(self.wait_for_ready());
                    return PostgresDbError::new(fields);
                }
                _ => {}
            }
        }
        Ok(())
    }

    fn set_type_names<'a, I>(&mut self, mut it: I) -> PostgresResult<()>
            where I: Iterator<&'a mut PostgresType> {
        for ty in it {
            if let &PgUnknownType { oid, ref mut name } = ty {
                *name = try!(self.get_type_name(oid));
            }
        }

        Ok(())
    }

    fn get_type_name(&mut self, oid: Oid) -> PostgresResult<String> {
        if let Some(name) = self.unknown_types.find(&oid) {
            return Ok(name.clone());
        }
        let name = try!(self.quick_query(format!("SELECT typname FROM pg_type \
                                                  WHERE oid={}", oid)[]))
            .into_iter().next().unwrap().into_iter().next().unwrap().unwrap();
        self.unknown_types.insert(oid, name.clone());
        Ok(name)
    }

    fn is_desynchronized(&self) -> bool {
        self.desynchronized
    }

    fn canary(&self) -> u32 {
        self.canary
    }

    fn wait_for_ready(&mut self) -> PostgresResult<()> {
        match try_pg!(self.read_message()) {
            ReadyForQuery { .. } => Ok(()),
            _ => bad_response!(self)
        }
    }

    fn quick_query(&mut self, query: &str) -> PostgresResult<Vec<Vec<Option<String>>>> {
        check_desync!(self);
        try_pg!(self.write_messages([Query { query: query }]));

        let mut result = vec![];
        loop {
            match try_pg!(self.read_message()) {
                ReadyForQuery { .. } => break,
                DataRow { row } => {
                    result.push(row.into_iter().map(|opt| {
                        opt.map(|b| String::from_utf8_lossy(b[]).into_string())
                    }).collect());
                }
                CopyInResponse { .. } => {
                    try_pg!(self.write_messages([
                        CopyFail {
                            message: "COPY queries cannot be directly executed",
                        },
                        Sync]));
                }
                ErrorResponse { fields } => {
                    try!(self.wait_for_ready());
                    return PostgresDbError::new(fields);
                }
                _ => {}
            }
        }
        Ok(result)
    }

    fn finish_inner(&mut self) -> PostgresResult<()> {
        check_desync!(self);
        self.canary = 0;
        try_pg!(self.write_messages([Terminate]));
        Ok(())
    }
}

/// A connection to a Postgres database.
pub struct PostgresConnection {
    conn: RefCell<InnerPostgresConnection>
}

impl PostgresConnection {
    /// Creates a new connection to a Postgres database.
    ///
    /// Most applications can use a URL string in the normal format:
    ///
    /// ```notrust
    /// postgresql://user[:password]@host[:port][/database][?param1=val1[[&param2=val2]...]]
    /// ```
    ///
    /// The password may be omitted if not required. The default Postgres port
    /// (5432) is used if none is specified. The database name defaults to the
    /// username if not specified.
    ///
    /// To connect to the server via Unix sockets, `host` should be set to the
    /// absolute path of the directory containing the socket file. Since `/` is
    /// a reserved character in URLs, the path should be URL encoded.  If the
    /// path contains non-UTF 8 characters, a `PostgresConnectParams` struct
    /// should be created manually and passed in. Note that Postgres does not
    /// support SSL over Unix sockets.
    ///
    /// ## Examples
    ///
    /// ```rust,no_run
    /// # use postgres::{PostgresConnection, NoSsl};
    /// # let _ = || {
    /// let url = "postgresql://postgres:hunter2@localhost:2994/foodb";
    /// let conn = try!(PostgresConnection::connect(url, &NoSsl));
    /// # Ok(()) };
    /// ```
    ///
    /// ```rust,no_run
    /// # use postgres::{PostgresConnection, NoSsl};
    /// # let _ = || {
    /// let url = "postgresql://postgres@%2Frun%2Fpostgres";
    /// let conn = try!(PostgresConnection::connect(url, &NoSsl));
    /// # Ok(()) };
    /// ```
    ///
    /// ```rust,no_run
    /// # use postgres::{PostgresConnection, PostgresUserInfo, PostgresConnectParams, NoSsl, TargetUnix};
    /// # let _ = || {
    /// # let some_crazy_path = Path::new("");
    /// let params = PostgresConnectParams {
    ///     target: TargetUnix(some_crazy_path),
    ///     port: None,
    ///     user: Some(PostgresUserInfo {
    ///         user: "postgres".into_string(),
    ///         password: None
    ///     }),
    ///     database: None,
    ///     options: vec![],
    /// };
    /// let conn = try!(PostgresConnection::connect(params, &NoSsl));
    /// # Ok(()) };
    /// ```
    pub fn connect<T>(params: T, ssl: &SslMode) -> Result<PostgresConnection, PostgresConnectError>
            where T: IntoConnectParams {
        InnerPostgresConnection::connect(params, ssl).map(|conn| {
            PostgresConnection { conn: RefCell::new(conn) }
        })
    }

    /// Sets the notice handler for the connection, returning the old handler.
    pub fn set_notice_handler(&self, handler: Box<PostgresNoticeHandler+Send>)
                              -> Box<PostgresNoticeHandler+Send> {
        self.conn.borrow_mut().set_notice_handler(handler)
    }

    /// Returns an iterator over asynchronous notification messages.
    ///
    /// Use the `LISTEN` command to register this connection for notifications.
    pub fn notifications<'a>(&'a self) -> PostgresNotifications<'a> {
        PostgresNotifications { conn: self }
    }

    /// Creates a new prepared statement.
    ///
    /// A statement may contain parameters, specified by `$n` where `n` is the
    /// index of the parameter in the list provided at execution time,
    /// 1-indexed.
    ///
    /// The statement is associated with the connection that created it and may
    /// not outlive that connection.
    ///
    /// ## Example
    ///
    /// ```rust,no_run
    /// # use postgres::{PostgresConnection, NoSsl};
    /// # let conn = PostgresConnection::connect("", &NoSsl).unwrap();
    /// let maybe_stmt = conn.prepare("SELECT foo FROM bar WHERE baz = $1");
    /// let stmt = match maybe_stmt {
    ///     Ok(stmt) => stmt,
    ///     Err(err) => fail!("Error preparing statement: {}", err)
    /// };
    pub fn prepare<'a>(&'a self, query: &str) -> PostgresResult<PostgresStatement<'a>> {
        let mut conn = self.conn.borrow_mut();
        if conn.trans_depth != 0 {
            return Err(PgWrongTransaction);
        }
        conn.prepare(query, self)
    }

    /// Creates a new COPY FROM STDIN prepared statement.
    ///
    /// These statements provide a method to efficiently bulk-upload data to
    /// the database.
    ///
    /// ## Example
    ///
    /// ```rust,no_run
    /// # use postgres::{PostgresConnection, NoSsl};
    /// # use postgres::types::ToSql;
    /// # let _ = || {
    /// # let conn = PostgresConnection::connect("", &NoSsl).unwrap();
    /// try!(conn.execute("CREATE TABLE foo (
    ///                     bar INT PRIMARY KEY,
    ///                     baz VARCHAR
    ///                    )", []));
    ///
    /// let stmt = try!(conn.prepare_copy_in("foo", ["bar", "baz"]));
    /// let data: &[&[&ToSql]] = &[&[&0i32, &"blah".into_string()],
    ///                            &[&1i32, &None::<String>]];
    /// try!(stmt.execute(data.iter().map(|r| r.iter().map(|&e| e))));
    /// # Ok(()) };
    /// ```
    pub fn prepare_copy_in<'a>(&'a self, table: &str, rows: &[&str])
                               -> PostgresResult<PostgresCopyInStatement<'a>> {
        let mut conn = self.conn.borrow_mut();
        if conn.trans_depth != 0 {
            return Err(PgWrongTransaction);
        }
        conn.prepare_copy_in(table, rows, self)
    }

    /// Begins a new transaction.
    ///
    /// Returns a `PostgresTransaction` object which should be used instead of
    /// the connection for the duration of the transaction. The transaction
    /// is active until the `PostgresTransaction` object falls out of scope.
    ///
    /// ## Note
    /// A transaction will roll back by default. The `set_commit`,
    /// `set_rollback`, and `commit` methods alter this behavior.
    ///
    /// ## Example
    ///
    /// ```rust,no_run
    /// # use postgres::{PostgresConnection, NoSsl};
    /// # fn foo() -> Result<(), postgres::error::PostgresError> {
    /// # let conn = PostgresConnection::connect("", &NoSsl).unwrap();
    /// let trans = try!(conn.transaction());
    /// try!(trans.execute("UPDATE foo SET bar = 10", []));
    /// // ...
    ///
    /// try!(trans.commit());
    /// # Ok(())
    /// # }
    /// ```
    pub fn transaction<'a>(&'a self) -> PostgresResult<PostgresTransaction<'a>> {
        let mut conn = self.conn.borrow_mut();
        check_desync!(conn);
        if conn.trans_depth != 0 {
            return Err(PgWrongTransaction);
        }
        try!(conn.quick_query("BEGIN"));
        conn.trans_depth += 1;
        Ok(PostgresTransaction {
            conn: self,
            commit: Cell::new(false),
            depth: 1,
            finished: false,
        })
    }

    /// A convenience function for queries that are only run once.
    ///
    /// If an error is returned, it could have come from either the preparation
    /// or execution of the statement.
    ///
    /// On success, returns the number of rows modified or 0 if not applicable.
    pub fn execute(&self, query: &str, params: &[&ToSql]) -> PostgresResult<uint> {
        self.prepare(query).and_then(|stmt| stmt.execute(params))
    }

    /// Execute a sequence of SQL statements.
    ///
    /// Statements should be separated by `;` characters. If an error occurs,
    /// execution of the sequence will stop at that point. This is intended for
    /// execution of batches of non-dynamic statements - for example, creation
    /// of a schema for a fresh database.
    ///
    /// ## Warning
    ///
    /// Prepared statements should be used for any SQL statement which contains
    /// user-specified data, as it provides functionality to safely embed that
    /// data in the statment. Do not form statements via string concatenation
    /// and feed them into this method.
    ///
    /// ## Example
    ///
    /// ```rust,no_run
    /// # use postgres::{PostgresConnection, PostgresResult};
    /// fn init_db(conn: &PostgresConnection) -> PostgresResult<()> {
    ///     conn.batch_execute("
    ///         CREATE TABLE person (
    ///             id SERIAL PRIMARY KEY,
    ///             name NOT NULL
    ///         );
    ///
    ///         CREATE TABLE purchase (
    ///             id SERIAL PRIMARY KEY,
    ///             person INT NOT NULL REFERENCES person (id),
    ///             time TIMESTAMPTZ NOT NULL,
    ///         );
    ///
    ///         CREATE INDEX ON purchase (time);
    ///         ")
    /// }
    /// ```
    pub fn batch_execute(&self, query: &str) -> PostgresResult<()> {
        let mut conn = self.conn.borrow_mut();
        if conn.trans_depth != 0 {
            return Err(PgWrongTransaction);
        }
        conn.quick_query(query).map(|_| ())
    }

    /// Returns information used to cancel pending queries.
    ///
    /// Used with the `cancel_query` function. The object returned can be used
    /// to cancel any query executed by the connection it was created from.
    pub fn cancel_data(&self) -> PostgresCancelData {
        self.conn.borrow().cancel_data
    }

    /// Returns whether or not the stream has been desynchronized due to an
    /// error in the communication channel with the server.
    ///
    /// If this has occurred, all further queries will immediately return an
    /// error.
    pub fn is_desynchronized(&self) -> bool {
        self.conn.borrow().is_desynchronized()
    }

    /// Consumes the connection, closing it.
    ///
    /// Functionally equivalent to the `Drop` implementation for
    /// `PostgresConnection` except that it returns any error encountered to
    /// the caller.
    pub fn finish(self) -> PostgresResult<()> {
        let mut conn = self.conn.borrow_mut();
        conn.finished = true;
        conn.finish_inner()
    }

    fn canary(&self) -> u32 {
        self.conn.borrow().canary()
    }

    fn write_messages(&self, messages: &[FrontendMessage]) -> IoResult<()> {
        self.conn.borrow_mut().write_messages(messages)
    }
}

/// Specifies the SSL support requested for a new connection
pub enum SslMode {
    /// The connection will not use SSL
    NoSsl,
    /// The connection will use SSL if the backend supports it
    PreferSsl(SslContext),
    /// The connection must use SSL
    RequireSsl(SslContext)
}

/// Represents a transaction on a database connection.
///
/// The transaction will roll back by default.
pub struct PostgresTransaction<'conn> {
    conn: &'conn PostgresConnection,
    commit: Cell<bool>,
    depth: u32,
    finished: bool,
}

#[unsafe_destructor]
impl<'conn> Drop for PostgresTransaction<'conn> {
    fn drop(&mut self) {
        if !self.finished {
            let _ = self.finish_inner();
        }
    }
}