Beginnigs of native Postgres frontend
This commit is contained in:
Steven Fackler
4
Makefile
4
Makefile
@@ -1,10 +1,10 @@
|
||||
RUSTC ?= rustc
|
||||
RUSTFLAGS += -L.
|
||||
RUSTFLAGS += -L. --cfg debug
|
||||
|
||||
.PHONY: all
|
||||
all: postgres.dummy
|
||||
|
||||
postgres.dummy: src/lib.rs
|
||||
postgres.dummy: src/lib.rs src/message.rs
|
||||
$(RUSTC) $(RUSTFLAGS) --lib src/lib.rs -o $@
|
||||
touch $@
|
||||
|
||||
|
||||
564
src/lib.rs
564
src/lib.rs
@@ -1,537 +1,107 @@
|
||||
extern mod extra;
|
||||
|
||||
use std::cell::Cell;
|
||||
use std::c_str::{ToCStr, CString};
|
||||
use std::str;
|
||||
use std::ptr;
|
||||
use std::libc::{c_void, c_char, c_int};
|
||||
use std::iterator::RandomAccessIterator;
|
||||
use std::vec;
|
||||
use std::hashmap::HashMap;
|
||||
use std::rt::io::io_error;
|
||||
use std::rt::io::net::ip::SocketAddr;
|
||||
use std::rt::io::net::tcp::TcpStream;
|
||||
use extra::url::Url;
|
||||
|
||||
mod ffi {
|
||||
use std::libc::{c_char, c_int, c_uchar, c_uint, c_void};
|
||||
use message::*;
|
||||
|
||||
pub type PGconn = c_void;
|
||||
pub type PGresult = c_void;
|
||||
pub type Oid = c_uint;
|
||||
|
||||
pub static CONNECTION_OK: c_int = 0;
|
||||
|
||||
pub static PGRES_EMPTY_QUERY: c_int = 0;
|
||||
pub static PGRES_COMMAND_OK: c_int = 1;
|
||||
pub static PGRES_TUPLES_OK: c_int = 2;
|
||||
|
||||
pub static TEXT_FORMAT: c_int = 0;
|
||||
|
||||
// FIXME when FFI gets fixed
|
||||
type PQnoticeProcessor = *c_uchar /*extern "C" fn(*c_void, *c_char)*/;
|
||||
|
||||
#[link_args = "-lpq"]
|
||||
extern "C" {
|
||||
fn PQconnectdb(conninfo: *c_char) -> *PGconn;
|
||||
fn PQsetNoticeProcessor(conn: *PGconn, proc: PQnoticeProcessor,
|
||||
arg: *c_void) -> PQnoticeProcessor;
|
||||
fn PQfinish(conn: *PGconn);
|
||||
fn PQstatus(conn: *PGconn) -> c_int;
|
||||
fn PQerrorMessage(conn: *PGconn) -> *c_char;
|
||||
fn PQexec(conn: *PGconn, query: *c_char) -> *PGresult;
|
||||
fn PQresultStatus(result: *PGresult) -> c_int;
|
||||
fn PQresultErrorMessage(result: *PGresult) -> *c_char;
|
||||
fn PQclear(result: *PGresult);
|
||||
fn PQprepare(conn: *PGconn, stmtName: *c_char, query: *c_char,
|
||||
nParams: c_int, paramTypes: *Oid) -> *PGresult;
|
||||
fn PQdescribePrepared(conn: *PGconn, stmtName: *c_char) -> *PGresult;
|
||||
fn PQexecPrepared(conn: *PGconn, stmtName: *c_char, nParams: c_int,
|
||||
paramValues: **c_char, paramLengths: *c_int,
|
||||
paramFormats: *c_int, resultFormat: c_int)
|
||||
-> *PGresult;
|
||||
fn PQntuples(result: *PGresult) -> c_int;
|
||||
fn PQnfields(result: *PGresult) -> c_int;
|
||||
fn PQnparams(result: *PGresult) -> c_int;
|
||||
fn PQcmdTuples(result: *PGresult) -> *c_char;
|
||||
fn PQgetisnull(result: *PGresult, row_number: c_int, col_number: c_int)
|
||||
-> c_int;
|
||||
fn PQgetvalue(result: *PGresult, row_number: c_int, col_number: c_int)
|
||||
-> *c_char;
|
||||
}
|
||||
}
|
||||
mod message;
|
||||
|
||||
pub struct PostgresConnection {
|
||||
priv conn: *ffi::PGconn,
|
||||
priv next_stmt_id: Cell<uint>,
|
||||
priv stream: Cell<TcpStream>,
|
||||
priv next_stmt_id: Cell<int>
|
||||
}
|
||||
|
||||
extern "C" fn notice_handler(_arg: *c_void, message: *c_char) {
|
||||
let message = unsafe { str::raw::from_c_str(message) };
|
||||
if message.starts_with("DEBUG") {
|
||||
debug!("%s", message);
|
||||
} else if message.starts_with("NOTICE") ||
|
||||
message.starts_with("INFO") ||
|
||||
message.starts_with("LOG") {
|
||||
info!("%s", message);
|
||||
} else if message.starts_with("WARNING") {
|
||||
warn!("%s", message);
|
||||
} else {
|
||||
error!("%s", message);
|
||||
}
|
||||
}
|
||||
|
||||
#[unsafe_destructor]
|
||||
impl Drop for PostgresConnection {
|
||||
fn drop(&self) {
|
||||
unsafe { ffi::PQfinish(self.conn) }
|
||||
do io_error::cond.trap(|_| { }).inside {
|
||||
do self.stream.with_mut_ref |s| {
|
||||
s.write_message(&Terminate);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl PostgresConnection {
|
||||
fn status(&self) -> c_int {
|
||||
unsafe { ffi::PQstatus(self.conn) }
|
||||
}
|
||||
pub fn connect(url: &str, username: &str) -> PostgresConnection {
|
||||
let parsed_url: Url = FromStr::from_str(url).unwrap();
|
||||
|
||||
fn get_error(&self) -> ~str {
|
||||
unsafe { str::raw::from_c_str(ffi::PQerrorMessage(self.conn)) }
|
||||
}
|
||||
|
||||
fn set_notice_processor(&mut self,
|
||||
handler: *u8 /* extern "C" fn(*c_void, *c_char) */,
|
||||
arg: *c_void) {
|
||||
unsafe { ffi::PQsetNoticeProcessor(self.conn, handler, arg); }
|
||||
}
|
||||
}
|
||||
|
||||
impl PostgresConnection {
|
||||
pub fn new(uri: &str) -> Result<~PostgresConnection, ~str> {
|
||||
let mut conn = ~PostgresConnection {
|
||||
conn: do uri.with_c_str |c_uri| {
|
||||
unsafe { ffi::PQconnectdb(c_uri) }
|
||||
},
|
||||
let socket_url = fmt!("%s:%s", parsed_url.host,
|
||||
parsed_url.port.unwrap_or_default(~"5432"));
|
||||
let addr: SocketAddr = FromStr::from_str(socket_url).unwrap();
|
||||
let conn = PostgresConnection {
|
||||
stream: Cell::new(TcpStream::connect(addr).unwrap()),
|
||||
next_stmt_id: Cell::new(0)
|
||||
};
|
||||
|
||||
conn.set_notice_processor(notice_handler, ptr::null());
|
||||
do conn.stream.with_mut_ref |s| {
|
||||
let mut args = HashMap::new();
|
||||
args.insert(~"user", username.to_owned());
|
||||
s.write_message(&StartupMessage(args));
|
||||
}
|
||||
|
||||
match conn.status() {
|
||||
ffi::CONNECTION_OK => Ok(conn),
|
||||
_ => Err(conn.get_error())
|
||||
let resp = do conn.stream.with_mut_ref |s| {
|
||||
s.read_message()
|
||||
};
|
||||
match resp {
|
||||
AuthenticationOk => (),
|
||||
_ => fail!("Bad response: %?", resp)
|
||||
}
|
||||
|
||||
conn.finish_connect();
|
||||
|
||||
conn
|
||||
}
|
||||
|
||||
fn finish_connect(&self) {
|
||||
loop {
|
||||
match self.stream.with_mut_ref(|s| { s.read_message() }) {
|
||||
ParameterStatus(param, value) =>
|
||||
printfln!("Param %s = %s", param, value),
|
||||
BackendKeyData(*) => loop,
|
||||
ReadyForQuery(_) => break,
|
||||
resp => fail!("Bad response: %?", resp)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn prepare<'a>(&'a self, query: &str)
|
||||
-> Result<~PostgresStatement<'a>, ~str> {
|
||||
pub fn prepare<'a>(&'a self, query: &str) -> PostgresStatement<'a> {
|
||||
let id = self.next_stmt_id.take();
|
||||
let name = fmt!("__libpostgres_stmt_%u", id);
|
||||
let query_name = ifmt!("statement_{}", id);
|
||||
self.next_stmt_id.put_back(id + 1);
|
||||
|
||||
let mut res = unsafe {
|
||||
let raw_res = do query.with_c_str |c_query| {
|
||||
do name.with_c_str |c_name| {
|
||||
ffi::PQprepare(self.conn, c_name, c_query,
|
||||
0, ptr::null())
|
||||
}
|
||||
};
|
||||
PostgresResult {result: raw_res}
|
||||
};
|
||||
|
||||
if res.status() != ffi::PGRES_COMMAND_OK {
|
||||
return Err(res.error());
|
||||
do self.stream.with_mut_ref |s| {
|
||||
s.write_message(&Parse(query_name.clone(), query.to_owned(), ~[]));
|
||||
}
|
||||
|
||||
res = unsafe {
|
||||
let raw_res = do name.with_c_str |c_name| {
|
||||
ffi::PQdescribePrepared(self.conn, c_name)
|
||||
};
|
||||
PostgresResult {result: raw_res}
|
||||
};
|
||||
|
||||
if res.status() != ffi::PGRES_COMMAND_OK {
|
||||
return Err(res.error());
|
||||
do self.stream.with_mut_ref |s| {
|
||||
s.write_message(&Sync);
|
||||
}
|
||||
|
||||
Ok(~PostgresStatement {conn: self, name: name,
|
||||
num_params: res.num_params()})
|
||||
}
|
||||
|
||||
pub fn update(&self, query: &str, params: &[&ToSql])
|
||||
-> Result<uint, ~str> {
|
||||
do self.prepare(query).chain |stmt| {
|
||||
stmt.update(params)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn query(&self, query: &str, params: &[&ToSql])
|
||||
-> Result<~PostgresResult, ~str> {
|
||||
do self.prepare(query).chain |stmt| {
|
||||
stmt.query(params)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn in_transaction<T>(&self,
|
||||
blk: &fn(&PostgresConnection) -> Result<T, ~str>)
|
||||
-> Result<T, ~str> {
|
||||
match self.update("BEGIN", []) {
|
||||
Ok(_) => (),
|
||||
Err(err) => return Err(err)
|
||||
};
|
||||
|
||||
// If the task fails in blk, the transaction will roll back when the
|
||||
// connection closes
|
||||
let ret = blk(self);
|
||||
|
||||
// TODO What to do about errors here?
|
||||
if ret.is_ok() {
|
||||
self.update("COMMIT", []);
|
||||
} else {
|
||||
self.update("ROLLBACK", []);
|
||||
match self.stream.with_mut_ref(|s| { s.read_message() }) {
|
||||
ParseComplete => (),
|
||||
ErrorResponse(ref data) => fail!("Error: %?", data),
|
||||
resp => fail!("Bad response: %?", resp)
|
||||
}
|
||||
|
||||
ret
|
||||
match self.stream.with_mut_ref(|s| { s.read_message() }) {
|
||||
ReadyForQuery(*) => (),
|
||||
resp => fail!("Bad response: %?", resp)
|
||||
}
|
||||
|
||||
PostgresStatement { conn: self, name: query_name }
|
||||
}
|
||||
}
|
||||
|
||||
pub struct PostgresStatement<'self> {
|
||||
priv conn: &'self PostgresConnection,
|
||||
priv name: ~str,
|
||||
priv num_params: uint
|
||||
}
|
||||
|
||||
#[unsafe_destructor]
|
||||
impl<'self> Drop for PostgresStatement<'self> {
|
||||
fn drop(&self) {
|
||||
// We can't do self.conn.update(...) since that will create a statement
|
||||
let query = fmt!("DEALLOCATE %s", self.name);
|
||||
unsafe {
|
||||
do query.with_c_str |c_query| {
|
||||
ffi::PQclear(ffi::PQexec(self.conn.conn, c_query));
|
||||
}
|
||||
}
|
||||
}
|
||||
priv name: ~str
|
||||
}
|
||||
|
||||
impl<'self> PostgresStatement<'self> {
|
||||
fn exec(&self, params: &[&ToSql]) -> Result<~PostgresResult, ~str> {
|
||||
if params.len() != self.num_params {
|
||||
return Err(~"Incorrect number of parameters");
|
||||
}
|
||||
|
||||
let res = unsafe {
|
||||
let raw_res = do self.name.with_c_str |c_name| {
|
||||
let str_params: ~[Option<~str>] = do params.map |param| {
|
||||
param.to_sql()
|
||||
};
|
||||
do with_c_str_array(str_params) |c_params| {
|
||||
ffi::PQexecPrepared(self.conn.conn, c_name,
|
||||
self.num_params as c_int,
|
||||
c_params, ptr::null(), ptr::null(),
|
||||
ffi::TEXT_FORMAT)
|
||||
}
|
||||
};
|
||||
~PostgresResult{result: raw_res}
|
||||
};
|
||||
|
||||
match res.status() {
|
||||
ffi::PGRES_EMPTY_QUERY |
|
||||
ffi::PGRES_COMMAND_OK |
|
||||
ffi::PGRES_TUPLES_OK => Ok(res),
|
||||
_ => Err(res.error())
|
||||
}
|
||||
}
|
||||
|
||||
pub fn update(&self, params: &[&ToSql]) -> Result<uint, ~str> {
|
||||
do self.exec(params).chain |res| {
|
||||
Ok(res.affected_rows())
|
||||
}
|
||||
}
|
||||
|
||||
pub fn query(&self, params: &[&ToSql]) -> Result<~PostgresResult, ~str> {
|
||||
do self.exec(params).chain |res| {
|
||||
Ok(res)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn with_c_str_array<T>(array: &[Option<~str>], blk: &fn(**c_char) -> T) -> T {
|
||||
let mut cstrs: ~[CString] = ~[];
|
||||
let mut c_array: ~[*c_char] = ~[];
|
||||
for s in array.iter() {
|
||||
match *s {
|
||||
None => {
|
||||
c_array.push(ptr::null())
|
||||
}
|
||||
Some(ref s) => {
|
||||
cstrs.push(s.to_c_str());
|
||||
do cstrs.last().with_ref |c_str| {
|
||||
c_array.push(c_str);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
blk(vec::raw::to_ptr(c_array))
|
||||
}
|
||||
|
||||
pub struct PostgresResult {
|
||||
priv result: *ffi::PGresult
|
||||
}
|
||||
|
||||
impl Drop for PostgresResult {
|
||||
fn drop(&self) {
|
||||
unsafe { ffi::PQclear(self.result) }
|
||||
}
|
||||
}
|
||||
|
||||
impl PostgresResult {
|
||||
fn status(&self) -> c_int {
|
||||
unsafe { ffi::PQresultStatus(self.result) }
|
||||
}
|
||||
|
||||
fn error(&self) -> ~str {
|
||||
unsafe { str::raw::from_c_str(ffi::PQresultErrorMessage(self.result)) }
|
||||
}
|
||||
|
||||
fn affected_rows(&self) -> uint {
|
||||
let s = unsafe {
|
||||
str::raw::from_c_str(ffi::PQcmdTuples(self.result))
|
||||
};
|
||||
|
||||
match FromStr::from_str(s) {
|
||||
Some(updates) => updates,
|
||||
None => 0
|
||||
}
|
||||
}
|
||||
|
||||
fn num_params(&self) -> uint {
|
||||
unsafe { ffi::PQnparams(self.result) as uint }
|
||||
}
|
||||
|
||||
fn get_value(&self, row: uint, col: uint) -> Option<~str> {
|
||||
unsafe {
|
||||
match ffi::PQgetisnull(self.result, row as c_int, col as c_int) {
|
||||
0 => {
|
||||
let raw_s = ffi::PQgetvalue(self.result,
|
||||
row as c_int,
|
||||
col as c_int);
|
||||
Some(str::raw::from_c_str(raw_s))
|
||||
}
|
||||
_ => None
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Container for PostgresResult {
|
||||
fn len(&self) -> uint {
|
||||
unsafe { ffi::PQntuples(self.result) as uint }
|
||||
}
|
||||
}
|
||||
|
||||
impl PostgresResult {
|
||||
pub fn iter<'a>(&'a self) -> PostgresResultIterator<'a> {
|
||||
PostgresResultIterator {result: self, next_row: 0}
|
||||
}
|
||||
|
||||
pub fn get<'a>(&'a self, idx: uint) -> PostgresRow<'a> {
|
||||
if idx >= self.len() {
|
||||
fail!("Out of bounds access");
|
||||
}
|
||||
|
||||
self.iter().idx(idx).unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
pub struct PostgresResultIterator<'self> {
|
||||
priv result: &'self PostgresResult,
|
||||
priv next_row: uint
|
||||
}
|
||||
|
||||
impl<'self> Iterator<PostgresRow<'self>> for PostgresResultIterator<'self> {
|
||||
fn next(&mut self) -> Option<PostgresRow<'self>> {
|
||||
if self.result.len() == self.next_row {
|
||||
return None;
|
||||
}
|
||||
|
||||
let row = self.next_row;
|
||||
self.next_row += 1;
|
||||
Some(PostgresRow {result: self.result, row: row})
|
||||
}
|
||||
|
||||
fn size_hint(&self) -> (uint, Option<uint>) {
|
||||
let rem = self.result.len() - self.next_row;
|
||||
(rem, Some(rem))
|
||||
}
|
||||
}
|
||||
|
||||
impl<'self> RandomAccessIterator<PostgresRow<'self>> for
|
||||
PostgresResultIterator<'self> {
|
||||
fn indexable(&self) -> uint {
|
||||
self.result.len()
|
||||
}
|
||||
|
||||
fn idx(&self, idx: uint) -> Option<PostgresRow<'self>> {
|
||||
if idx < self.indexable() {
|
||||
Some(PostgresRow {result: self.result, row: idx})
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct PostgresRow<'self> {
|
||||
priv result: &'self PostgresResult,
|
||||
priv row: uint
|
||||
}
|
||||
|
||||
impl<'self> Container for PostgresRow<'self> {
|
||||
fn len(&self) -> uint {
|
||||
unsafe { ffi::PQnfields(self.result.result) as uint }
|
||||
}
|
||||
}
|
||||
|
||||
impl<'self, T: FromSql> Index<uint, T> for PostgresRow<'self> {
|
||||
fn index(&self, idx: &uint) -> T {
|
||||
self.get(*idx)
|
||||
}
|
||||
}
|
||||
|
||||
impl<'self> PostgresRow<'self> {
|
||||
fn get_value(&self, col: uint) -> Option<~str> {
|
||||
self.result.get_value(self.row, col)
|
||||
}
|
||||
}
|
||||
|
||||
impl<'self> PostgresRow<'self> {
|
||||
pub fn get<T: FromSql>(&self, idx: uint) -> T {
|
||||
if idx >= self.len() {
|
||||
fail!("Out of bounds access");
|
||||
}
|
||||
FromSql::from_sql(self, idx)
|
||||
}
|
||||
}
|
||||
|
||||
pub trait FromSql {
|
||||
fn from_sql(row: &PostgresRow, idx: uint) -> Self;
|
||||
}
|
||||
|
||||
macro_rules! from_opt_impl(
|
||||
($t:ty) => (
|
||||
impl FromSql for $t {
|
||||
fn from_sql(row: &PostgresRow, idx: uint) -> $t {
|
||||
FromSql::from_sql::<Option<$t>>(row, idx).unwrap()
|
||||
}
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
macro_rules! from_str_opt_impl(
|
||||
($t:ty) => (
|
||||
impl FromSql for Option<$t> {
|
||||
fn from_sql(row: &PostgresRow, idx: uint) -> Option<$t> {
|
||||
do row.get_value(idx).chain |s| {
|
||||
Some(FromStr::from_str(s).unwrap())
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
from_opt_impl!(int)
|
||||
from_str_opt_impl!(int)
|
||||
from_opt_impl!(i8)
|
||||
from_str_opt_impl!(i8)
|
||||
from_opt_impl!(i16)
|
||||
from_str_opt_impl!(i16)
|
||||
from_opt_impl!(i32)
|
||||
from_str_opt_impl!(i32)
|
||||
from_opt_impl!(i64)
|
||||
from_str_opt_impl!(i64)
|
||||
from_opt_impl!(uint)
|
||||
from_str_opt_impl!(uint)
|
||||
from_opt_impl!(u8)
|
||||
from_str_opt_impl!(u8)
|
||||
from_opt_impl!(u16)
|
||||
from_str_opt_impl!(u16)
|
||||
from_opt_impl!(u32)
|
||||
from_str_opt_impl!(u32)
|
||||
from_opt_impl!(u64)
|
||||
from_str_opt_impl!(u64)
|
||||
from_opt_impl!(float)
|
||||
from_str_opt_impl!(float)
|
||||
from_opt_impl!(f32)
|
||||
from_str_opt_impl!(f32)
|
||||
from_opt_impl!(f64)
|
||||
from_str_opt_impl!(f64)
|
||||
|
||||
impl FromSql for Option<~str> {
|
||||
fn from_sql(row: &PostgresRow, idx: uint) -> Option<~str> {
|
||||
row.get_value(idx)
|
||||
}
|
||||
}
|
||||
from_opt_impl!(~str)
|
||||
|
||||
pub trait ToSql {
|
||||
fn to_sql(&self) -> Option<~str>;
|
||||
}
|
||||
|
||||
macro_rules! to_str_impl(
|
||||
($t:ty) => (
|
||||
impl ToSql for $t {
|
||||
fn to_sql(&self) -> Option<~str> {
|
||||
Some(self.to_str())
|
||||
}
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
macro_rules! to_str_opt_impl(
|
||||
($t:ty) => (
|
||||
impl ToSql for Option<$t> {
|
||||
fn to_sql(&self) -> Option<~str> {
|
||||
match *self {
|
||||
Some(ref val) => Some(val.to_sql().unwrap()),
|
||||
None => None
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
to_str_impl!(int)
|
||||
to_str_opt_impl!(int)
|
||||
to_str_impl!(i8)
|
||||
to_str_opt_impl!(i8)
|
||||
to_str_impl!(i16)
|
||||
to_str_opt_impl!(i16)
|
||||
to_str_impl!(i32)
|
||||
to_str_opt_impl!(i32)
|
||||
to_str_impl!(i64)
|
||||
to_str_opt_impl!(i64)
|
||||
to_str_impl!(uint)
|
||||
to_str_opt_impl!(uint)
|
||||
to_str_impl!(u8)
|
||||
to_str_opt_impl!(u8)
|
||||
to_str_impl!(u16)
|
||||
to_str_opt_impl!(u16)
|
||||
to_str_impl!(u32)
|
||||
to_str_opt_impl!(u32)
|
||||
to_str_impl!(u64)
|
||||
to_str_opt_impl!(u64)
|
||||
to_str_impl!(float)
|
||||
to_str_opt_impl!(float)
|
||||
to_str_impl!(f32)
|
||||
to_str_opt_impl!(f32)
|
||||
to_str_impl!(f64)
|
||||
to_str_opt_impl!(f64)
|
||||
|
||||
impl<'self> ToSql for &'self str {
|
||||
fn to_sql(&self) -> Option<~str> {
|
||||
Some(self.to_str())
|
||||
}
|
||||
}
|
||||
|
||||
impl ToSql for ~str {
|
||||
fn to_sql(&self) -> Option<~str> {
|
||||
Some(self.clone())
|
||||
}
|
||||
}
|
||||
|
||||
impl ToSql for Option<~str> {
|
||||
fn to_sql(&self) -> Option<~str> {
|
||||
self.clone()
|
||||
}
|
||||
}
|
||||
|
||||
160
src/message.rs
Normal file
160
src/message.rs
Normal file
@@ -0,0 +1,160 @@
|
||||
use std::str;
|
||||
use std::rt::io::{Decorator, Reader, Writer};
|
||||
use std::rt::io::extensions::{ReaderUtil, ReaderByteConversions,
|
||||
WriterByteConversions};
|
||||
use std::rt::io::mem::{MemWriter, MemReader};
|
||||
use std::hashmap::HashMap;
|
||||
use std::sys;
|
||||
|
||||
pub static PROTOCOL_VERSION: i32 = 0x0003_0000;
|
||||
|
||||
pub enum BackendMessage {
|
||||
AuthenticationOk,
|
||||
BackendKeyData(i32, i32),
|
||||
ErrorResponse(HashMap<u8, ~str>),
|
||||
ParameterStatus(~str, ~str),
|
||||
ParseComplete,
|
||||
ReadyForQuery(u8)
|
||||
}
|
||||
|
||||
pub enum FrontendMessage {
|
||||
/// name, query, parameter types
|
||||
Parse(~str, ~str, ~[i32]),
|
||||
Query(~str),
|
||||
StartupMessage(HashMap<~str, ~str>),
|
||||
Sync,
|
||||
Terminate
|
||||
}
|
||||
|
||||
trait WriteString {
|
||||
fn write_string(&mut self, s: &str);
|
||||
}
|
||||
|
||||
impl<W: Writer> WriteString for W {
|
||||
fn write_string(&mut self, s: &str) {
|
||||
self.write(s.as_bytes());
|
||||
self.write_u8_(0);
|
||||
}
|
||||
}
|
||||
|
||||
pub trait WriteMessage {
|
||||
fn write_message(&mut self, &FrontendMessage);
|
||||
}
|
||||
|
||||
impl<W: Writer> WriteMessage for W {
|
||||
fn write_message(&mut self, message: &FrontendMessage) {
|
||||
debug!("Writing message %?", message);
|
||||
let mut buf = MemWriter::new();
|
||||
let mut ident = None;
|
||||
|
||||
match *message {
|
||||
Parse(ref name, ref query, ref param_types) => {
|
||||
ident = Some('P');
|
||||
buf.write_string(*name);
|
||||
buf.write_string(*query);
|
||||
buf.write_be_i16_(param_types.len() as i16);
|
||||
for ty in param_types.iter() {
|
||||
buf.write_be_i32_(*ty);
|
||||
}
|
||||
}
|
||||
Query(ref query) => {
|
||||
ident = Some('Q');
|
||||
buf.write_string(*query);
|
||||
}
|
||||
StartupMessage(ref params) => {
|
||||
buf.write_be_i32_(PROTOCOL_VERSION);
|
||||
for (k, v) in params.iter() {
|
||||
buf.write_string(*k);
|
||||
buf.write_string(*v);
|
||||
}
|
||||
buf.write_u8_(0);
|
||||
}
|
||||
Sync => {
|
||||
ident = Some('S');
|
||||
}
|
||||
Terminate => {
|
||||
ident = Some('X');
|
||||
}
|
||||
}
|
||||
|
||||
match ident {
|
||||
Some(ident) => self.write_u8_(ident as u8),
|
||||
None => ()
|
||||
}
|
||||
|
||||
// add size of length value
|
||||
self.write_be_i32_((buf.inner_ref().len() + sys::size_of::<i32>())
|
||||
as i32);
|
||||
self.write(buf.inner());
|
||||
}
|
||||
}
|
||||
|
||||
trait ReadString {
|
||||
fn read_string(&mut self) -> ~str;
|
||||
}
|
||||
|
||||
impl<R: Reader> ReadString for R {
|
||||
fn read_string(&mut self) -> ~str {
|
||||
let mut buf = ~[];
|
||||
loop {
|
||||
let byte = self.read_u8_();
|
||||
if byte == 0 {
|
||||
break;
|
||||
}
|
||||
|
||||
buf.push(byte);
|
||||
}
|
||||
|
||||
str::from_bytes_owned(buf)
|
||||
}
|
||||
}
|
||||
|
||||
pub trait ReadMessage {
|
||||
fn read_message(&mut self) -> BackendMessage;
|
||||
}
|
||||
|
||||
impl<R: Reader> ReadMessage for R {
|
||||
fn read_message(&mut self) -> BackendMessage {
|
||||
debug!("Reading message");
|
||||
|
||||
let ident = self.read_u8_();
|
||||
debug!("Ident %?", ident);
|
||||
// subtract size of length value
|
||||
let len = self.read_be_i32_() as uint - sys::size_of::<i32>();
|
||||
let mut buf = MemReader::new(self.read_bytes(len));
|
||||
|
||||
let ret = match ident as char {
|
||||
'1' => ParseComplete,
|
||||
'E' => read_error_message(&mut buf),
|
||||
'K' => BackendKeyData(buf.read_be_i32_(), buf.read_be_i32_()),
|
||||
'R' => read_auth_message(&mut buf),
|
||||
'S' => ParameterStatus(buf.read_string(), buf.read_string()),
|
||||
'Z' => ReadyForQuery(buf.read_u8_()),
|
||||
ident => fail!("Unknown message identifier `%c`", ident)
|
||||
};
|
||||
assert!(buf.eof());
|
||||
debug!("Read message %?", ret);
|
||||
ret
|
||||
}
|
||||
}
|
||||
|
||||
fn read_error_message(buf: &mut MemReader) -> BackendMessage {
|
||||
let mut fields = HashMap::new();
|
||||
loop {
|
||||
let ty = buf.read_u8_();
|
||||
if ty == 0 {
|
||||
break;
|
||||
}
|
||||
|
||||
fields.insert(ty, buf.read_string());
|
||||
}
|
||||
|
||||
ErrorResponse(fields)
|
||||
}
|
||||
|
||||
fn read_auth_message(buf: &mut MemReader) -> BackendMessage {
|
||||
match buf.read_be_i32_() {
|
||||
0 => AuthenticationOk,
|
||||
val => fail!("Unknown Authentication identifier `%?`", val)
|
||||
}
|
||||
}
|
||||
114
src/test.rs
114
src/test.rs
@@ -1,115 +1,11 @@
|
||||
extern mod postgres;
|
||||
|
||||
use postgres::{PostgresConnection, ToSql};
|
||||
|
||||
macro_rules! params(
|
||||
($($e:expr),+) => (
|
||||
[$(
|
||||
$e as &ToSql
|
||||
),+]
|
||||
)
|
||||
)
|
||||
|
||||
macro_rules! chk(
|
||||
($e:expr) => (
|
||||
match $e {
|
||||
Ok(ok) => ok,
|
||||
Err(err) => fail!(err)
|
||||
}
|
||||
)
|
||||
)
|
||||
use postgres::PostgresConnection;
|
||||
|
||||
#[test]
|
||||
fn test_basic() {
|
||||
let conn = chk!(PostgresConnection::new("postgres://postgres@localhost"));
|
||||
fn test_connect() {
|
||||
let conn = PostgresConnection::connect("postgres://127.0.0.1:54322",
|
||||
"postgres");
|
||||
|
||||
do conn.in_transaction |conn| {
|
||||
chk!(conn.update("CREATE TABLE basic (id INT PRIMARY KEY)", []));
|
||||
chk!(conn.update("INSERT INTO basic (id) VALUES (101)", []));
|
||||
|
||||
let res = chk!(conn.query("SELECT id from basic WHERE id = 101", []));
|
||||
assert_eq!(1, res.len());
|
||||
assert_eq!(1, res.get(0).len());
|
||||
assert_eq!(1, res.get(0).len());
|
||||
assert_eq!(Some(101), res.get(0)[0]);
|
||||
|
||||
Err::<(), ~str>(~"")
|
||||
};
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_multiple_stmts() {
|
||||
let conn = chk!(PostgresConnection::new("postgres://postgres@localhost"));
|
||||
|
||||
do conn.in_transaction |conn| {
|
||||
chk!(conn.update("CREATE TABLE foo (id INT PRIMARY KEY)", []));
|
||||
let stmt1 = chk!(conn.prepare("INSERT INTO foo (id) VALUES (101)"));
|
||||
let stmt2 = chk!(conn.prepare("INSERT INTO foo (id) VALUES (102)"));
|
||||
|
||||
chk!(stmt1.update([]));
|
||||
chk!(stmt2.update([]));
|
||||
|
||||
Err::<(), ~str>(~"")
|
||||
};
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_params() {
|
||||
let conn = chk!(PostgresConnection::new("postgres://postgres@localhost"));
|
||||
|
||||
do conn.in_transaction |conn| {
|
||||
chk!(conn.update("CREATE TABLE basic (id INT PRIMARY KEY)", []));
|
||||
chk!(conn.update("INSERT INTO basic (id) VALUES ($1)",
|
||||
params!(&101)));
|
||||
|
||||
let res = chk!(conn.query("SELECT id from basic WHERE id = $1",
|
||||
params!(&101)));
|
||||
assert_eq!(Some(101), res.get(0)[0]);
|
||||
|
||||
Err::<(), ~str>(~"")
|
||||
};
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_null() {
|
||||
let conn = chk!(PostgresConnection::new("postgres://postgres@localhost"));
|
||||
|
||||
do conn.in_transaction |conn| {
|
||||
chk!(conn.update("CREATE TABLE basic (id INT PRIMARY KEY, foo INT)",
|
||||
[]));
|
||||
chk!(conn.update("INSERT INTO basic (id, foo) VALUES ($1, $2), ($3, $4)",
|
||||
params!(&101, &None::<int>, &102, &0)));
|
||||
|
||||
let res = chk!(conn.query("SELECT foo from basic ORDER BY id", []));
|
||||
assert_eq!(None, res.get(0).get::<Option<int>>(0));
|
||||
assert_eq!(Some(0), res.get(1).get::<Option<int>>(0));
|
||||
|
||||
Err::<(), ~str>(~"")
|
||||
};
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_types() {
|
||||
let conn = chk!(PostgresConnection::new("postgres://postgres@localhost"));
|
||||
|
||||
do conn.in_transaction |conn| {
|
||||
chk!(conn.update(
|
||||
"CREATE TABLE foo (
|
||||
id INT PRIMARY KEY,
|
||||
str VARCHAR,
|
||||
float REAL
|
||||
)", []));
|
||||
chk!(conn.update("INSERT INTO foo (id, str, float)
|
||||
VALUES ($1, $2, $3), ($4, $5, $6)",
|
||||
params!(&101, & &"foobar", &10.5,
|
||||
&102, &None::<~str>, &None::<float>)));
|
||||
|
||||
let res = chk!(conn.query("SELECT str, float from foo ORDER BY id", []));
|
||||
assert_eq!(~"foobar", res.get(0).get::<~str>(0));
|
||||
assert_eq!(10.5, res.get(0).get::<float>(1));
|
||||
assert_eq!(None::<~str>, res.get(1).get::<Option<~str>>(0));
|
||||
assert_eq!(None::<float>, res.get(1).get::<Option<float>>(1));
|
||||
|
||||
Err::<(), ~str>(~"")
|
||||
};
|
||||
conn.prepare("CREATE TABLE foo (id BIGINT PRIMARY KEY)");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user