rust-postgres/src/lib.rs

1735 lines
55 KiB
Rust
Raw Normal View History

2014-07-03 05:23:17 +00:00
//! 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;
2014-11-18 03:11:32 +00:00
//! # #[cfg(feature = "time")]
2014-07-03 05:23:17 +00:00
//! extern crate time;
//!
2014-11-18 03:11:32 +00:00
//! # #[cfg(not(feature = "time"))] fn main() {}
//! # #[cfg(feature = "time")] fn main() {
2014-07-03 05:23:17 +00:00
//! use time::Timespec;
//!
2014-11-17 21:46:33 +00:00
//! use postgres::{Connection, SslMode};
2014-07-03 05:23:17 +00:00
//!
//! struct Person {
//! id: i32,
//! name: String,
//! time_created: Timespec,
//! data: Option<Vec<u8>>
//! }
//!
//! fn main() {
2014-11-17 21:46:33 +00:00
//! let conn = Connection::connect("postgresql://postgres@localhost", &SslMode::None)
2014-11-01 23:38:52 +00:00
//! .unwrap();
2014-07-03 05:23:17 +00:00
//!
//! conn.execute("CREATE TABLE person (
//! id SERIAL PRIMARY KEY,
//! name VARCHAR NOT NULL,
//! time_created TIMESTAMP NOT NULL,
//! data BYTEA
2014-11-25 20:59:31 +00:00
//! )", &[]).unwrap();
2014-07-03 05:23:17 +00:00
//! let me = Person {
//! id: 0,
2014-10-18 18:17:12 +00:00
//! name: "Steven".into_string(),
2014-07-03 05:23:17 +00:00
//! 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();
2014-07-03 05:23:17 +00:00
//!
//! let stmt = conn.prepare("SELECT id, name, time_created, data FROM person")
//! .unwrap();
2014-11-25 20:59:31 +00:00
//! for row in stmt.query(&[]).unwrap() {
2014-07-03 05:23:17 +00:00
//! let person = Person {
//! id: row.get(0),
//! name: row.get(1),
//! time_created: row.get(2),
//! data: row.get(3)
2014-07-03 05:23:17 +00:00
//! };
//! println!("Found person {}", person.name);
//! }
//! }
2014-11-18 03:11:32 +00:00
//! # }
2014-07-03 05:23:17 +00:00
//! ```
2014-10-26 20:11:14 +00:00
#![doc(html_root_url="https://sfackler.github.io/doc")]
2014-11-18 02:20:48 +00:00
#![feature(globs, macro_rules, phase, unsafe_destructor, slicing_syntax, default_type_params, if_let)]
2014-10-31 15:51:27 +00:00
#![warn(missing_docs)]
2013-10-08 04:11:54 +00:00
2014-02-15 23:03:35 +00:00
extern crate openssl;
extern crate serialize;
extern crate phf;
2014-06-11 06:27:07 +00:00
#[phase(plugin)]
2014-04-23 05:10:06 +00:00
extern crate phf_mac;
2014-06-11 06:27:07 +00:00
#[phase(plugin, link)]
2014-03-18 04:05:04 +00:00
extern crate log;
2013-08-04 02:17:32 +00:00
2014-11-01 23:14:08 +00:00
use url::Url;
2014-01-03 01:13:52 +00:00
use openssl::crypto::hash::{MD5, Hasher};
use openssl::ssl::SslContext;
2014-02-14 03:26:52 +00:00
use serialize::hex::ToHex;
2014-02-08 02:17:40 +00:00
use std::cell::{Cell, RefCell};
2014-11-02 18:40:03 +00:00
use std::collections::{RingBuf, HashMap};
2014-11-20 04:54:32 +00:00
use std::io::{BufferedStream, IoResult};
use std::io::net::ip::Port;
2014-02-12 07:42:46 +00:00
use std::mem;
use std::fmt;
2014-11-01 23:12:05 +00:00
use std::result;
2013-08-04 02:17:32 +00:00
2014-05-28 04:07:58 +00:00
use io::{MaybeSslStream, InternalStream};
2014-11-04 05:47:53 +00:00
use message::{FrontendMessage, BackendMessage, RowDescriptionEntry};
use message::FrontendMessage::*;
use message::BackendMessage::*;
use message::{WriteMessage, ReadMessage};
2014-11-04 05:29:16 +00:00
#[doc(inline)]
pub use types::{Oid, Type, ToSql, FromSql};
2014-11-04 06:24:11 +00:00
pub use error::{Error, ConnectError, SqlState, DbError, ErrorPosition};
2013-07-25 07:10:18 +00:00
2014-07-07 02:02:22 +00:00
#[macro_escape]
mod macros;
mod io;
mod message;
mod url;
2014-09-30 05:56:43 +00:00
mod util;
2014-11-04 06:24:11 +00:00
mod error;
pub mod types;
2014-11-01 23:02:50 +00:00
const CANARY: u32 = 0xdeadbeef;
2013-10-21 00:32:14 +00:00
2014-04-03 04:26:41 +00:00
/// A typedef of the result returned by many methods.
2014-11-04 06:24:11 +00:00
pub type Result<T> = result::Result<T, Error>;
2014-04-03 04:26:41 +00:00
/// Specifies the target server to connect to.
#[deriving(Clone)]
2014-11-01 23:13:01 +00:00
pub enum ConnectTarget {
/// Connect via TCP to the specified host.
Tcp(String),
/// Connect via a Unix domain socket in the specified directory.
Unix(Path)
}
/// Authentication information
#[deriving(Clone)]
2014-11-01 23:14:08 +00:00
pub struct UserInfo {
/// 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)]
2014-11-01 23:15:30 +00:00
pub struct ConnectParams {
/// The target server
2014-11-01 23:13:01 +00:00
pub target: ConnectTarget,
/// The target port.
///
/// Defaults to 5432 if not specified.
pub port: Option<Port>,
/// The user to login as.
///
2014-11-02 18:38:45 +00:00
/// `Connection::connect` requires a user but `cancel_query` does not.
2014-11-01 23:14:08 +00:00
pub user: Option<UserInfo>,
/// The database to connect to. Defaults the value of `user`.
2014-05-26 03:38:40 +00:00
pub database: Option<String>,
/// Runtime parameters to be passed to the Postgres backend.
2014-05-26 03:38:40 +00:00
pub options: Vec<(String, String)>,
}
2014-11-02 18:38:45 +00:00
/// A trait implemented by types that can be converted into a `ConnectParams`.
pub trait IntoConnectParams {
2014-11-01 23:15:30 +00:00
/// Converts the value of `self` into a `ConnectParams`.
2014-11-04 06:24:11 +00:00
fn into_connect_params(self) -> result::Result<ConnectParams, ConnectError>;
}
2014-11-01 23:15:30 +00:00
impl IntoConnectParams for ConnectParams {
2014-11-04 06:24:11 +00:00
fn into_connect_params(self) -> result::Result<ConnectParams, ConnectError> {
Ok(self)
}
}
impl<'a> IntoConnectParams for &'a str {
2014-11-04 06:24:11 +00:00
fn into_connect_params(self) -> result::Result<ConnectParams, ConnectError> {
2014-07-07 13:41:43 +00:00
match Url::parse(self) {
2014-05-17 03:47:03 +00:00
Ok(url) => url.into_connect_params(),
2014-11-17 21:46:33 +00:00
Err(err) => return Err(ConnectError::InvalidUrl(err)),
2014-05-17 03:47:03 +00:00
}
}
}
impl IntoConnectParams for Url {
2014-11-04 06:24:11 +00:00
fn into_connect_params(self) -> result::Result<ConnectParams, ConnectError> {
let Url {
host,
port,
user,
2014-07-07 13:41:43 +00:00
path: url::Path { path, query: options, .. },
..
2014-05-17 03:47:03 +00:00
} = self;
2014-11-17 21:46:33 +00:00
let maybe_path = try!(url::decode_component(host[]).map_err(ConnectError::InvalidUrl));
2014-10-05 03:08:44 +00:00
let target = if maybe_path[].starts_with("/") {
ConnectTarget::Unix(Path::new(maybe_path))
} else {
ConnectTarget::Tcp(host)
};
2014-11-02 18:38:45 +00:00
let user = user.map(|url::UserInfo { user, pass }| {
UserInfo { user: user, password: pass }
});
2014-11-19 18:11:23 +00:00
// path contains the leading /
let database = path.slice_shift_char().map(|(_, path)| path.into_string());
2014-11-01 23:15:30 +00:00
Ok(ConnectParams {
target: target,
port: port,
user: user,
database: database,
options: options,
})
}
}
2013-09-30 02:12:20 +00:00
/// Trait for types that can handle Postgres notice messages
2014-11-01 23:16:50 +00:00
pub trait NoticeHandler {
2013-09-30 02:12:20 +00:00
/// Handle a Postgres notice message
2014-11-04 06:24:11 +00:00
fn handle(&mut self, notice: DbError);
}
2013-09-30 05:22:10 +00:00
/// A notice handler which logs at the `info` level.
2013-09-30 02:12:20 +00:00
///
2014-11-01 23:21:47 +00:00
/// This is the default handler used by a `Connection`.
pub struct DefaultNoticeHandler;
2014-11-01 23:16:50 +00:00
impl NoticeHandler for DefaultNoticeHandler {
2014-11-04 06:24:11 +00:00
fn handle(&mut self, notice: DbError) {
2013-10-21 04:06:12 +00:00
info!("{}: {}", notice.severity, notice.message);
}
}
/// An asynchronous notification
2014-11-01 23:18:09 +00:00
pub struct Notification {
/// The process ID of the notifying backend process
2014-08-16 03:14:46 +00:00
pub pid: u32,
/// The name of the channel that the notify has been raised on
2014-05-26 03:38:40 +00:00
pub channel: String,
/// The "payload" string passed from the notifying process
2014-05-26 03:38:40 +00:00
pub payload: String,
}
/// An iterator over asynchronous notifications
2014-11-01 23:18:09 +00:00
pub struct Notifications<'conn> {
2014-11-01 23:21:47 +00:00
conn: &'conn Connection
}
2014-11-01 23:18:09 +00:00
impl<'conn> Iterator<Notification> for Notifications<'conn> {
/// Returns the oldest pending notification or `None` if there are none.
///
2014-07-30 02:48:56 +00:00
/// ## Note
///
/// `next` may return `Some` notification after returning `None` if a new
/// notification was received.
2014-11-01 23:18:09 +00:00
fn next(&mut self) -> Option<Notification> {
2014-03-23 04:20:22 +00:00
self.conn.conn.borrow_mut().notifications.pop_front()
}
}
impl<'conn> Notifications<'conn> {
/// Returns the oldest pending notification.
///
/// If no notifications are pending, blocks until one arrives.
pub fn next_block(&mut self) -> Result<Notification> {
if let Some(notification) = self.next() {
return Ok(notification);
}
2014-11-26 03:17:53 +00:00
check_desync!(self.conn.conn.borrow());
match try!(self.conn.conn.borrow_mut().read_message_with_notification()) {
NotificationResponse { pid, channel, payload } => {
Ok(Notification {
pid: pid,
channel: channel,
payload: payload
})
}
_ => unreachable!()
}
}
}
2013-10-21 00:32:14 +00:00
/// Contains information necessary to cancel queries for a session
2014-11-01 23:19:02 +00:00
pub struct CancelData {
2013-10-21 00:32:14 +00:00
/// The process ID of the session
2014-08-16 03:14:46 +00:00
pub process_id: u32,
2013-10-21 00:32:14 +00:00
/// The secret key for the session
2014-08-16 03:14:46 +00:00
pub secret_key: u32,
2013-10-21 00:32:14 +00:00
}
/// 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.
///
2014-11-03 00:48:38 +00:00
/// A `CancelData` object can be created via `Connection::cancel_data`. The
/// object can cancel any query made on that connection.
2014-03-09 22:22:20 +00:00
///
2014-08-21 05:40:11 +00:00
/// Only the host and port of the connection info are used. See
2014-11-01 23:21:47 +00:00
/// `Connection::connect` for details of the `params` argument.
2014-03-31 02:21:51 +00:00
///
2014-07-30 02:48:56 +00:00
/// ## Example
2014-03-09 22:22:20 +00:00
///
2014-03-10 17:06:40 +00:00
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
2014-03-09 22:22:20 +00:00
/// # let url = "";
2014-11-17 21:46:33 +00:00
/// let conn = Connection::connect(url, &SslMode::None).unwrap();
2014-03-09 22:22:20 +00:00
/// let cancel_data = conn.cancel_data();
/// spawn(proc() {
/// conn.execute("SOME EXPENSIVE QUERY", &[]).unwrap();
2014-03-09 22:22:20 +00:00
/// });
/// # let _ =
2014-11-17 21:46:33 +00:00
/// postgres::cancel_query(url, &SslMode::None, cancel_data);
2014-03-09 22:22:20 +00:00
/// ```
2014-11-01 23:19:02 +00:00
pub fn cancel_query<T>(params: T, ssl: &SslMode, data: CancelData)
2014-11-04 06:24:11 +00:00
-> result::Result<(), ConnectError> where T: IntoConnectParams {
let params = try!(params.into_connect_params());
2014-09-06 23:33:43 +00:00
let mut socket = try!(io::initialize_stream(&params, ssl));
2013-10-21 00:32:14 +00:00
2014-11-17 06:54:57 +00:00
try!(socket.write_message(&CancelRequest {
2013-10-21 00:32:14 +00:00
code: message::CANCEL_CODE,
process_id: data.process_id,
secret_key: data.secret_key
2014-02-07 06:59:33 +00:00
}));
2014-11-17 06:54:57 +00:00
try!(socket.flush());
2013-10-21 00:32:14 +00:00
2013-12-06 07:20:50 +00:00
Ok(())
2013-10-21 00:32:14 +00:00
}
2014-11-01 23:21:47 +00:00
struct InnerConnection {
stream: BufferedStream<MaybeSslStream<InternalStream>>,
2013-12-06 07:20:50 +00:00
next_stmt_id: uint,
2014-11-01 23:16:50 +00:00
notice_handler: Box<NoticeHandler+Send>,
2014-11-01 23:18:09 +00:00
notifications: RingBuf<Notification>,
2014-11-01 23:19:02 +00:00
cancel_data: CancelData,
2014-05-26 03:38:40 +00:00
unknown_types: HashMap<Oid, String>,
desynchronized: bool,
finished: bool,
trans_depth: u32,
canary: u32,
2013-08-29 06:19:53 +00:00
}
2014-11-01 23:21:47 +00:00
impl Drop for InnerConnection {
2013-09-18 05:06:47 +00:00
fn drop(&mut self) {
if !self.finished {
let _ = self.finish_inner();
}
2013-08-29 06:19:53 +00:00
}
}
2014-11-01 23:21:47 +00:00
impl InnerConnection {
2014-11-18 02:20:48 +00:00
fn connect<T>(params: T, ssl: &SslMode) -> result::Result<InnerConnection, ConnectError>
2014-08-17 20:57:58 +00:00
where T: IntoConnectParams {
let params = try!(params.into_connect_params());
let stream = try!(io::initialize_stream(&params, ssl));
2014-11-01 23:15:30 +00:00
let ConnectParams {
user,
database,
mut options,
..
} = params;
2014-11-17 21:46:33 +00:00
let user = try!(user.ok_or(ConnectError::MissingUser));
2014-11-01 23:21:47 +00:00
let mut conn = InnerConnection {
2014-05-28 04:07:58 +00:00
stream: BufferedStream::new(stream),
next_stmt_id: 0,
notice_handler: box DefaultNoticeHandler,
notifications: RingBuf::new(),
2014-11-01 23:19:02 +00:00
cancel_data: CancelData { process_id: 0, secret_key: 0 },
2014-05-28 04:07:58 +00:00
unknown_types: HashMap::new(),
desynchronized: false,
finished: false,
trans_depth: 0,
canary: CANARY,
};
2014-10-18 18:17:12 +00:00
options.push(("client_encoding".into_string(), "UTF8".into_string()));
2013-09-09 04:33:41 +00:00
// Postgres uses the value of TimeZone as the time zone for TIMESTAMP
// WITH TIME ZONE values. Timespec converts to GMT internally.
2014-10-18 18:17:12 +00:00
options.push(("TimeZone".into_string(), "GMT".into_string()));
// We have to clone here since we need the user again for auth
2014-10-18 18:17:12 +00:00
options.push(("user".into_string(), user.user.clone()));
match database {
2014-10-18 18:17:12 +00:00
Some(database) => options.push(("database".into_string(), database)),
2014-11-21 05:48:05 +00:00
None => {}
}
2014-11-19 17:58:30 +00:00
try!(conn.write_messages(&[StartupMessage {
2013-09-09 05:40:08 +00:00
version: message::PROTOCOL_VERSION,
2014-10-05 03:08:44 +00:00
parameters: options[]
}]));
try!(conn.handle_auth(user));
2013-08-04 02:17:32 +00:00
2013-08-22 05:52:15 +00:00
loop {
2014-11-17 06:54:57 +00:00
match try!(conn.read_message()) {
2013-10-21 00:32:14 +00:00
BackendKeyData { process_id, secret_key } => {
conn.cancel_data.process_id = process_id;
conn.cancel_data.secret_key = secret_key;
}
ReadyForQuery { .. } => break,
2014-11-04 06:24:11 +00:00
ErrorResponse { fields } => return DbError::new_connect(fields),
_ => return Err(ConnectError::BadResponse),
}
2013-08-05 00:48:48 +00:00
}
2013-08-23 05:24:14 +00:00
Ok(conn)
2013-08-05 00:48:48 +00:00
}
2014-08-20 04:49:27 +00:00
fn write_messages(&mut self, messages: &[FrontendMessage]) -> IoResult<()> {
2014-07-26 03:15:24 +00:00
debug_assert!(!self.desynchronized);
2013-11-10 04:58:38 +00:00
for message in messages.iter() {
2014-07-10 19:35:57 +00:00
try_desync!(self, self.stream.write_message(message));
2013-07-25 07:10:18 +00:00
}
2014-07-10 19:35:57 +00:00
Ok(try_desync!(self, self.stream.flush()))
2013-08-22 06:41:26 +00:00
}
2013-08-04 02:17:32 +00:00
fn read_message_with_notification(&mut self) -> IoResult<BackendMessage> {
2014-07-26 03:15:24 +00:00
debug_assert!(!self.desynchronized);
loop {
2014-07-10 19:35:57 +00:00
match try_desync!(self, self.stream.read_message()) {
2014-08-16 02:50:11 +00:00
NoticeResponse { fields } => {
2014-11-04 06:24:11 +00:00
if let Ok(err) = DbError::new_raw(fields) {
2014-10-14 04:12:25 +00:00
self.notice_handler.handle(err);
}
2014-08-16 02:50:11 +00:00
}
ParameterStatus { parameter, value } => {
debug!("Parameter {} = {}", parameter, value)
}
val => return Ok(val)
}
}
}
fn read_message(&mut self) -> IoResult<BackendMessage> {
debug_assert!(!self.desynchronized);
loop {
match try!(self.read_message_with_notification()) {
2014-08-16 02:50:11 +00:00
NotificationResponse { pid, channel, payload } => {
2014-11-07 16:54:10 +00:00
self.notifications.push_back(Notification {
pid: pid,
channel: channel,
payload: payload
2014-08-16 02:50:11 +00:00
})
}
val => return Ok(val)
}
2013-08-04 05:21:16 +00:00
}
2013-08-22 06:41:26 +00:00
}
2014-11-04 06:24:11 +00:00
fn handle_auth(&mut self, user: UserInfo) -> result::Result<(), ConnectError> {
2014-11-17 06:54:57 +00:00
match try!(self.read_message()) {
AuthenticationOk => return Ok(()),
AuthenticationCleartextPassword => {
2014-11-17 21:46:33 +00:00
let pass = try!(user.password.ok_or(ConnectError::MissingPassword));
2014-11-19 17:58:30 +00:00
try!(self.write_messages(&[PasswordMessage {
2014-10-05 03:08:44 +00:00
password: pass[],
}]));
}
AuthenticationMD5Password { salt } => {
2014-11-17 21:46:33 +00:00
let pass = try!(user.password.ok_or(ConnectError::MissingPassword));
2014-01-03 01:13:52 +00:00
let hasher = Hasher::new(MD5);
hasher.update(pass.as_bytes());
hasher.update(user.user.as_bytes());
2014-10-10 03:43:14 +00:00
let output = hasher.finalize()[].to_hex();
2014-01-03 01:13:52 +00:00
let hasher = Hasher::new(MD5);
hasher.update(output.as_bytes());
2014-11-19 17:58:30 +00:00
hasher.update(&salt);
2014-11-17 16:56:25 +00:00
let output = format!("md5{}", hasher.finalize()[].to_hex());
2014-11-19 17:58:30 +00:00
try!(self.write_messages(&[PasswordMessage {
2014-10-05 03:08:44 +00:00
password: output[]
}]));
}
AuthenticationKerberosV5
2013-09-12 05:02:32 +00:00
| AuthenticationSCMCredential
| AuthenticationGSS
2014-11-17 21:46:33 +00:00
| AuthenticationSSPI => return Err(ConnectError::UnsupportedAuthentication),
2014-11-04 06:24:11 +00:00
ErrorResponse { fields } => return DbError::new_connect(fields),
2014-07-07 02:02:22 +00:00
_ => {
self.desynchronized = true;
return Err(ConnectError::BadResponse);
2014-07-07 02:02:22 +00:00
}
}
2014-11-17 06:54:57 +00:00
match try!(self.read_message()) {
AuthenticationOk => Ok(()),
2014-11-04 06:24:11 +00:00
ErrorResponse { fields } => return DbError::new_connect(fields),
2014-07-07 02:02:22 +00:00
_ => {
self.desynchronized = true;
return Err(ConnectError::BadResponse);
2014-07-07 02:02:22 +00:00
}
}
}
2014-11-02 18:38:45 +00:00
fn set_notice_handler(&mut self, handler: Box<NoticeHandler+Send>) -> Box<NoticeHandler+Send> {
2014-02-11 03:20:53 +00:00
mem::replace(&mut self.notice_handler, handler)
}
2014-09-30 05:56:43 +00:00
fn raw_prepare(&mut self, query: &str)
2014-11-04 05:29:16 +00:00
-> Result<(String, Vec<Type>, Vec<ResultDescription>)> {
2014-05-24 03:08:14 +00:00
let stmt_name = format!("s{}", self.next_stmt_id);
self.next_stmt_id += 1;
2013-08-05 00:48:48 +00:00
2014-11-19 17:58:30 +00:00
try!(self.write_messages(&[
2013-11-10 04:58:38 +00:00
Parse {
2014-10-05 03:08:44 +00:00
name: stmt_name[],
2013-09-02 17:27:09 +00:00
query: query,
2014-11-19 17:58:30 +00:00
param_types: &[]
2013-09-02 17:27:09 +00:00
},
2013-11-10 04:58:38 +00:00
Describe {
2014-09-06 23:33:43 +00:00
variant: b'S',
2014-10-05 03:08:44 +00:00
name: stmt_name[],
2013-09-02 17:27:09 +00:00
},
Sync]));
2013-08-22 06:41:26 +00:00
2014-11-17 06:54:57 +00:00
match try!(self.read_message()) {
2013-10-25 05:30:34 +00:00
ParseComplete => {}
ErrorResponse { fields } => {
2014-02-22 07:18:39 +00:00
try!(self.wait_for_ready());
2014-11-04 06:24:11 +00:00
return DbError::new(fields);
}
2014-07-10 19:35:57 +00:00
_ => bad_response!(self),
}
2013-08-22 06:41:26 +00:00
2014-11-17 06:54:57 +00:00
let mut param_types: Vec<_> = match try!(self.read_message()) {
2014-08-16 02:50:11 +00:00
ParameterDescription { types } => {
2014-11-04 05:29:16 +00:00
types.iter().map(|ty| Type::from_oid(*ty)).collect()
2014-08-16 02:50:11 +00:00
}
2014-07-10 19:35:57 +00:00
_ => bad_response!(self),
};
2013-08-22 06:41:26 +00:00
2014-11-17 06:54:57 +00:00
let mut result_desc: Vec<_> = match try!(self.read_message()) {
2014-08-16 02:50:11 +00:00
RowDescription { descriptions } => {
2014-09-18 05:57:23 +00:00
descriptions.into_iter().map(|RowDescriptionEntry { name, type_oid, .. }| {
2014-04-03 05:56:16 +00:00
ResultDescription {
name: name,
2014-11-04 05:29:16 +00:00
ty: Type::from_oid(type_oid)
2014-04-03 05:56:16 +00:00
}
2014-08-16 02:50:11 +00:00
}).collect()
}
2014-05-26 18:41:18 +00:00
NoData => vec![],
2014-07-10 19:35:57 +00:00
_ => bad_response!(self)
};
2013-08-17 22:09:26 +00:00
2014-02-22 07:18:39 +00:00
try!(self.wait_for_ready());
2013-08-22 06:41:26 +00:00
2013-12-04 08:18:28 +00:00
// now that the connection is ready again, get unknown type names
2014-09-24 07:18:27 +00:00
try!(self.set_type_names(param_types.iter_mut()));
try!(self.set_type_names(result_desc.iter_mut().map(|d| &mut d.ty)));
2013-12-04 08:18:28 +00:00
2014-09-30 05:56:43 +00:00
Ok((stmt_name, param_types, result_desc))
}
2014-11-03 00:48:38 +00:00
fn prepare<'a>(&mut self, query: &str, conn: &'a Connection) -> Result<Statement<'a>> {
2014-09-30 05:56:43 +00:00
let (stmt_name, param_types, result_desc) = try!(self.raw_prepare(query));
2014-11-01 23:24:24 +00:00
Ok(Statement {
conn: conn,
name: stmt_name,
param_types: param_types,
result_desc: result_desc,
next_portal_id: Cell::new(0),
2014-04-26 06:14:55 +00:00
finished: false,
})
2013-08-22 06:41:26 +00:00
}
2014-11-01 23:21:47 +00:00
fn prepare_copy_in<'a>(&mut self, table: &str, rows: &[&str], conn: &'a Connection)
-> Result<CopyInStatement<'a>> {
2014-11-20 04:54:32 +00:00
let mut query = vec![];
2014-11-20 18:54:40 +00:00
let _ = write!(&mut query, "SELECT ");
2014-09-30 05:56:43 +00:00
let _ = util::comma_join(&mut query, rows.iter().map(|&e| e));
2014-11-20 18:54:40 +00:00
let _ = write!(&mut query, " FROM {}", table);
2014-11-20 04:54:32 +00:00
let query = String::from_utf8(query).unwrap();
2014-10-05 03:08:44 +00:00
let (stmt_name, _, result_desc) = try!(self.raw_prepare(query[]));
2014-09-30 05:56:43 +00:00
let column_types = result_desc.iter().map(|desc| desc.ty.clone()).collect();
2014-10-05 03:08:44 +00:00
try!(self.close_statement(stmt_name[]));
2014-09-30 05:56:43 +00:00
2014-11-20 04:54:32 +00:00
let mut query = vec![];
2014-11-20 18:54:40 +00:00
let _ = write!(&mut query, "COPY {} (", table);
2014-09-30 05:56:43 +00:00
let _ = util::comma_join(&mut query, rows.iter().map(|&e| e));
2014-11-20 18:54:40 +00:00
let _ = write!(&mut query, ") FROM STDIN WITH (FORMAT binary)");
2014-11-20 04:54:32 +00:00
let query = String::from_utf8(query).unwrap();
2014-10-05 03:08:44 +00:00
let (stmt_name, _, _) = try!(self.raw_prepare(query[]));
2014-09-30 05:56:43 +00:00
Ok(CopyInStatement {
2014-09-30 05:56:43 +00:00
conn: conn,
name: stmt_name,
column_types: column_types,
finished: false,
})
}
2014-11-01 23:12:05 +00:00
fn close_statement(&mut self, stmt_name: &str) -> Result<()> {
2014-11-19 17:58:30 +00:00
try!(self.write_messages(&[
2014-09-30 05:56:43 +00:00
Close {
variant: b'S',
name: stmt_name,
},
Sync]));
loop {
2014-11-17 06:54:57 +00:00
match try!(self.read_message()) {
2014-09-30 05:56:43 +00:00
ReadyForQuery { .. } => break,
ErrorResponse { fields } => {
try!(self.wait_for_ready());
2014-11-04 06:24:11 +00:00
return DbError::new(fields);
2014-09-30 05:56:43 +00:00
}
_ => {}
}
}
Ok(())
}
2014-11-18 02:20:48 +00:00
fn set_type_names<'a, I>(&mut self, mut it: I) -> Result<()> where I: Iterator<&'a mut Type> {
2014-03-30 23:19:04 +00:00
for ty in it {
2014-11-04 05:29:16 +00:00
if let &Type::Unknown { oid, ref mut name } = ty {
2014-10-26 00:58:46 +00:00
*name = try!(self.get_type_name(oid));
2014-03-30 23:19:04 +00:00
}
}
2014-08-16 02:50:11 +00:00
2014-03-30 23:19:04 +00:00
Ok(())
}
2014-11-01 23:12:05 +00:00
fn get_type_name(&mut self, oid: Oid) -> Result<String> {
2014-11-07 16:54:10 +00:00
if let Some(name) = self.unknown_types.get(&oid) {
2014-10-14 04:12:25 +00:00
return Ok(name.clone());
2013-12-04 08:18:28 +00:00
}
2014-04-03 05:56:16 +00:00
let name = try!(self.quick_query(format!("SELECT typname FROM pg_type \
2014-10-05 03:08:44 +00:00
WHERE oid={}", oid)[]))
2014-09-18 05:57:23 +00:00
.into_iter().next().unwrap().into_iter().next().unwrap().unwrap();
2013-12-04 08:18:28 +00:00
self.unknown_types.insert(oid, name.clone());
Ok(name)
2013-12-04 08:18:28 +00:00
}
2014-03-30 23:19:04 +00:00
fn is_desynchronized(&self) -> bool {
self.desynchronized
}
fn canary(&self) -> u32 {
self.canary
}
2014-11-01 23:12:05 +00:00
fn wait_for_ready(&mut self) -> Result<()> {
2014-11-17 06:54:57 +00:00
match try!(self.read_message()) {
ReadyForQuery { .. } => Ok(()),
2014-07-10 19:35:57 +00:00
_ => bad_response!(self)
}
}
2013-12-04 07:48:46 +00:00
2014-11-01 23:12:05 +00:00
fn quick_query(&mut self, query: &str) -> Result<Vec<Vec<Option<String>>>> {
check_desync!(self);
2014-11-19 17:58:30 +00:00
try!(self.write_messages(&[Query { query: query }]));
2013-12-04 07:48:46 +00:00
2014-05-26 18:41:18 +00:00
let mut result = vec![];
2013-12-04 07:48:46 +00:00
loop {
2014-11-17 06:54:57 +00:00
match try!(self.read_message()) {
2013-12-04 07:48:46 +00:00
ReadyForQuery { .. } => break,
2014-04-26 21:46:38 +00:00
DataRow { row } => {
2014-09-18 05:57:23 +00:00
result.push(row.into_iter().map(|opt| {
2014-10-05 03:08:44 +00:00
opt.map(|b| String::from_utf8_lossy(b[]).into_string())
2014-05-16 02:27:19 +00:00
}).collect());
2014-04-26 21:46:38 +00:00
}
2014-10-06 00:31:25 +00:00
CopyInResponse { .. } => {
2014-11-19 17:58:30 +00:00
try!(self.write_messages(&[
2014-10-06 00:31:25 +00:00
CopyFail {
message: "COPY queries cannot be directly executed",
},
Sync]));
}
ErrorResponse { fields } => {
2014-02-22 07:18:39 +00:00
try!(self.wait_for_ready());
2014-11-04 06:24:11 +00:00
return DbError::new(fields);
}
2013-12-04 07:48:46 +00:00
_ => {}
}
}
Ok(result)
2013-12-04 07:48:46 +00:00
}
2014-11-01 23:12:05 +00:00
fn finish_inner(&mut self) -> Result<()> {
check_desync!(self);
self.canary = 0;
2014-11-19 17:58:30 +00:00
try!(self.write_messages(&[Terminate]));
2014-04-26 06:14:55 +00:00
Ok(())
}
}
2013-09-30 02:12:20 +00:00
/// A connection to a Postgres database.
2014-11-01 23:21:47 +00:00
pub struct Connection {
conn: RefCell<InnerConnection>
}
2014-11-01 23:21:47 +00:00
impl Connection {
/// Creates a new connection to a Postgres database.
2013-09-30 02:12:20 +00:00
///
/// Most applications can use a URL string in the normal format:
2013-09-30 02:12:20 +00:00
///
2014-03-09 06:01:24 +00:00
/// ```notrust
/// postgresql://user[:password]@host[:port][/database][?param1=val1[[&param2=val2]...]]
2013-09-30 02:12:20 +00:00
/// ```
///
/// 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.
2014-03-09 22:22:20 +00:00
///
/// To connect to the server via Unix sockets, `host` should be set to the
2014-04-21 05:53:54 +00:00
/// absolute path of the directory containing the socket file. Since `/` is
/// a reserved character in URLs, the path should be URL encoded. If the
2014-11-01 23:15:30 +00:00
/// path contains non-UTF 8 characters, a `ConnectParams` struct
/// should be created manually and passed in. Note that Postgres does not
/// support SSL over Unix sockets.
///
2014-07-30 02:48:56 +00:00
/// ## Examples
2014-03-09 22:22:20 +00:00
///
2014-03-10 17:06:40 +00:00
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
2014-08-20 04:49:27 +00:00
/// # let _ = || {
/// let url = "postgresql://postgres:hunter2@localhost:2994/foodb";
2014-11-17 21:46:33 +00:00
/// let conn = try!(Connection::connect(url, &SslMode::None));
2014-08-20 04:49:27 +00:00
/// # Ok(()) };
2014-03-09 22:22:20 +00:00
/// ```
///
2014-04-19 18:10:27 +00:00
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
2014-08-20 04:49:27 +00:00
/// # let _ = || {
/// let url = "postgresql://postgres@%2Frun%2Fpostgres";
2014-11-17 21:46:33 +00:00
/// let conn = try!(Connection::connect(url, &SslMode::None));
2014-08-20 04:49:27 +00:00
/// # Ok(()) };
/// ```
///
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, UserInfo, ConnectParams, SslMode, ConnectTarget};
2014-08-20 04:49:27 +00:00
/// # let _ = || {
/// # let some_crazy_path = Path::new("");
2014-11-01 23:15:30 +00:00
/// let params = ConnectParams {
/// target: ConnectTarget::Unix(some_crazy_path),
/// port: None,
2014-11-01 23:14:08 +00:00
/// user: Some(UserInfo {
2014-10-18 18:17:12 +00:00
/// user: "postgres".into_string(),
/// password: None
/// }),
/// database: None,
2014-05-26 18:41:18 +00:00
/// options: vec![],
/// };
2014-11-17 21:46:33 +00:00
/// let conn = try!(Connection::connect(params, &SslMode::None));
2014-08-20 04:49:27 +00:00
/// # Ok(()) };
/// ```
2014-11-04 06:24:11 +00:00
pub fn connect<T>(params: T, ssl: &SslMode) -> result::Result<Connection, ConnectError>
2014-08-17 20:57:58 +00:00
where T: IntoConnectParams {
2014-11-01 23:21:47 +00:00
InnerConnection::connect(params, ssl).map(|conn| {
Connection { conn: RefCell::new(conn) }
})
}
2013-09-30 02:12:20 +00:00
/// Sets the notice handler for the connection, returning the old handler.
2014-11-01 23:16:50 +00:00
pub fn set_notice_handler(&self, handler: Box<NoticeHandler+Send>) -> Box<NoticeHandler+Send> {
2014-03-19 04:00:06 +00:00
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.
2014-11-01 23:18:09 +00:00
pub fn notifications<'a>(&'a self) -> Notifications<'a> {
Notifications { conn: self }
}
/// Creates a new prepared statement.
2013-09-30 02:12:20 +00:00
///
/// 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.
2014-03-09 22:22:20 +00:00
///
2014-07-30 02:48:56 +00:00
/// ## Example
2014-03-09 22:22:20 +00:00
///
2014-03-10 17:06:40 +00:00
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
/// # let conn = Connection::connect("", &SslMode::None).unwrap();
/// let maybe_stmt = conn.prepare("SELECT foo FROM bar WHERE baz = $1");
2014-03-09 22:22:20 +00:00
/// let stmt = match maybe_stmt {
/// Ok(stmt) => stmt,
2014-10-30 04:26:03 +00:00
/// Err(err) => panic!("Error preparing statement: {}", err)
2014-03-09 22:22:20 +00:00
/// };
2014-11-01 23:24:24 +00:00
pub fn prepare<'a>(&'a self, query: &str) -> Result<Statement<'a>> {
let mut conn = self.conn.borrow_mut();
if conn.trans_depth != 0 {
return Err(Error::WrongTransaction);
}
conn.prepare(query, self)
}
2014-09-30 07:01:15 +00:00
/// 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
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
2014-09-30 07:01:15 +00:00
/// # use postgres::types::ToSql;
/// # let _ = || {
2014-11-17 21:46:33 +00:00
/// # let conn = Connection::connect("", &SslMode::None).unwrap();
2014-09-30 07:01:15 +00:00
/// try!(conn.execute("CREATE TABLE foo (
/// bar INT PRIMARY KEY,
/// baz VARCHAR
/// )", &[]));
2014-09-30 07:01:15 +00:00
///
/// let stmt = try!(conn.prepare_copy_in("foo", &["bar", "baz"]));
2014-10-18 18:17:12 +00:00
/// let data: &[&[&ToSql]] = &[&[&0i32, &"blah".into_string()],
2014-09-30 07:01:15 +00:00
/// &[&1i32, &None::<String>]];
/// try!(stmt.execute(data.iter().map(|r| r.iter().map(|&e| e))));
/// # Ok(()) };
/// ```
2014-09-30 05:56:43 +00:00
pub fn prepare_copy_in<'a>(&'a self, table: &str, rows: &[&str])
-> Result<CopyInStatement<'a>> {
2014-09-30 05:56:43 +00:00
let mut conn = self.conn.borrow_mut();
if conn.trans_depth != 0 {
return Err(Error::WrongTransaction);
2014-09-30 05:56:43 +00:00
}
conn.prepare_copy_in(table, rows, self)
}
/// Begins a new transaction.
2013-09-30 02:12:20 +00:00
///
2014-11-01 23:25:11 +00:00
/// Returns a `Transaction` object which should be used instead of
2013-10-14 01:58:31 +00:00
/// the connection for the duration of the transaction. The transaction
2014-11-01 23:25:11 +00:00
/// is active until the `Transaction` object falls out of scope.
2014-07-26 05:41:10 +00:00
///
2014-07-30 02:48:56 +00:00
/// ## Note
2014-10-06 03:49:09 +00:00
/// A transaction will roll back by default. The `set_commit`,
/// `set_rollback`, and `commit` methods alter this behavior.
2014-03-09 22:22:20 +00:00
///
2014-07-30 02:48:56 +00:00
/// ## Example
2014-03-09 22:22:20 +00:00
///
2014-03-10 17:06:40 +00:00
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
2014-11-04 06:24:11 +00:00
/// # fn foo() -> Result<(), postgres::Error> {
2014-11-17 21:46:33 +00:00
/// # let conn = Connection::connect("", &SslMode::None).unwrap();
/// let trans = try!(conn.transaction());
/// try!(trans.execute("UPDATE foo SET bar = 10", &[]));
2014-07-26 05:41:10 +00:00
/// // ...
2014-03-09 22:22:20 +00:00
///
2014-10-06 03:47:48 +00:00
/// try!(trans.commit());
2014-03-09 22:22:20 +00:00
/// # Ok(())
/// # }
/// ```
2014-11-01 23:25:11 +00:00
pub fn transaction<'a>(&'a self) -> Result<Transaction<'a>> {
let mut conn = self.conn.borrow_mut();
check_desync!(conn);
if conn.trans_depth != 0 {
return Err(Error::WrongTransaction);
}
try!(conn.quick_query("BEGIN"));
conn.trans_depth += 1;
2014-11-01 23:25:11 +00:00
Ok(Transaction {
2013-08-27 05:40:23 +00:00
conn: self,
2014-07-26 05:41:10 +00:00
commit: Cell::new(false),
depth: 1,
finished: false,
})
}
/// A convenience function for queries that are only run once.
2013-09-30 02:12:20 +00:00
///
/// 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.
2014-11-01 23:12:05 +00:00
pub fn execute(&self, query: &str, params: &[&ToSql]) -> Result<uint> {
self.prepare(query).and_then(|stmt| stmt.execute(params))
2013-09-01 18:06:33 +00:00
}
2014-06-08 23:07:15 +00:00
/// 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.
///
2014-07-30 02:48:56 +00:00
/// ## Warning
2014-06-08 23:07:15 +00:00
///
/// 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.
///
2014-07-30 02:48:56 +00:00
/// ## Example
2014-06-08 23:07:15 +00:00
///
2014-06-08 23:12:27 +00:00
/// ```rust,no_run
2014-11-01 23:21:47 +00:00
/// # use postgres::{Connection, Result};
/// fn init_db(conn: &Connection) -> Result<()> {
2014-07-26 00:56:43 +00:00
/// conn.batch_execute("
2014-06-08 23:07:15 +00:00
/// 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);
2014-07-26 00:56:43 +00:00
/// ")
2014-06-08 23:07:15 +00:00
/// }
/// ```
2014-11-01 23:12:05 +00:00
pub fn batch_execute(&self, query: &str) -> Result<()> {
2014-06-08 23:07:15 +00:00
let mut conn = self.conn.borrow_mut();
if conn.trans_depth != 0 {
return Err(Error::WrongTransaction);
2014-06-08 23:07:15 +00:00
}
conn.quick_query(query).map(|_| ())
}
2013-10-21 00:32:14 +00:00
/// 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.
2014-11-01 23:19:02 +00:00
pub fn cancel_data(&self) -> CancelData {
2014-03-19 04:00:06 +00:00
self.conn.borrow().cancel_data
2013-10-21 00:32:14 +00:00
}
/// 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 {
2014-03-19 04:00:06 +00:00
self.conn.borrow().is_desynchronized()
}
/// Consumes the connection, closing it.
///
2014-11-03 00:48:38 +00:00
/// Functionally equivalent to the `Drop` implementation for `Connection`
/// except that it returns any error encountered to the caller.
2014-11-01 23:12:05 +00:00
pub fn finish(self) -> Result<()> {
2014-03-19 04:00:06 +00:00
let mut conn = self.conn.borrow_mut();
conn.finished = true;
conn.finish_inner()
}
fn canary(&self) -> u32 {
2014-04-10 04:28:20 +00:00
self.conn.borrow().canary()
}
fn write_messages(&self, messages: &[FrontendMessage]) -> IoResult<()> {
2014-03-19 04:00:06 +00:00
self.conn.borrow_mut().write_messages(messages)
2013-08-17 22:09:26 +00:00
}
}
2013-11-10 04:58:38 +00:00
/// Specifies the SSL support requested for a new connection
pub enum SslMode {
2013-11-10 04:58:38 +00:00
/// The connection will not use SSL
2014-11-17 21:46:33 +00:00
None,
2013-11-10 04:58:38 +00:00
/// The connection will use SSL if the backend supports it
2014-11-17 21:46:33 +00:00
Prefer(SslContext),
2013-11-10 04:58:38 +00:00
/// The connection must use SSL
2014-11-17 21:46:33 +00:00
Require(SslContext)
2013-11-10 04:58:38 +00:00
}
2014-05-19 02:46:21 +00:00
/// Represents a transaction on a database connection.
///
2014-07-26 05:41:10 +00:00
/// The transaction will roll back by default.
2014-11-01 23:25:11 +00:00
pub struct Transaction<'conn> {
2014-11-01 23:21:47 +00:00
conn: &'conn Connection,
commit: Cell<bool>,
depth: u32,
finished: bool,
}
#[unsafe_destructor]
2014-11-01 23:25:11 +00:00
impl<'conn> Drop for Transaction<'conn> {
2013-09-18 05:06:47 +00:00
fn drop(&mut self) {
if !self.finished {
let _ = self.finish_inner();
}
}
}
2014-11-01 23:25:11 +00:00
impl<'conn> Transaction<'conn> {
2014-11-01 23:12:05 +00:00
fn finish_inner(&mut self) -> Result<()> {
let mut conn = self.conn.conn.borrow_mut();
debug_assert!(self.depth == conn.trans_depth);
2014-07-26 05:41:10 +00:00
let query = match (self.commit.get(), self.depth != 1) {
(false, true) => "ROLLBACK TO sp",
(false, false) => "ROLLBACK",
(true, true) => "RELEASE sp",
(true, false) => "COMMIT",
2014-05-18 18:37:52 +00:00
};
conn.trans_depth -= 1;
conn.quick_query(query).map(|_| ())
}
2013-08-27 05:40:23 +00:00
2014-11-01 23:21:47 +00:00
/// Like `Connection::prepare`.
2014-11-01 23:24:24 +00:00
pub fn prepare<'a>(&'a self, query: &str) -> Result<Statement<'a>> {
let mut conn = self.conn.conn.borrow_mut();
if conn.trans_depth != self.depth {
return Err(Error::WrongTransaction);
}
conn.prepare(query, self.conn)
2013-09-01 18:06:33 +00:00
}
2014-11-01 23:21:47 +00:00
/// Like `Connection::prepare_copy_in`.
2014-09-30 06:49:58 +00:00
pub fn prepare_copy_in<'a>(&'a self, table: &str, cols: &[&str])
-> Result<CopyInStatement<'a>> {
2014-09-30 06:49:58 +00:00
let mut conn = self.conn.conn.borrow_mut();
if conn.trans_depth != self.depth {
return Err(Error::WrongTransaction);
2014-09-30 06:49:58 +00:00
}
conn.prepare_copy_in(table, cols, self.conn)
}
2014-11-01 23:21:47 +00:00
/// Like `Connection::execute`.
2014-11-01 23:12:05 +00:00
pub fn execute(&self, query: &str, params: &[&ToSql]) -> Result<uint> {
self.prepare(query).and_then(|s| s.execute(params))
2013-09-30 02:12:20 +00:00
}
2014-11-01 23:21:47 +00:00
/// Like `Connection::batch_execute`.
2014-11-01 23:12:05 +00:00
pub fn batch_execute(&self, query: &str) -> Result<()> {
2014-07-30 02:48:56 +00:00
let mut conn = self.conn.conn.borrow_mut();
if conn.trans_depth != self.depth {
return Err(Error::WrongTransaction);
2014-06-08 23:07:15 +00:00
}
2014-07-30 02:48:56 +00:00
conn.quick_query(query).map(|_| ())
2014-06-08 23:07:15 +00:00
}
2014-11-01 23:21:47 +00:00
/// Like `Connection::transaction`.
2014-11-01 23:25:11 +00:00
pub fn transaction<'a>(&'a self) -> Result<Transaction<'a>> {
let mut conn = self.conn.conn.borrow_mut();
check_desync!(conn);
if conn.trans_depth != self.depth {
return Err(Error::WrongTransaction);
}
try!(conn.quick_query("SAVEPOINT sp"));
conn.trans_depth += 1;
2014-11-01 23:25:11 +00:00
Ok(Transaction {
2013-09-05 06:28:44 +00:00
conn: self.conn,
2014-07-26 05:41:10 +00:00
commit: Cell::new(false),
depth: self.depth + 1,
finished: false,
})
}
2014-07-30 02:50:44 +00:00
/// Executes a prepared statement, returning a lazily loaded iterator over
/// the resulting rows.
///
/// No more than `row_limit` rows will be stored in memory at a time. Rows
/// will be pulled from the database in batches of `row_limit` as needed.
/// If `row_limit` is less than or equal to 0, `lazy_query` is equivalent
/// to `query`.
pub fn lazy_query<'trans, 'stmt>(&'trans self,
2014-11-01 23:24:24 +00:00
stmt: &'stmt Statement,
2014-07-30 02:50:44 +00:00
params: &[&ToSql],
row_limit: i32)
2014-11-01 23:28:38 +00:00
-> Result<LazyRows<'trans, 'stmt>> {
2014-07-30 02:50:44 +00:00
if self.conn as *const _ != stmt.conn as *const _ {
return Err(Error::WrongConnection);
2014-07-30 02:50:44 +00:00
}
check_desync!(self.conn);
stmt.lazy_query(row_limit, params).map(|result| {
2014-11-01 23:28:38 +00:00
LazyRows {
2014-07-30 02:50:44 +00:00
_trans: self,
result: result
}
})
}
2013-09-30 02:12:20 +00:00
/// Determines if the transaction is currently set to commit or roll back.
2013-08-27 05:40:23 +00:00
pub fn will_commit(&self) -> bool {
self.commit.get()
2013-08-27 05:40:23 +00:00
}
2013-09-30 02:12:20 +00:00
/// Sets the transaction to commit at its completion.
2013-08-27 05:40:23 +00:00
pub fn set_commit(&self) {
self.commit.set(true);
2013-08-27 05:40:23 +00:00
}
2013-09-30 02:12:20 +00:00
/// Sets the transaction to roll back at its completion.
2013-08-27 05:40:23 +00:00
pub fn set_rollback(&self) {
self.commit.set(false);
2013-08-27 05:40:23 +00:00
}
2014-07-26 05:41:10 +00:00
/// A convenience method which consumes and commits a transaction.
2014-11-01 23:12:05 +00:00
pub fn commit(self) -> Result<()> {
2014-07-26 05:41:10 +00:00
self.set_commit();
self.finish()
}
/// Consumes the transaction, commiting or rolling it back as appropriate.
///
/// Functionally equivalent to the `Drop` implementation of
2014-11-01 23:25:11 +00:00
/// `Transaction` except that it returns any error to the caller.
2014-11-01 23:12:05 +00:00
pub fn finish(mut self) -> Result<()> {
self.finished = true;
self.finish_inner()
}
2013-08-27 05:40:23 +00:00
}
/// A prepared statement
2014-11-01 23:24:24 +00:00
pub struct Statement<'conn> {
2014-11-01 23:21:47 +00:00
conn: &'conn Connection,
2014-05-26 03:38:40 +00:00
name: String,
2014-11-04 05:29:16 +00:00
param_types: Vec<Type>,
result_desc: Vec<ResultDescription>,
next_portal_id: Cell<uint>,
2014-04-26 06:14:55 +00:00
finished: bool,
}
#[unsafe_destructor]
2014-11-01 23:24:24 +00:00
impl<'conn> Drop for Statement<'conn> {
fn drop(&mut self) {
2014-04-26 06:14:55 +00:00
if !self.finished {
let _ = self.finish_inner();
}
}
}
2014-11-01 23:24:24 +00:00
impl<'conn> Statement<'conn> {
2014-11-01 23:12:05 +00:00
fn finish_inner(&mut self) -> Result<()> {
2014-09-30 05:56:43 +00:00
let mut conn = self.conn.conn.borrow_mut();
check_desync!(conn);
2014-10-05 03:08:44 +00:00
conn.close_statement(self.name[])
}
2014-11-18 02:20:48 +00:00
fn inner_execute(&self, portal_name: &str, row_limit: i32, params: &[&ToSql]) -> Result<()> {
let mut conn = self.conn.conn.borrow_mut();
2014-03-28 04:39:03 +00:00
if self.param_types.len() != params.len() {
return Err(Error::WrongParamCount {
2014-03-28 04:39:03 +00:00
expected: self.param_types.len(),
actual: params.len(),
});
}
2014-05-26 18:41:18 +00:00
let mut values = vec![];
for (param, ty) in params.iter().zip(self.param_types.iter()) {
2014-10-14 04:12:25 +00:00
values.push(try!(param.to_sql(ty)));
};
2014-11-19 17:58:30 +00:00
try!(conn.write_messages(&[
Bind {
portal: portal_name,
2014-10-05 03:08:44 +00:00
statement: self.name[],
2014-11-19 17:58:30 +00:00
formats: &[1],
2014-10-05 03:08:44 +00:00
values: values[],
2014-11-19 17:58:30 +00:00
result_formats: &[1]
},
Execute {
portal: portal_name,
max_rows: row_limit
},
Sync]));
2014-11-17 06:54:57 +00:00
match try!(conn.read_message()) {
BindComplete => Ok(()),
ErrorResponse { fields } => {
2014-09-30 00:43:21 +00:00
try!(conn.wait_for_ready());
2014-11-04 06:24:11 +00:00
DbError::new(fields)
}
2014-07-07 02:02:22 +00:00
_ => {
conn.desynchronized = true;
Err(Error::BadResponse)
2014-07-07 02:02:22 +00:00
}
}
}
2014-11-18 02:20:48 +00:00
fn lazy_query<'a>(&'a self, row_limit: i32, params: &[&ToSql]) -> Result<Rows<'a>> {
let id = self.next_portal_id.get();
self.next_portal_id.set(id + 1);
2014-05-24 03:08:14 +00:00
let portal_name = format!("{}p{}", self.name, id);
2014-10-05 03:08:44 +00:00
try!(self.inner_execute(portal_name[], row_limit, params));
2014-11-01 23:27:30 +00:00
let mut result = Rows {
stmt: self,
name: portal_name,
data: RingBuf::new(),
row_limit: row_limit,
more_rows: true,
finished: false,
};
try!(result.read_rows())
Ok(result)
}
/// Returns a slice containing the expected parameter types.
2014-11-04 05:29:16 +00:00
pub fn param_types(&self) -> &[Type] {
2014-10-05 03:08:44 +00:00
self.param_types[]
}
/// Returns a slice describing the columns of the result of the query.
2014-07-22 05:46:40 +00:00
pub fn result_descriptions(&self) -> &[ResultDescription] {
2014-10-05 03:08:44 +00:00
self.result_desc[]
}
/// Executes the prepared statement, returning the number of rows modified.
///
/// If the statement does not modify any rows (e.g. SELECT), 0 is returned.
///
2014-07-30 02:48:56 +00:00
/// ## Example
///
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
/// # let conn = Connection::connect("", &SslMode::None).unwrap();
/// # let bar = 1i32;
/// # let baz = true;
/// let stmt = conn.prepare("UPDATE foo SET bar = $1 WHERE baz = $2").unwrap();
/// match stmt.execute(&[&bar, &baz]) {
/// Ok(count) => println!("{} row(s) updated", count),
/// Err(err) => println!("Error executing query: {}", err)
/// }
2014-11-01 23:12:05 +00:00
pub fn execute(&self, params: &[&ToSql]) -> Result<uint> {
check_desync!(self.conn);
try!(self.inner_execute("", 0, params));
let mut conn = self.conn.conn.borrow_mut();
let num;
loop {
2014-11-17 06:54:57 +00:00
match try!(conn.read_message()) {
DataRow { .. } => {}
ErrorResponse { fields } => {
try!(conn.wait_for_ready());
2014-11-04 06:24:11 +00:00
return DbError::new(fields);
}
CommandComplete { tag } => {
num = util::parse_update_count(tag);
break;
}
EmptyQueryResponse => {
num = 0;
break;
}
2014-09-30 04:05:42 +00:00
CopyInResponse { .. } => {
2014-11-19 17:58:30 +00:00
try!(conn.write_messages(&[
2014-09-30 04:05:42 +00:00
CopyFail {
message: "COPY queries cannot be directly executed",
},
Sync]));
}
2014-07-07 02:02:22 +00:00
_ => {
conn.desynchronized = true;
return Err(Error::BadResponse);
2014-07-07 02:02:22 +00:00
}
}
}
try!(conn.wait_for_ready());
Ok(num)
}
/// Executes the prepared statement, returning an iterator over the
/// resulting rows.
///
2014-07-30 02:48:56 +00:00
/// ## Example
///
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
/// # let conn = Connection::connect("", &SslMode::None).unwrap();
/// let stmt = conn.prepare("SELECT foo FROM bar WHERE baz = $1").unwrap();
/// # let baz = true;
/// let mut rows = match stmt.query(&[&baz]) {
/// Ok(rows) => rows,
2014-10-30 04:26:03 +00:00
/// Err(err) => panic!("Error running query: {}", err)
/// };
/// for row in rows {
2014-07-10 19:35:57 +00:00
/// let foo: i32 = row.get("foo");
/// println!("foo: {}", foo);
/// }
/// ```
2014-11-01 23:27:30 +00:00
pub fn query<'a>(&'a self, params: &[&ToSql]) -> Result<Rows<'a>> {
check_desync!(self.conn);
self.lazy_query(0, params)
}
/// Consumes the statement, clearing it from the Postgres session.
///
/// Functionally identical to the `Drop` implementation of the
2014-11-01 23:24:24 +00:00
/// `Statement` except that it returns any error to the caller.
2014-11-01 23:12:05 +00:00
pub fn finish(mut self) -> Result<()> {
2014-04-26 06:14:55 +00:00
self.finished = true;
self.finish_inner()
}
}
/// Information about a column of the result of a query.
2014-06-02 02:57:27 +00:00
#[deriving(PartialEq, Eq)]
pub struct ResultDescription {
/// The name of the column
2014-05-26 03:38:40 +00:00
pub name: String,
/// The type of the data in the column
2014-11-04 05:29:16 +00:00
pub ty: Type
}
/// An iterator over the resulting rows of a query.
2014-11-01 23:27:30 +00:00
pub struct Rows<'stmt> {
2014-11-01 23:24:24 +00:00
stmt: &'stmt Statement<'stmt>,
2014-05-26 03:38:40 +00:00
name: String,
2014-04-08 03:02:05 +00:00
data: RingBuf<Vec<Option<Vec<u8>>>>,
row_limit: i32,
more_rows: bool,
finished: bool,
}
#[unsafe_destructor]
2014-11-01 23:27:30 +00:00
impl<'stmt> Drop for Rows<'stmt> {
fn drop(&mut self) {
if !self.finished {
let _ = self.finish_inner();
}
}
}
2014-11-01 23:27:30 +00:00
impl<'stmt> Rows<'stmt> {
2014-11-01 23:12:05 +00:00
fn finish_inner(&mut self) -> Result<()> {
let mut conn = self.stmt.conn.conn.borrow_mut();
check_desync!(conn);
2014-11-19 17:58:30 +00:00
try!(conn.write_messages(&[
Close {
2014-09-06 23:33:43 +00:00
variant: b'P',
2014-10-05 03:08:44 +00:00
name: self.name[]
},
Sync]));
loop {
2014-11-17 06:54:57 +00:00
match try!(conn.read_message()) {
ReadyForQuery { .. } => break,
ErrorResponse { fields } => {
try!(conn.wait_for_ready());
2014-11-04 06:24:11 +00:00
return DbError::new(fields);
}
_ => {}
}
}
Ok(())
}
2014-11-01 23:12:05 +00:00
fn read_rows(&mut self) -> Result<()> {
let mut conn = self.stmt.conn.conn.borrow_mut();
loop {
2014-11-17 06:54:57 +00:00
match try!(conn.read_message()) {
2014-08-16 02:50:11 +00:00
EmptyQueryResponse | CommandComplete { .. } => {
self.more_rows = false;
break;
},
PortalSuspended => {
self.more_rows = true;
break;
},
2014-11-07 16:54:10 +00:00
DataRow { row } => self.data.push_back(row),
ErrorResponse { fields } => {
try!(conn.wait_for_ready());
2014-11-04 06:24:11 +00:00
return DbError::new(fields);
}
2014-09-30 04:05:42 +00:00
CopyInResponse { .. } => {
2014-11-19 17:58:30 +00:00
try!(conn.write_messages(&[
2014-09-30 04:05:42 +00:00
CopyFail {
message: "COPY queries cannot be directly executed",
},
Sync]));
}
2014-07-07 02:02:22 +00:00
_ => {
conn.desynchronized = true;
return Err(Error::BadResponse);
2014-07-07 02:02:22 +00:00
}
}
}
conn.wait_for_ready()
}
2014-11-01 23:12:05 +00:00
fn execute(&mut self) -> Result<()> {
2014-11-19 17:58:30 +00:00
try!(self.stmt.conn.write_messages(&[
Execute {
2014-10-05 03:08:44 +00:00
portal: self.name[],
max_rows: self.row_limit
},
Sync]));
self.read_rows()
}
/// Returns a slice describing the columns of the `Rows`.
pub fn result_descriptions(&self) -> &'stmt [ResultDescription] {
self.stmt.result_descriptions()
}
2014-11-01 23:27:30 +00:00
/// Consumes the `Rows`, cleaning up associated state.
///
2014-11-01 23:27:30 +00:00
/// Functionally identical to the `Drop` implementation on `Rows`
/// except that it returns any error to the caller.
2014-11-01 23:12:05 +00:00
pub fn finish(mut self) -> Result<()> {
self.finished = true;
self.finish_inner()
}
2014-11-01 23:27:30 +00:00
fn try_next(&mut self) -> Option<Result<Row<'stmt>>> {
if self.data.is_empty() && self.more_rows {
2014-10-14 04:12:25 +00:00
if let Err(err) = self.execute() {
return Some(Err(err));
2014-03-28 04:20:04 +00:00
}
}
2014-11-03 00:48:38 +00:00
self.data.pop_front().map(|row| Ok(Row { stmt: self.stmt, data: row }))
}
}
2014-11-01 23:27:30 +00:00
impl<'stmt> Iterator<Row<'stmt>> for Rows<'stmt> {
2014-04-03 05:56:16 +00:00
#[inline]
2014-11-01 23:27:30 +00:00
fn next(&mut self) -> Option<Row<'stmt>> {
2014-03-28 04:20:04 +00:00
// we'll never hit the network on a non-lazy result
self.try_next().map(|r| r.unwrap())
}
2014-04-03 05:56:16 +00:00
#[inline]
fn size_hint(&self) -> (uint, Option<uint>) {
let lower = self.data.len();
let upper = if self.more_rows {
2014-11-21 05:48:05 +00:00
None
} else {
Some(lower)
};
(lower, upper)
}
}
/// A single result row of a query.
2014-11-01 23:27:30 +00:00
pub struct Row<'stmt> {
2014-11-01 23:24:24 +00:00
stmt: &'stmt Statement<'stmt>,
2014-04-08 03:02:05 +00:00
data: Vec<Option<Vec<u8>>>
}
2014-11-01 23:27:30 +00:00
impl<'stmt> Row<'stmt> {
/// Returns the number of values in the row
#[inline]
pub fn len(&self) -> uint {
self.data.len()
}
/// Returns a slice describing the columns of the `Row`.
pub fn result_descriptions(&self) -> &'stmt [ResultDescription] {
self.stmt.result_descriptions()
}
/// Retrieves the contents of a field of the row.
///
/// A field can be accessed by the name or index of its column, though
2014-06-30 06:05:26 +00:00
/// access by index is more efficient. Rows are 0-indexed.
///
/// Returns an `Error` value if the index does not reference a column or
/// the return type is not compatible with the Postgres type.
2014-11-01 23:12:05 +00:00
pub fn get_opt<I, T>(&self, idx: I) -> Result<T> where I: RowIndex, T: FromSql {
let idx = try!(idx.idx(self.stmt).ok_or(Error::InvalidColumn));
2014-08-17 20:57:58 +00:00
FromSql::from_sql(&self.stmt.result_desc[idx].ty, &self.data[idx])
}
2014-07-10 19:35:57 +00:00
/// Retrieves the contents of a field of the row.
///
/// A field can be accessed by the name or index of its column, though
/// access by index is more efficient. Rows are 0-indexed.
///
2014-07-30 02:48:56 +00:00
/// ## Failure
///
/// Fails if the index does not reference a column or the return type is
/// not compatible with the Postgres type.
///
2014-07-30 02:48:56 +00:00
/// ## Example
///
/// ```rust,no_run
2014-11-17 21:46:33 +00:00
/// # use postgres::{Connection, SslMode};
/// # let conn = Connection::connect("", &SslMode::None).unwrap();
/// # let stmt = conn.prepare("").unwrap();
/// # let mut result = stmt.query(&[]).unwrap();
/// # let row = result.next().unwrap();
2014-07-10 19:35:57 +00:00
/// let foo: i32 = row.get(0u);
/// let bar: String = row.get("bar");
/// ```
2014-08-17 20:57:58 +00:00
pub fn get<I, T>(&self, idx: I) -> T where I: RowIndex + fmt::Show + Clone, T: FromSql {
2014-07-10 19:35:57 +00:00
match self.get_opt(idx.clone()) {
Ok(ok) => ok,
2014-10-30 04:26:03 +00:00
Err(err) => panic!("error retrieving column {}: {}", idx, err)
}
}
}
/// A trait implemented by types that can index into columns of a row.
2014-03-31 02:21:51 +00:00
pub trait RowIndex {
/// Returns the index of the appropriate column, or `None` if no such
/// column exists.
2014-11-01 23:24:24 +00:00
fn idx(&self, stmt: &Statement) -> Option<uint>;
}
impl RowIndex for uint {
#[inline]
2014-11-01 23:24:24 +00:00
fn idx(&self, stmt: &Statement) -> Option<uint> {
if *self > stmt.result_desc.len() {
2014-11-21 05:48:05 +00:00
None
} else {
Some(*self)
}
}
}
impl<'a> RowIndex for &'a str {
2014-05-18 18:37:52 +00:00
#[inline]
2014-11-01 23:24:24 +00:00
fn idx(&self, stmt: &Statement) -> Option<uint> {
2014-10-05 03:08:44 +00:00
stmt.result_descriptions().iter().position(|d| d.name[] == *self)
}
}
/// A lazily-loaded iterator over the resulting rows of a query
2014-11-01 23:28:38 +00:00
pub struct LazyRows<'trans, 'stmt> {
2014-11-01 23:27:30 +00:00
result: Rows<'stmt>,
2014-11-01 23:25:11 +00:00
_trans: &'trans Transaction<'trans>,
}
2014-11-01 23:28:38 +00:00
impl<'trans, 'stmt> LazyRows<'trans, 'stmt> {
2014-11-01 23:27:30 +00:00
/// Like `Rows::finish`.
2014-04-03 05:56:16 +00:00
#[inline]
2014-11-01 23:12:05 +00:00
pub fn finish(self) -> Result<()> {
self.result.finish()
}
}
2014-11-01 23:28:38 +00:00
impl<'trans, 'stmt> Iterator<Result<Row<'stmt>>> for LazyRows<'trans, 'stmt> {
2014-04-03 05:56:16 +00:00
#[inline]
2014-11-01 23:27:30 +00:00
fn next(&mut self) -> Option<Result<Row<'stmt>>> {
2014-03-28 04:20:04 +00:00
self.result.try_next()
}
2014-04-03 05:56:16 +00:00
#[inline]
fn size_hint(&self) -> (uint, Option<uint>) {
self.result.size_hint()
}
}
2014-09-30 05:56:43 +00:00
/// A prepared COPY FROM STDIN statement
pub struct CopyInStatement<'a> {
2014-11-01 23:21:47 +00:00
conn: &'a Connection,
2014-09-30 05:56:43 +00:00
name: String,
2014-11-04 05:29:16 +00:00
column_types: Vec<Type>,
2014-09-30 05:56:43 +00:00
finished: bool,
}
#[unsafe_destructor]
impl<'a> Drop for CopyInStatement<'a> {
2014-09-30 05:56:43 +00:00
fn drop(&mut self) {
if !self.finished {
let _ = self.finish_inner();
}
}
}
impl<'a> CopyInStatement<'a> {
2014-11-01 23:12:05 +00:00
fn finish_inner(&mut self) -> Result<()> {
2014-09-30 05:56:43 +00:00
let mut conn = self.conn.conn.borrow_mut();
check_desync!(conn);
2014-10-05 03:08:44 +00:00
conn.close_statement(self.name[])
2014-09-30 05:56:43 +00:00
}
2014-09-30 07:15:17 +00:00
/// Returns a slice containing the expected column types.
2014-11-04 05:29:16 +00:00
pub fn column_types(&self) -> &[Type] {
2014-10-05 03:08:44 +00:00
self.column_types[]
2014-09-30 07:15:17 +00:00
}
/// Executes the prepared statement.
///
2014-10-26 03:26:45 +00:00
/// Each iterator returned by the `rows` iterator will be interpreted as
/// providing a single result row.
///
/// Returns the number of rows copied.
2014-11-01 23:12:05 +00:00
pub fn execute<'b, I, J>(&self, mut rows: I) -> Result<uint>
2014-09-30 05:56:43 +00:00
where I: Iterator<J>, J: Iterator<&'b ToSql + 'b> {
let mut conn = self.conn.conn.borrow_mut();
2014-11-19 17:58:30 +00:00
try!(conn.write_messages(&[
2014-09-30 05:56:43 +00:00
Bind {
portal: "",
2014-10-05 03:08:44 +00:00
statement: self.name[],
2014-11-19 17:58:30 +00:00
formats: &[],
values: &[],
result_formats: &[]
2014-09-30 05:56:43 +00:00
},
Execute {
portal: "",
max_rows: 0,
},
Sync]));
2014-11-17 06:54:57 +00:00
match try!(conn.read_message()) {
2014-09-30 05:56:43 +00:00
BindComplete => {},
ErrorResponse { fields } => {
try!(conn.wait_for_ready());
2014-11-04 06:24:11 +00:00
return DbError::new(fields);
2014-09-30 05:56:43 +00:00
}
_ => {
conn.desynchronized = true;
return Err(Error::BadResponse);
2014-09-30 05:56:43 +00:00
}
}
2014-11-17 06:54:57 +00:00
match try!(conn.read_message()) {
2014-09-30 05:56:43 +00:00
CopyInResponse { .. } => {}
_ => {
conn.desynchronized = true;
return Err(Error::BadResponse);
2014-09-30 05:56:43 +00:00
}
}
2014-11-20 04:54:32 +00:00
let mut buf = vec![];
2014-09-30 05:56:43 +00:00
let _ = buf.write(b"PGCOPY\n\xff\r\n\x00");
let _ = buf.write_be_i32(0);
let _ = buf.write_be_i32(0);
2014-10-01 21:40:43 +00:00
'l: for mut row in rows {
2014-09-30 05:56:43 +00:00
let _ = buf.write_be_i16(self.column_types.len() as i16);
let mut types = self.column_types.iter();
loop {
match (row.next(), types.next()) {
(Some(val), Some(ty)) => {
2014-10-26 03:26:45 +00:00
match val.to_sql(ty) {
2014-11-21 05:48:05 +00:00
Ok(None) => {
let _ = buf.write_be_i32(-1);
}
2014-10-26 03:26:45 +00:00
Ok(Some(val)) => {
let _ = buf.write_be_i32(val.len() as i32);
2014-10-05 03:08:44 +00:00
let _ = buf.write(val[]);
}
2014-10-26 03:26:45 +00:00
Err(err) => {
// FIXME this is not the right way to handle this
2014-11-17 06:54:57 +00:00
try_desync!(conn, conn.stream.write_message(
2014-10-26 03:26:45 +00:00
&CopyFail {
message: err.to_string()[],
}));
break 'l;
}
}
2014-09-30 05:56:43 +00:00
}
2014-11-21 05:48:05 +00:00
(Some(_), None) | (None, Some(_)) => {
2014-11-17 06:54:57 +00:00
try_desync!(conn, conn.stream.write_message(
&CopyFail {
message: "Invalid column count",
}));
2014-10-01 21:40:43 +00:00
break 'l;
2014-09-30 05:56:43 +00:00
}
2014-11-21 05:48:05 +00:00
(None, None) => break
2014-09-30 05:56:43 +00:00
}
}
2014-11-17 06:54:57 +00:00
try_desync!(conn, conn.stream.write_message(
&CopyData {
2014-11-20 04:54:32 +00:00
data: buf[]
}));
2014-11-20 04:54:32 +00:00
buf.clear();
2014-09-30 05:56:43 +00:00
}
let _ = buf.write_be_i16(-1);
2014-11-19 17:58:30 +00:00
try!(conn.write_messages(&[
2014-10-01 21:40:43 +00:00
CopyData {
2014-11-20 04:54:32 +00:00
data: buf[],
2014-10-01 21:40:43 +00:00
},
CopyDone,
Sync]));
2014-09-30 05:56:43 +00:00
2014-11-17 06:54:57 +00:00
let num = match try!(conn.read_message()) {
CommandComplete { tag } => util::parse_update_count(tag),
2014-09-30 05:56:43 +00:00
ErrorResponse { fields } => {
try!(conn.wait_for_ready());
2014-11-04 06:24:11 +00:00
return DbError::new(fields);
2014-09-30 05:56:43 +00:00
}
_ => {
conn.desynchronized = true;
return Err(Error::BadResponse);
2014-09-30 05:56:43 +00:00
}
};
2014-09-30 05:56:43 +00:00
try!(conn.wait_for_ready());
Ok(num)
2014-09-30 05:56:43 +00:00
}
/// Consumes the statement, clearing it from the Postgres session.
///
/// Functionally identical to the `Drop` implementation of the
/// `CopyInStatement` except that it returns any error to the
2014-09-30 05:56:43 +00:00
/// caller.
2014-11-01 23:12:05 +00:00
pub fn finish(mut self) -> Result<()> {
2014-09-30 05:56:43 +00:00
self.finished = true;
self.finish_inner()
}
}
/// A trait allowing abstraction over connections and transactions
pub trait GenericConnection {
2014-11-01 23:21:47 +00:00
/// Like `Connection::prepare`.
2014-11-01 23:24:24 +00:00
fn prepare<'a>(&'a self, query: &str) -> Result<Statement<'a>>;
2014-11-01 23:21:47 +00:00
/// Like `Connection::execute`.
2014-11-01 23:12:05 +00:00
fn execute(&self, query: &str, params: &[&ToSql]) -> Result<uint> {
self.prepare(query).and_then(|s| s.execute(params))
}
2014-11-01 23:21:47 +00:00
/// Like `Connection::prepare_copy_in`.
fn prepare_copy_in<'a>(&'a self, table: &str, columns: &[&str])
-> Result<CopyInStatement<'a>>;
2014-11-01 23:21:47 +00:00
/// Like `Connection::transaction`.
2014-11-01 23:25:11 +00:00
fn transaction<'a>(&'a self) -> Result<Transaction<'a>>;
2014-11-01 23:21:47 +00:00
/// Like `Connection::batch_execute`.
2014-11-01 23:12:05 +00:00
fn batch_execute(&self, query: &str) -> Result<()>;
}
2014-11-01 23:21:47 +00:00
impl GenericConnection for Connection {
2014-11-01 23:24:24 +00:00
fn prepare<'a>(&'a self, query: &str) -> Result<Statement<'a>> {
self.prepare(query)
}
2014-11-01 23:25:11 +00:00
fn transaction<'a>(&'a self) -> Result<Transaction<'a>> {
self.transaction()
}
fn prepare_copy_in<'a>(&'a self, table: &str, columns: &[&str])
-> Result<CopyInStatement<'a>> {
self.prepare_copy_in(table, columns)
}
2014-11-01 23:12:05 +00:00
fn batch_execute(&self, query: &str) -> Result<()> {
self.batch_execute(query)
}
}
2014-11-01 23:25:11 +00:00
impl<'a> GenericConnection for Transaction<'a> {
2014-11-01 23:24:24 +00:00
fn prepare<'a>(&'a self, query: &str) -> Result<Statement<'a>> {
self.prepare(query)
}
2014-11-01 23:25:11 +00:00
fn transaction<'a>(&'a self) -> Result<Transaction<'a>> {
self.transaction()
}
fn prepare_copy_in<'a>(&'a self, table: &str, columns: &[&str])
-> Result<CopyInStatement<'a>> {
self.prepare_copy_in(table, columns)
}
2014-11-01 23:12:05 +00:00
fn batch_execute(&self, query: &str) -> Result<()> {
self.batch_execute(query)
}
}