dnim/rust-postgres
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

try! -> ?

Browse Source
This commit is contained in:
Steven Fackler 2017-02-14 22:46:19 -08:00
parent 73e4ec1c0d
commit 4c91a68dcc
31 changed files with 332 additions and 336 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
README.md
View File

@ -63,8 +63,8 @@ fn main() {
### Connecting
Connect to a Postgres server using the standard URI format:
```rust
let conn = try!(Connection::connect("postgres://user:pass@host:port/database?arg1=val1&arg2=val2",
TlsMode::None));
let conn = Connection::connect("postgres://user:pass@host:port/database?arg1=val1&arg2=val2",
TlsMode::None)?;
```
`pass` may be omitted if not needed. `port` defaults to `5432` and `database`
defaults to the value of `user` if not specified. The driver supports `trust`,
@ -75,7 +75,7 @@ be set to the absolute path to the directory containing the socket file. Since
`/` is a reserved character in URLs, the path should be URL encoded. If Postgres
stored its socket files in `/run/postgres`, the connection would then look like:
```rust
let conn = try!(Connection::connect("postgres://postgres@%2Frun%2Fpostgres", TlsMode::None));
let conn = Connection::connect("postgres://postgres@%2Frun%2Fpostgres", TlsMode::None)?;
```
Paths which contain non-UTF8 characters can be handled in a different manner;
see the documentation for details.
@ -89,7 +89,7 @@ that query parameters are 1-indexed rather than the more common 0-indexing.
`execute` returns the number of rows affected by the query (or 0 if not
applicable):
```rust
let updates = try!(conn.execute("UPDATE foo SET bar = $1 WHERE baz = $2", &[&1i32, &"biz"]));
let updates = conn.execute("UPDATE foo SET bar = $1 WHERE baz = $2", &[&1i32, &"biz"])?;
println!("{} rows were updated", updates);
```
@ -98,7 +98,7 @@ The fields in a row can be accessed either by their indices or their column
names, though access by index is more efficient. Unlike statement parameters,
result columns are zero-indexed.
```rust
for row in &try!(conn.query("SELECT bar, baz FROM foo WHERE buz = $1", &[&1i32])) {
for row in &conn.query("SELECT bar, baz FROM foo WHERE buz = $1", &[&1i32])? {
let bar: i32 = row.get(0);
let baz: String = row.get("baz");
println!("bar: {}, baz: {}", bar, baz);
@ -110,9 +110,9 @@ If the same statement will be executed repeatedly (possibly with different
parameters), explicitly preparing it can improve performance:
```rust
let stmt = try!(conn.prepare("UPDATE foo SET bar = $1 WHERE baz = $2"));
let stmt = conn.prepare("UPDATE foo SET bar = $1 WHERE baz = $2")?;
for (bar, baz) in updates {
try!(stmt.execute(&[bar, baz]));
stmt.execute(&[bar, baz])?;
}
```
@ -122,13 +122,13 @@ The `transaction` method will start a new transaction. It returns a
`Connection` as well as methods to control the result of the
transaction:
```rust
let trans = try!(conn.transaction());
let trans = conn.transaction()?;
try!(trans.execute(...));
let stmt = try!(trans.prepare(...));
trans.execute(...)?;
let stmt = trans.prepare(...)?;
// ...
try!(trans.commit());
trans.commit()?;
```
The transaction will be active until the `Transaction` object falls out of
scope. A transaction will roll back by default. Nested transactions are

2
codegen/src/type_gen.rs
View File

@ -159,7 +159,7 @@ r#"impl fmt::Display for Type {{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {{
match self.schema() {{
"public" | "pg_catalog" => {{}}
schema => try!(write!(fmt, "{{}}.", schema)),
schema => write!(fmt, "{{}}.", schema)?,
}}
fmt.write_str(self.name())
}}

36
postgres-shared/src/error/mod.rs
View File

@ -159,7 +159,7 @@ impl DbError {
let mut line = None;
let mut routine = None;
while let Some(field) = try!(fields.next()) {
while let Some(field) = fields.next()? {
match field.type_() {
b'S' => severity = Some(field.value().to_owned()),
b'C' => code = Some(SqlState::from_code(field.value())),
@ -167,16 +167,16 @@ impl DbError {
b'D' => detail = Some(field.value().to_owned()),
b'H' => hint = Some(field.value().to_owned()),
b'P' => {
normal_position = Some(try!(field.value().parse::<u32>().map_err(|_| {
normal_position = Some(field.value().parse::<u32>().map_err(|_| {
io::Error::new(io::ErrorKind::InvalidInput,
"`P` field did not contain an integer")
})));
})?);
}
b'p' => {
internal_position = Some(try!(field.value().parse::<u32>().map_err(|_| {
internal_position = Some(field.value().parse::<u32>().map_err(|_| {
io::Error::new(io::ErrorKind::InvalidInput,
"`p` field did not contain an integer")
})));
})?);
}
b'q' => internal_query = Some(field.value().to_owned()),
b'W' => where_ = Some(field.value().to_owned()),
@ -187,31 +187,31 @@ impl DbError {
b'n' => constraint = Some(field.value().to_owned()),
b'F' => file = Some(field.value().to_owned()),
b'L' => {
line = Some(try!(field.value().parse::<u32>().map_err(|_| {
line = Some(field.value().parse::<u32>().map_err(|_| {
io::Error::new(io::ErrorKind::InvalidInput,
"`L` field did not contain an integer")
})));
})?);
}
b'R' => routine = Some(field.value().to_owned()),
b'V' => {
parsed_severity = Some(try!(Severity::from_str(field.value()).ok_or_else(|| {
parsed_severity = Some(Severity::from_str(field.value()).ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput,
"`V` field contained an invalid value")
})));
})?);
}
_ => {},
}
}
Ok(DbError {
severity: try!(severity.ok_or_else(|| {
severity: severity.ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput, "`S` field missing")
})),
})?,
parsed_severity: parsed_severity,
code: try!(code.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput,
"`C` field missing"))),
message: try!(message.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput,
"`M` field missing"))),
code: code.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput,
"`C` field missing"))?,
message: message.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput,
"`M` field missing"))?,
detail: detail,
hint: hint,
position: match normal_position {
@ -221,10 +221,10 @@ impl DbError {
Some(position) => {
Some(ErrorPosition::Internal {
position: position,
query: try!(internal_query.ok_or_else(|| {
query: internal_query.ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidInput,
"`q` field missing but `p` field present")
})),
})?,
})
}
None => None,
@ -310,7 +310,7 @@ pub enum ConnectError {
impl fmt::Display for ConnectError {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
try!(fmt.write_str(error::Error::description(self)));
fmt.write_str(error::Error::description(self))?;
match *self {
ConnectError::ConnectParams(ref msg) => write!(fmt, ": {}", msg),
ConnectError::Db(ref err) => write!(fmt, ": {}", err),

2
postgres-shared/src/params/mod.rs
View File

@ -78,7 +78,7 @@ impl IntoConnectParams for Url {
fn into_connect_params(self) -> Result<ConnectParams, Box<Error + Sync + Send>> {
let Url { host, port, user, path: url::Path { mut path, query: options, .. }, .. } = self;
let maybe_path = try!(url::decode_component(&host));
let maybe_path = url::decode_component(&host)?;
let target = if maybe_path.starts_with('/') {
ConnectTarget::Unix(PathBuf::from(maybe_path))
} else {

26
postgres-shared/src/params/url.rs
View File

@ -51,17 +51,17 @@ impl Url {
pub fn parse(rawurl: &str) -> DecodeResult<Url> {
// scheme
let (scheme, rest) = try!(get_scheme(rawurl));
let (scheme, rest) = get_scheme(rawurl)?;
// authority
let (userinfo, host, port, rest) = try!(get_authority(rest));
let (userinfo, host, port, rest) = get_authority(rest)?;
// path
let has_authority = !host.is_empty();
let (path, rest) = try!(get_path(rest, has_authority));
let (path, rest) = get_path(rest, has_authority)?;
// query and fragment
let (query, fragment) = try!(get_query_fragment(rest));
let (query, fragment) = get_query_fragment(rest)?;
let url = Url::new(scheme.to_owned(),
userinfo,
@ -84,10 +84,10 @@ impl Path {
}
pub fn parse(rawpath: &str) -> DecodeResult<Path> {
let (path, rest) = try!(get_path(rawpath, false));
let (path, rest) = get_path(rawpath, false)?;
// query and fragment
let (query, fragment) = try!(get_query_fragment(&rest));
let (query, fragment) = get_query_fragment(&rest)?;
Ok(Path {
path: path,
@ -177,7 +177,7 @@ fn query_from_str(rawquery: &str) -> DecodeResult<Query> {
if !rawquery.is_empty() {
for p in rawquery.split('&') {
let (k, v) = split_char_first(p, '=');
query.push((try!(decode_component(k)), try!(decode_component(v))));
query.push((decode_component(k)?, decode_component(v)?));
}
}
@ -316,13 +316,13 @@ fn get_authority(rawurl: &str) -> DecodeResult<(Option<UserInfo>, &str, Option<u
colon_count = 0; // reset count
match st {
State::Start => {
let user = try!(decode_component(&rawurl[begin..i]));
let user = decode_component(&rawurl[begin..i])?;
userinfo = Some(UserInfo::new(user, None));
st = State::InHost;
}
State::PassHostPort => {
let user = try!(decode_component(&rawurl[begin..pos]));
let pass = try!(decode_component(&rawurl[pos + 1..i]));
let user = decode_component(&rawurl[begin..pos])?;
let pass = decode_component(&rawurl[pos + 1..i])?;
userinfo = Some(UserInfo::new(user, Some(pass)));
st = State::InHost;
}
@ -392,7 +392,7 @@ fn get_path(rawurl: &str, is_authority: bool) -> DecodeResult<(String, &str)> {
if is_authority && end != 0 && !rawurl.starts_with('/') {
Err("Non-empty path must begin with '/' in presence of authority.".to_owned())
} else {
Ok((try!(decode_component(&rawurl[0..end])), &rawurl[end..len]))
Ok((decode_component(&rawurl[0..end])?, &rawurl[end..len]))
}
}
@ -403,11 +403,11 @@ fn get_query_fragment(rawurl: &str) -> DecodeResult<(Query, Option<String>)> {
// Parse the fragment if available
let fragment = match raw_fragment {
"" => None,
raw => Some(try!(decode_component(raw))),
raw => Some(decode_component(raw)?),
};
match before_fragment.chars().next() {
Some('?') => Ok((try!(query_from_str(&before_fragment[1..])), fragment)),
Some('?') => Ok((query_from_str(&before_fragment[1..])?, fragment)),
None => Ok((vec![], fragment)),
_ => Err(format!("Query didn't start with '?': '{}..'", before_fragment)),
}

2
postgres-shared/src/rows.rs
View File

@ -60,7 +60,7 @@ impl<'a> FromFallibleIterator<Option<&'a [u8]>> for RowData {
indices: Vec::with_capacity(it.size_hint().0),
};
while let Some(cell) = try!(it.next()) {
while let Some(cell) = it.next()? {
let index = match cell {
Some(cell) => {
let base = row.buf.len();

4
postgres-shared/src/types/bit_vec.rs
View File

@ -8,7 +8,7 @@ use types::{FromSql, ToSql, IsNull, Type};
impl FromSql for BitVec {
fn from_sql(_: &Type, raw: &[u8]) -> Result<BitVec, Box<Error + Sync + Send>> {
let varbit = try!(types::varbit_from_sql(raw));
let varbit = types::varbit_from_sql(raw)?;
let mut bitvec = BitVec::from_bytes(varbit.bytes());
while bitvec.len() > varbit.len() {
bitvec.pop();
@ -25,7 +25,7 @@ impl ToSql for BitVec {
_: &Type,
mut out: &mut Vec<u8>)
-> Result<IsNull, Box<Error + Sync + Send>> {
try!(types::varbit_to_sql(self.len(), self.to_bytes().into_iter(), out));
types::varbit_to_sql(self.len(), self.to_bytes().into_iter(), out)?;
Ok(IsNull::No)
}

12
postgres-shared/src/types/chrono.rs
View File

@ -15,7 +15,7 @@ impl FromSql for NaiveDateTime {
fn from_sql(_: &Type,
raw: &[u8])
-> Result<NaiveDateTime, Box<Error + Sync + Send>> {
let t = try!(types::timestamp_from_sql(raw));
let t = types::timestamp_from_sql(raw)?;
Ok(base() + Duration::microseconds(t))
}
@ -43,7 +43,7 @@ impl FromSql for DateTime<UTC> {
fn from_sql(type_: &Type,
raw: &[u8])
-> Result<DateTime<UTC>, Box<Error + Sync + Send>> {
let naive = try!(NaiveDateTime::from_sql(type_, raw));
let naive = NaiveDateTime::from_sql(type_, raw)?;
Ok(DateTime::from_utc(naive, UTC))
}
@ -66,7 +66,7 @@ impl FromSql for DateTime<Local> {
fn from_sql(type_: &Type,
raw: &[u8])
-> Result<DateTime<Local>, Box<Error + Sync + Send>> {
let utc = try!(DateTime::<UTC>::from_sql(type_, raw));
let utc = DateTime::<UTC>::from_sql(type_, raw)?;
Ok(utc.with_timezone(&Local))
}
@ -89,7 +89,7 @@ impl FromSql for DateTime<FixedOffset> {
fn from_sql(type_: &Type,
raw: &[u8])
-> Result<DateTime<FixedOffset>, Box<Error + Sync + Send>> {
let utc = try!(DateTime::<UTC>::from_sql(type_, raw));
let utc = DateTime::<UTC>::from_sql(type_, raw)?;
Ok(utc.with_timezone(&FixedOffset::east(0)))
}
@ -112,7 +112,7 @@ impl FromSql for NaiveDate {
fn from_sql(_: &Type,
raw: &[u8])
-> Result<NaiveDate, Box<Error + Sync + Send>> {
let jd = try!(types::date_from_sql(raw));
let jd = types::date_from_sql(raw)?;
Ok(base().date() + Duration::days(jd as i64))
}
@ -141,7 +141,7 @@ impl FromSql for NaiveTime {
fn from_sql(_: &Type,
raw: &[u8])
-> Result<NaiveTime, Box<Error + Sync + Send>> {
let usec = try!(types::time_from_sql(raw));
let usec = types::time_from_sql(raw)?;
Ok(NaiveTime::from_hms(0, 0, 0) + Duration::microseconds(usec))
}

2
postgres-shared/src/types/eui48.rs
View File

@ -10,7 +10,7 @@ impl FromSql for MacAddress {
fn from_sql(_: &Type,
raw: &[u8])
-> Result<MacAddress, Box<Error + Sync + Send>> {
let bytes = try!(types::macaddr_from_sql(raw));
let bytes = types::macaddr_from_sql(raw)?;
Ok(MacAddress::new(bytes))
}

17
postgres-shared/src/types/mod.rs
View File

@ -343,8 +343,8 @@ impl<T: FromSql> FromSql for Vec<T> {
_ => panic!("expected array type"),
};
let array = try!(types::array_from_sql(raw));
if try!(array.dimensions().count()) > 1 {
let array = types::array_from_sql(raw)?;
if array.dimensions().count()? > 1 {
return Err("array contains too many dimensions".into());
}
@ -412,7 +412,7 @@ impl FromSql for HashMap<String, Option<String>> {
fn from_sql(_: &Type,
raw: &[u8])
-> Result<HashMap<String, Option<String>>, Box<Error + Sync + Send>> {
try!(types::hstore_from_sql(raw))
types::hstore_from_sql(raw)?
.map(|(k, v)| (k.to_owned(), v.map(str::to_owned)))
.collect()
}
@ -564,21 +564,21 @@ impl<'a, T: ToSql> ToSql for &'a [T] {
};
let dimension = ArrayDimension {
len: try!(downcast(self.len())),
len: downcast(self.len())?,
lower_bound: 1,
};
try!(types::array_to_sql(Some(dimension),
types::array_to_sql(Some(dimension),
true,
member_type.oid(),
self.iter(),
|e, w| {
match try!(e.to_sql(member_type, w)) {
match e.to_sql(member_type, w)? {
IsNull::No => Ok(postgres_protocol::IsNull::No),
IsNull::Yes => Ok(postgres_protocol::IsNull::Yes),
}
},
w));
w)?;
Ok(IsNull::No)
}
@ -703,8 +703,7 @@ impl ToSql for HashMap<String, Option<String>> {
_: &Type,
w: &mut Vec<u8>)
-> Result<IsNull, Box<Error + Sync + Send>> {
try!(types::hstore_to_sql(self.iter().map(|(k, v)| (&**k, v.as_ref().map(|v| &**v))),
w));
types::hstore_to_sql(self.iter().map(|(k, v)| (&**k, v.as_ref().map(|v| &**v))), w)?;
Ok(IsNull::No)
}

4
postgres-shared/src/types/rustc_serialize.rs
View File

@ -12,7 +12,7 @@ impl FromSql for json::Json {
-> Result<json::Json, Box<Error + Sync + Send>> {
if let Type::Jsonb = *ty {
let mut b = [0; 1];
try!(raw.read_exact(&mut b));
raw.read_exact(&mut b)?;
// We only support version 1 of the jsonb binary format
if b[0] != 1 {
return Err("unsupported JSONB encoding version".into());
@ -32,7 +32,7 @@ impl ToSql for json::Json {
if let Type::Jsonb = *ty {
out.push(1);
}
try!(write!(out, "{}", self));
write!(out, "{}", self)?;
Ok(IsNull::No)
}

4
postgres-shared/src/types/serde_json.rs
View File

@ -12,7 +12,7 @@ impl FromSql for Value {
-> Result<Value, Box<Error + Sync + Send>> {
if let Type::Jsonb = *ty {
let mut b = [0; 1];
try!(raw.read_exact(&mut b));
raw.read_exact(&mut b)?;
// We only support version 1 of the jsonb binary format
if b[0] != 1 {
return Err("unsupported JSONB encoding version".into());
@ -32,7 +32,7 @@ impl ToSql for Value {
if let Type::Jsonb = *ty {
out.push(1);
}
try!(write!(out, "{}", self));
write!(out, "{}", self)?;
Ok(IsNull::No)
}

4
postgres-shared/src/types/special.rs
View File

@ -19,7 +19,7 @@ impl<T: FromSql> FromSql for Date<T> {
fn from_sql(ty: &Type,
raw: &[u8])
-> Result<Self, Box<Error + Sync + Send>> {
match try!(types::date_from_sql(raw)) {
match types::date_from_sql(raw)? {
i32::MAX => Ok(Date::PosInfinity),
i32::MIN => Ok(Date::NegInfinity),
_ => T::from_sql(ty, raw).map(Date::Value),
@ -68,7 +68,7 @@ impl<T: FromSql> FromSql for Timestamp<T> {
fn from_sql(ty: &Type,
raw: &[u8])
-> Result<Self, Box<Error + Sync + Send>> {
match try!(types::timestamp_from_sql(raw)) {
match types::timestamp_from_sql(raw)? {
i64::MAX => Ok(Timestamp::PosInfinity),
i64::MIN => Ok(Timestamp::NegInfinity),
_ => T::from_sql(ty, raw).map(Timestamp::Value),

2
postgres-shared/src/types/time.rs
View File

@ -16,7 +16,7 @@ impl FromSql for Timespec {
fn from_sql(_: &Type,
raw: &[u8])
-> Result<Timespec, Box<Error + Sync + Send>> {
let t = try!(types::timestamp_from_sql(raw));
let t = types::timestamp_from_sql(raw)?;
let mut sec = t / USEC_PER_SEC + TIME_SEC_CONVERSION;
let mut usec = t % USEC_PER_SEC;

2
postgres-shared/src/types/type_gen.rs
View File

@ -332,7 +332,7 @@ impl fmt::Display for Type {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
match self.schema() {
"public" | "pg_catalog" => {}
schema => try!(write!(fmt, "{}.", schema)),
schema => write!(fmt, "{}.", schema)?,
}
fmt.write_str(self.name())
}

2
postgres-shared/src/types/uuid.rs
View File

@ -8,7 +8,7 @@ use types::{FromSql, ToSql, Type, IsNull};
impl FromSql for Uuid {
fn from_sql(_: &Type, raw: &[u8]) -> Result<Uuid, Box<Error + Sync + Send>> {
let bytes = try!(types::uuid_from_sql(raw));
let bytes = types::uuid_from_sql(raw)?;
Ok(Uuid::from_bytes(&bytes).unwrap())
}

2
postgres/src/error.rs
View File

@ -19,7 +19,7 @@ pub enum Error {
impl fmt::Display for Error {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
try!(fmt.write_str(error::Error::description(self)));
fmt.write_str(error::Error::description(self))?;
match *self {
Error::Db(ref err) => write!(fmt, ": {}", err),
Error::Io(ref err) => write!(fmt, ": {}", err),

276
postgres/src/lib.rs
View File

@ -185,13 +185,13 @@ pub fn cancel_query<T>(params: T,
-> result::Result<(), ConnectError>
where T: IntoConnectParams
{
let params = try!(params.into_connect_params().map_err(ConnectError::ConnectParams));
let mut socket = try!(priv_io::initialize_stream(&params, tls));
let params = params.into_connect_params().map_err(ConnectError::ConnectParams)?;
let mut socket = priv_io::initialize_stream(&params, tls)?;
let mut buf = vec![];
frontend::cancel_request(data.process_id, data.secret_key, &mut buf);
try!(socket.write_all(&buf));
try!(socket.flush());
socket.write_all(&buf)?;
socket.flush()?;
Ok(())
}
@ -252,8 +252,8 @@ impl InnerConnection {
fn connect<T>(params: T, tls: TlsMode) -> result::Result<InnerConnection, ConnectError>
where T: IntoConnectParams
{
let params = try!(params.into_connect_params().map_err(ConnectError::ConnectParams));
let stream = try!(priv_io::initialize_stream(&params, tls));
let params = params.into_connect_params().map_err(ConnectError::ConnectParams)?;
let stream = priv_io::initialize_stream(&params, tls)?;
let ConnectParams { user, database, mut options, .. } = params;
@ -296,13 +296,13 @@ impl InnerConnection {
}
let options = options.iter().map(|&(ref a, ref b)| (&**a, &**b));
try!(conn.stream.write_message(|buf| frontend::startup_message(options, buf)));
try!(conn.stream.flush());
conn.stream.write_message(|buf| frontend::startup_message(options, buf))?;
conn.stream.flush()?;
try!(conn.handle_auth(user));
conn.handle_auth(user)?;
loop {
match try!(conn.read_message()) {
match conn.read_message()? {
backend::Message::BackendKeyData(body) => {
conn.cancel_data.process_id = body.process_id();
conn.cancel_data.secret_key = body.secret_key();
@ -328,8 +328,8 @@ impl InnerConnection {
}
}
backend::Message::ParameterStatus(body) => {
self.parameters.insert(try!(body.name()).to_owned(),
try!(body.value()).to_owned());
self.parameters.insert(body.name()?.to_owned(),
body.value()?.to_owned());
}
val => return Ok(val),
}
@ -348,8 +348,8 @@ impl InnerConnection {
}
}
Some(backend::Message::ParameterStatus(body)) => {
self.parameters.insert(try!(body.name()).to_owned(),
try!(body.value()).to_owned());
self.parameters.insert(body.name()?.to_owned(),
body.value()?.to_owned());
}
val => return Ok(val),
}
@ -367,8 +367,8 @@ impl InnerConnection {
}
}
Some(backend::Message::ParameterStatus(body)) => {
self.parameters.insert(try!(body.name()).to_owned(),
try!(body.value()).to_owned());
self.parameters.insert(body.name()?.to_owned(),
body.value()?.to_owned());
}
val => return Ok(val),
}
@ -377,12 +377,12 @@ impl InnerConnection {
fn read_message(&mut self) -> io::Result<backend::Message<Vec<u8>>> {
loop {
match try!(self.read_message_with_notification()) {
match self.read_message_with_notification()? {
backend::Message::NotificationResponse(body) => {
self.notifications.push_back(Notification {
process_id: body.process_id(),
channel: try!(body.channel()).to_owned(),
payload: try!(body.message()).to_owned(),
channel: body.channel()?.to_owned(),
payload: body.message()?.to_owned(),
})
}
val => return Ok(val),
@ -391,24 +391,24 @@ impl InnerConnection {
}
fn handle_auth(&mut self, user: UserInfo) -> result::Result<(), ConnectError> {
match try!(self.read_message()) {
match self.read_message()? {
backend::Message::AuthenticationOk => return Ok(()),
backend::Message::AuthenticationCleartextPassword => {
let pass = try!(user.password.ok_or_else(|| {
let pass = user.password.ok_or_else(|| {
ConnectError::ConnectParams("a password was requested but not provided".into())
}));
try!(self.stream.write_message(|buf| frontend::password_message(&pass, buf)));
try!(self.stream.flush());
})?;
self.stream.write_message(|buf| frontend::password_message(&pass, buf))?;
self.stream.flush()?;
}
backend::Message::AuthenticationMd5Password(body) => {
let pass = try!(user.password.ok_or_else(|| {
let pass = user.password.ok_or_else(|| {
ConnectError::ConnectParams("a password was requested but not provided".into())
}));
})?;
let output = authentication::md5_hash(user.user.as_bytes(),
pass.as_bytes(),
body.salt());
try!(self.stream.write_message(|buf| frontend::password_message(&output, buf)));
try!(self.stream.flush());
self.stream.write_message(|buf| frontend::password_message(&output, buf))?;
self.stream.flush()?;
}
backend::Message::AuthenticationKerberosV5 |
backend::Message::AuthenticationScmCredential |
@ -421,7 +421,7 @@ impl InnerConnection {
_ => return Err(ConnectError::Io(bad_response())),
}
match try!(self.read_message()) {
match self.read_message()? {
backend::Message::AuthenticationOk => Ok(()),
backend::Message::ErrorResponse(body) => Err(connect_err(&mut body.fields())),
_ => Err(ConnectError::Io(bad_response())),
@ -435,47 +435,47 @@ impl InnerConnection {
fn raw_prepare(&mut self, stmt_name: &str, query: &str) -> Result<(Vec<Type>, Vec<Column>)> {
debug!("preparing query with name `{}`: {}", stmt_name, query);
try!(self.stream.write_message(|buf| frontend::parse(stmt_name, query, None, buf)));
try!(self.stream.write_message(|buf| frontend::describe(b'S', stmt_name, buf)));
try!(self.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(self.stream.flush());
self.stream.write_message(|buf| frontend::parse(stmt_name, query, None, buf))?;
self.stream.write_message(|buf| frontend::describe(b'S', stmt_name, buf))?;
self.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
self.stream.flush()?;
match try!(self.read_message()) {
match self.read_message()? {
backend::Message::ParseComplete => {}
backend::Message::ErrorResponse(body) => {
try!(self.wait_for_ready());
self.wait_for_ready()?;
return Err(err(&mut body.fields()));
}
_ => bad_response!(self),
}
let raw_param_types = match try!(self.read_message()) {
let raw_param_types = match self.read_message()? {
backend::Message::ParameterDescription(body) => body,
_ => bad_response!(self),
};
let raw_columns = match try!(self.read_message()) {
let raw_columns = match self.read_message()? {
backend::Message::RowDescription(body) => Some(body),
backend::Message::NoData => None,
_ => bad_response!(self),
};
try!(self.wait_for_ready());
self.wait_for_ready()?;
let param_types = try!(raw_param_types
let param_types = raw_param_types
.parameters()
.map_err(Into::into)
.and_then(|oid| self.get_type(oid))
.collect());
.collect()?;
let columns = match raw_columns {
Some(body) => {
try!(body.fields()
body.fields()
.and_then(|field| {
Ok(Column::new(field.name().to_owned(),
try!(self.get_type(field.type_oid()))))
self.get_type(field.type_oid())?))
})
.collect())
.collect()?
}
None => vec![],
};
@ -488,7 +488,7 @@ impl InnerConnection {
{
let more_rows;
loop {
match try!(self.read_message()) {
match self.read_message()? {
backend::Message::EmptyQueryResponse |
backend::Message::CommandComplete(_) => {
more_rows = false;
@ -498,22 +498,22 @@ impl InnerConnection {
more_rows = true;
break;
}
backend::Message::DataRow(body) => consumer(try!(body.values().collect())),
backend::Message::DataRow(body) => consumer(body.values().collect()?),
backend::Message::ErrorResponse(body) => {
try!(self.wait_for_ready());
self.wait_for_ready()?;
return Err(err(&mut body.fields()));
}
backend::Message::CopyInResponse(_) => {
try!(self.stream.write_message(|buf| {
self.stream.write_message(|buf| {
frontend::copy_fail("COPY queries cannot be directly executed", buf)
}));
try!(self.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(self.stream.flush());
})?;
self.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
self.stream.flush()?;
}
backend::Message::CopyOutResponse(_) => {
loop {
if let backend::Message::ReadyForQuery(_) = try!(self.read_message()) {
if let backend::Message::ReadyForQuery(_) = self.read_message()? {
break;
}
}
@ -527,7 +527,7 @@ impl InnerConnection {
}
}
}
try!(self.wait_for_ready());
self.wait_for_ready()?;
Ok(more_rows)
}
@ -569,14 +569,14 @@ impl InnerConnection {
}
}
try!(self.stream.write_message(|buf| frontend::execute(portal_name, row_limit, buf)));
try!(self.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(self.stream.flush());
self.stream.write_message(|buf| frontend::execute(portal_name, row_limit, buf))?;
self.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
self.stream.flush()?;
match try!(self.read_message()) {
match self.read_message()? {
backend::Message::BindComplete => Ok(()),
backend::Message::ErrorResponse(body) => {
try!(self.wait_for_ready());
self.wait_for_ready()?;
Err(err(&mut body.fields()))
}
_ => {
@ -594,7 +594,7 @@ impl InnerConnection {
fn prepare<'a>(&mut self, query: &str, conn: &'a Connection) -> Result<Statement<'a>> {
let stmt_name = self.make_stmt_name();
let (param_types, columns) = try!(self.raw_prepare(&stmt_name, query));
let (param_types, columns) = self.raw_prepare(&stmt_name, query)?;
let info = Arc::new(StatementInfo {
name: stmt_name,
param_types: param_types,
@ -610,7 +610,7 @@ impl InnerConnection {
Some(info) => info,
None => {
let stmt_name = self.make_stmt_name();
let (param_types, columns) = try!(self.raw_prepare(&stmt_name, query));
let (param_types, columns) = self.raw_prepare(&stmt_name, query)?;
let info = Arc::new(StatementInfo {
name: stmt_name,
param_types: param_types,
@ -625,15 +625,15 @@ impl InnerConnection {
}
fn close_statement(&mut self, name: &str, type_: u8) -> Result<()> {
try!(self.stream.write_message(|buf| frontend::close(type_, name, buf)));
try!(self.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(self.stream.flush());
let resp = match try!(self.read_message()) {
self.stream.write_message(|buf| frontend::close(type_, name, buf))?;
self.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
self.stream.flush()?;
let resp = match self.read_message()? {
backend::Message::CloseComplete => Ok(()),
backend::Message::ErrorResponse(body) => Err(err(&mut body.fields())),
_ => bad_response!(self),
};
try!(self.wait_for_ready());
self.wait_for_ready()?;
resp
}
@ -646,7 +646,7 @@ impl InnerConnection {
return Ok(Type::Other(ty.clone()));
}
let ty = try!(self.read_type(oid));
let ty = self.read_type(oid)?;
self.unknown_types.insert(oid, ty.clone());
Ok(Type::Other(ty))
}
@ -668,13 +668,13 @@ impl InnerConnection {
Ok(..) => {}
// Range types weren't added until Postgres 9.2, so pg_range may not exist
Err(Error::Db(ref e)) if e.code == SqlState::UndefinedTable => {
try!(self.raw_prepare(TYPEINFO_QUERY,
"SELECT t.typname, t.typtype, t.typelem, NULL::OID, \
t.typbasetype, n.nspname, t.typrelid \
FROM pg_catalog.pg_type t \
INNER JOIN pg_catalog.pg_namespace n \
ON t.typnamespace = n.oid \
WHERE t.oid = $1"));
self.raw_prepare(TYPEINFO_QUERY,
"SELECT t.typname, t.typtype, t.typelem, NULL::OID, \
t.typbasetype, n.nspname, t.typrelid \
FROM pg_catalog.pg_type t \
INNER JOIN pg_catalog.pg_namespace n \
ON t.typnamespace = n.oid \
WHERE t.oid = $1")?;
}
Err(e) => return Err(e),
}
@ -685,46 +685,46 @@ impl InnerConnection {
#[allow(if_not_else)]
fn read_type(&mut self, oid: Oid) -> Result<Other> {
try!(self.setup_typeinfo_query());
try!(self.raw_execute(TYPEINFO_QUERY, "", 0, &[Type::Oid], &[&oid]));
self.setup_typeinfo_query()?;
self.raw_execute(TYPEINFO_QUERY, "", 0, &[Type::Oid], &[&oid])?;
let mut row = None;
try!(self.read_rows(|r| row = Some(r)));
self.read_rows(|r| row = Some(r))?;
let get_raw = |i: usize| {
row.as_ref().and_then(|r| r.get(i))
};
let (name, type_, elem_oid, rngsubtype, basetype, schema, relid) = {
let name = try!(String::from_sql_nullable(&Type::Name, get_raw(0))
.map_err(Error::Conversion));
let type_ = try!(i8::from_sql_nullable(&Type::Char, get_raw(1))
.map_err(Error::Conversion));
let elem_oid = try!(Oid::from_sql_nullable(&Type::Oid, get_raw(2))
.map_err(Error::Conversion));
let rngsubtype = try!(Option::<Oid>::from_sql_nullable(&Type::Oid, get_raw(3))
.map_err(Error::Conversion));
let basetype = try!(Oid::from_sql_nullable(&Type::Oid, get_raw(4))
.map_err(Error::Conversion));
let schema = try!(String::from_sql_nullable(&Type::Name, get_raw(5))
.map_err(Error::Conversion));
let relid = try!(Oid::from_sql_nullable(&Type::Oid, get_raw(6))
.map_err(Error::Conversion));
let name = String::from_sql_nullable(&Type::Name, get_raw(0))
.map_err(Error::Conversion)?;
let type_ = i8::from_sql_nullable(&Type::Char, get_raw(1))
.map_err(Error::Conversion)?;
let elem_oid = Oid::from_sql_nullable(&Type::Oid, get_raw(2))
.map_err(Error::Conversion)?;
let rngsubtype = Option::<Oid>::from_sql_nullable(&Type::Oid, get_raw(3))
.map_err(Error::Conversion)?;
let basetype = Oid::from_sql_nullable(&Type::Oid, get_raw(4))
.map_err(Error::Conversion)?;
let schema = String::from_sql_nullable(&Type::Name, get_raw(5))
.map_err(Error::Conversion)?;
let relid = Oid::from_sql_nullable(&Type::Oid, get_raw(6))
.map_err(Error::Conversion)?;
(name, type_, elem_oid, rngsubtype, basetype, schema, relid)
};
let kind = if type_ == b'e' as i8 {
Kind::Enum(try!(self.read_enum_variants(oid)))
Kind::Enum(self.read_enum_variants(oid)?)
} else if type_ == b'p' as i8 {
Kind::Pseudo
} else if basetype != 0 {
Kind::Domain(try!(self.get_type(basetype)))
Kind::Domain(self.get_type(basetype)?)
} else if elem_oid != 0 {
Kind::Array(try!(self.get_type(elem_oid)))
Kind::Array(self.get_type(elem_oid)?)
} else if relid != 0 {
Kind::Composite(try!(self.read_composite_fields(relid)))
Kind::Composite(self.read_composite_fields(relid)?)
} else {
match rngsubtype {
Some(oid) => Kind::Range(try!(self.get_type(oid))),
Some(oid) => Kind::Range(self.get_type(oid)?),
None => Kind::Simple,
}
};
@ -745,11 +745,11 @@ impl InnerConnection {
Ok(..) => {}
// Postgres 9.0 doesn't have enumsortorder
Err(Error::Db(ref e)) if e.code == SqlState::UndefinedColumn => {
try!(self.raw_prepare(TYPEINFO_ENUM_QUERY,
"SELECT enumlabel \
FROM pg_catalog.pg_enum \
WHERE enumtypid = $1 \
ORDER BY oid"));
self.raw_prepare(TYPEINFO_ENUM_QUERY,
"SELECT enumlabel \
FROM pg_catalog.pg_enum \
WHERE enumtypid = $1 \
ORDER BY oid")?;
}
Err(e) => return Err(e),
}
@ -759,15 +759,15 @@ impl InnerConnection {
}
fn read_enum_variants(&mut self, oid: Oid) -> Result<Vec<String>> {
try!(self.setup_typeinfo_enum_query());
try!(self.raw_execute(TYPEINFO_ENUM_QUERY, "", 0, &[Type::Oid], &[&oid]));
self.setup_typeinfo_enum_query()?;
self.raw_execute(TYPEINFO_ENUM_QUERY, "", 0, &[Type::Oid], &[&oid])?;
let mut rows = vec![];
try!(self.read_rows(|row| rows.push(row)));
self.read_rows(|row| rows.push(row))?;
let mut variants = vec![];
for row in rows {
variants.push(try!(String::from_sql_nullable(&Type::Name, row.get(0))
.map_err(Error::Conversion)));
variants.push(String::from_sql_nullable(&Type::Name, row.get(0))
.map_err(Error::Conversion)?);
}
Ok(variants)
@ -778,34 +778,34 @@ impl InnerConnection {
return Ok(());
}
try!(self.raw_prepare(TYPEINFO_COMPOSITE_QUERY,
"SELECT attname, atttypid \
FROM pg_catalog.pg_attribute \
WHERE attrelid = $1 \
AND NOT attisdropped \
AND attnum > 0 \
ORDER BY attnum"));
self.raw_prepare(TYPEINFO_COMPOSITE_QUERY,
"SELECT attname, atttypid \
FROM pg_catalog.pg_attribute \
WHERE attrelid = $1 \
AND NOT attisdropped \
AND attnum > 0 \
ORDER BY attnum")?;
self.has_typeinfo_composite_query = true;
Ok(())
}
fn read_composite_fields(&mut self, relid: Oid) -> Result<Vec<Field>> {
try!(self.setup_typeinfo_composite_query());
try!(self.raw_execute(TYPEINFO_COMPOSITE_QUERY, "", 0, &[Type::Oid], &[&relid]));
self.setup_typeinfo_composite_query()?;
self.raw_execute(TYPEINFO_COMPOSITE_QUERY, "", 0, &[Type::Oid], &[&relid])?;
let mut rows = vec![];
try!(self.read_rows(|row| rows.push(row)));
self.read_rows(|row| rows.push(row))?;
let mut fields = vec![];
for row in rows {
let (name, type_) = {
let name = try!(String::from_sql_nullable(&Type::Name, row.get(0))
.map_err(Error::Conversion));
let type_ = try!(Oid::from_sql_nullable(&Type::Oid, row.get(1))
.map_err(Error::Conversion));
let name = String::from_sql_nullable(&Type::Name, row.get(0))
.map_err(Error::Conversion)?;
let type_ = Oid::from_sql_nullable(&Type::Oid, row.get(1))
.map_err(Error::Conversion)?;
(name, type_)
};
let type_ = try!(self.get_type(type_));
let type_ = self.get_type(type_)?;
fields.push(Field::new(name, type_));
}
@ -818,7 +818,7 @@ impl InnerConnection {
#[allow(needless_return)]
fn wait_for_ready(&mut self) -> Result<()> {
match try!(self.read_message()) {
match self.read_message()? {
backend::Message::ReadyForQuery(_) => Ok(()),
_ => bad_response!(self),
}
@ -827,29 +827,29 @@ impl InnerConnection {
fn quick_query(&mut self, query: &str) -> Result<Vec<Vec<Option<String>>>> {
check_desync!(self);
debug!("executing query: {}", query);
try!(self.stream.write_message(|buf| frontend::query(query, buf)));
try!(self.stream.flush());
self.stream.write_message(|buf| frontend::query(query, buf))?;
self.stream.flush()?;
let mut result = vec![];
loop {
match try!(self.read_message()) {
match self.read_message()? {
backend::Message::ReadyForQuery(_) => break,
backend::Message::DataRow(body) => {
let row = try!(body.values()
let row = body.values()
.map(|v| v.map(|v| String::from_utf8_lossy(v).into_owned()))
.collect());
.collect()?;
result.push(row);
}
backend::Message::CopyInResponse(_) => {
try!(self.stream.write_message(|buf| {
self.stream.write_message(|buf| {
frontend::copy_fail("COPY queries cannot be directly executed", buf)
}));
try!(self.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(self.stream.flush());
})?;
self.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
self.stream.flush()?;
}
backend::Message::ErrorResponse(body) => {
try!(self.wait_for_ready());
self.wait_for_ready()?;
return Err(err(&mut body.fields()));
}
_ => {}
@ -860,8 +860,8 @@ impl InnerConnection {
fn finish_inner(&mut self) -> Result<()> {
check_desync!(self);
try!(self.stream.write_message(|buf| Ok::<(), io::Error>(frontend::terminate(buf))));
try!(self.stream.flush());
self.stream.write_message(|buf| Ok::<(), io::Error>(frontend::terminate(buf)))?;
self.stream.flush()?;
Ok(())
}
}
@ -985,7 +985,7 @@ impl Connection {
/// println!("{} rows updated", rows_updated);
/// ```
pub fn execute(&self, query: &str, params: &[&ToSql]) -> Result<u64> {
let (param_types, columns) = try!(self.0.borrow_mut().raw_prepare("", query));
let (param_types, columns) = self.0.borrow_mut().raw_prepare("", query)?;
let info = Arc::new(StatementInfo {
name: String::new(),
param_types: param_types,
@ -1021,7 +1021,7 @@ impl Connection {
/// }
/// ```
pub fn query<'a>(&'a self, query: &str, params: &[&ToSql]) -> Result<Rows<'a>> {
let (param_types, columns) = try!(self.0.borrow_mut().raw_prepare("", query));
let (param_types, columns) = self.0.borrow_mut().raw_prepare("", query)?;
let info = Arc::new(StatementInfo {
name: String::new(),
param_types: param_types,
@ -1068,7 +1068,7 @@ impl Connection {
"`transaction` must be called on the active transaction");
let mut query = "BEGIN".to_owned();
config.build_command(&mut query);
try!(conn.quick_query(&query));
conn.quick_query(&query)?;
conn.trans_depth += 1;
Ok(Transaction::new(self, 1))
}
@ -1128,7 +1128,7 @@ impl Connection {
pub fn transaction_isolation(&self) -> Result<IsolationLevel> {
let mut conn = self.0.borrow_mut();
check_desync!(conn);
let result = try!(conn.quick_query("SHOW TRANSACTION ISOLATION LEVEL"));
let result = conn.quick_query("SHOW TRANSACTION ISOLATION LEVEL")?;
IsolationLevel::new(result[0][0].as_ref().unwrap())
}

12
postgres/src/notification.rs
View File

@ -116,8 +116,8 @@ impl<'a> FallibleIterator for Iter<'a> {
Ok(Some(backend::Message::NotificationResponse(body))) => {
Ok(Some(Notification {
process_id: body.process_id(),
channel: try!(body.channel()).to_owned(),
payload: try!(body.message()).to_owned(),
channel: body.channel()?.to_owned(),
payload: body.message()?.to_owned(),
}))
}
Ok(None) => Ok(None),
@ -155,8 +155,8 @@ impl<'a> FallibleIterator for BlockingIter<'a> {
Ok(backend::Message::NotificationResponse(body)) => {
Ok(Some(Notification {
process_id: body.process_id(),
channel: try!(body.channel()).to_owned(),
payload: try!(body.message()).to_owned(),
channel: body.channel()?.to_owned(),
payload: body.message()?.to_owned(),
}))
}
Err(err) => Err(Error::Io(err)),
@ -191,8 +191,8 @@ impl<'a> FallibleIterator for TimeoutIter<'a> {
Ok(Some(backend::Message::NotificationResponse(body))) => {
Ok(Some(Notification {
process_id: body.process_id(),
channel: try!(body.channel()).to_owned(),
payload: try!(body.message()).to_owned(),
channel: body.channel()?.to_owned(),
payload: body.message()?.to_owned(),
}))
}
Ok(None) => Ok(None),

32
postgres/src/priv_io.rs
View File

@ -43,26 +43,26 @@ impl MessageStream {
E: From<io::Error>
{
self.buf.clear();
try!(f(&mut self.buf));
f(&mut self.buf)?;
self.stream.write_all(&self.buf).map_err(From::from)
}
fn inner_read_message(&mut self, b: u8) -> io::Result<backend::Message<Vec<u8>>> {
self.buf.resize(MESSAGE_HEADER_SIZE, 0);
self.buf[0] = b;
try!(self.stream.read_exact(&mut self.buf[1..]));
self.stream.read_exact(&mut self.buf[1..])?;
let len = match try!(backend::Message::parse_owned(&self.buf)) {
let len = match backend::Message::parse_owned(&self.buf)? {
ParseResult::Complete { message, .. } => return Ok(message),
ParseResult::Incomplete { required_size } => Some(required_size.unwrap()),
};
if let Some(len) = len {
self.buf.resize(len, 0);
try!(self.stream.read_exact(&mut self.buf[MESSAGE_HEADER_SIZE..]));
self.stream.read_exact(&mut self.buf[MESSAGE_HEADER_SIZE..])?;
};
match try!(backend::Message::parse_owned(&self.buf)) {
match backend::Message::parse_owned(&self.buf)? {
ParseResult::Complete { message, .. } => Ok(message),
ParseResult::Incomplete { .. } => unreachable!(),
}
@ -70,17 +70,17 @@ impl MessageStream {
pub fn read_message(&mut self) -> io::Result<backend::Message<Vec<u8>>> {
let mut b = [0; 1];
try!(self.stream.read_exact(&mut b));
self.stream.read_exact(&mut b)?;
self.inner_read_message(b[0])
}
pub fn read_message_timeout(&mut self,
timeout: Duration)
-> io::Result<Option<backend::Message<Vec<u8>>>> {
try!(self.set_read_timeout(Some(timeout)));
self.set_read_timeout(Some(timeout))?;
let mut b = [0; 1];
let r = self.stream.read_exact(&mut b);
try!(self.set_read_timeout(None));
self.set_read_timeout(None)?;
match r {
Ok(()) => self.inner_read_message(b[0]).map(Some),
@ -91,10 +91,10 @@ impl MessageStream {
}
pub fn read_message_nonblocking(&mut self) -> io::Result<Option<backend::Message<Vec<u8>>>> {
try!(self.set_nonblocking(true));
self.set_nonblocking(true)?;
let mut b = [0; 1];
let r = self.stream.read_exact(&mut b);
try!(self.set_nonblocking(false));
self.set_nonblocking(false)?;
match r {
Ok(()) => self.inner_read_message(b[0]).map(Some),
@ -224,12 +224,12 @@ fn open_socket(params: &ConnectParams) -> Result<InternalStream, ConnectError> {
let port = params.port.unwrap_or(DEFAULT_PORT);
match params.target {
ConnectTarget::Tcp(ref host) => {
Ok(try!(TcpStream::connect(&(&**host, port)).map(InternalStream::Tcp)))
Ok(TcpStream::connect(&(&**host, port)).map(InternalStream::Tcp)?)
}
#[cfg(unix)]
ConnectTarget::Unix(ref path) => {
let path = path.join(&format!(".s.PGSQL.{}", port));
Ok(try!(UnixStream::connect(&path).map(InternalStream::Unix)))
Ok(UnixStream::connect(&path).map(InternalStream::Unix)?)
}
#[cfg(not(unix))]
ConnectTarget::Unix(..) => {
@ -242,7 +242,7 @@ fn open_socket(params: &ConnectParams) -> Result<InternalStream, ConnectError> {
pub fn initialize_stream(params: &ConnectParams,
tls: TlsMode)
-> Result<Box<TlsStream>, ConnectError> {
let mut socket = Stream(try!(open_socket(params)));
let mut socket = Stream(open_socket(params)?);
let (tls_required, handshaker) = match tls {
TlsMode::None => return Ok(Box::new(socket)),
@ -252,11 +252,11 @@ pub fn initialize_stream(params: &ConnectParams,
let mut buf = vec![];
frontend::ssl_request(&mut buf);
try!(socket.write_all(&buf));
try!(socket.flush());
socket.write_all(&buf)?;
socket.flush()?;
let mut b = [0; 1];
try!(socket.read_exact(&mut b));
socket.read_exact(&mut b)?;
if b[0] == b'N' {
if tls_required {
return Err(ConnectError::Tls("the server does not support TLS".into()));

8
postgres/src/rows.rs
View File

@ -348,9 +348,9 @@ impl<'trans, 'stmt> LazyRows<'trans, 'stmt> {
fn execute(&mut self) -> Result<()> {
let mut conn = self.stmt.conn().0.borrow_mut();
try!(conn.stream.write_message(|buf| frontend::execute(&self.name, self.row_limit, buf)));
try!(conn.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(conn.stream.flush());
conn.stream.write_message(|buf| frontend::execute(&self.name, self.row_limit, buf))?;
conn.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
conn.stream.flush()?;
conn.read_rows(|row| self.data.push_back(row)).map(|more_rows| self.more_rows = more_rows)
}
@ -374,7 +374,7 @@ impl<'trans, 'stmt> FallibleIterator for LazyRows<'trans, 'stmt> {
fn next(&mut self) -> Result<Option<Row<'stmt>>> {
if self.data.is_empty() && self.more_rows {
try!(self.execute());
self.execute()?;
}
let row = self.data

131
postgres/src/stmt.rs
View File

@ -61,7 +61,7 @@ impl<'conn> StatementInternals<'conn> for Statement<'conn> {
fn into_query(self, params: &[&ToSql]) -> Result<Rows<'conn>> {
check_desync!(self.conn);
let mut rows = vec![];
try!(self.inner_query("", 0, params, |row| rows.push(row)));
self.inner_query("", 0, params, |row| rows.push(row))?;
Ok(Rows::new_owned(self, rows))
}
}
@ -89,11 +89,11 @@ impl<'conn> Statement<'conn> {
{
let mut conn = self.conn.0.borrow_mut();
try!(conn.raw_execute(&self.info.name,
portal_name,
row_limit,
self.param_types(),
params));
conn.raw_execute(&self.info.name,
portal_name,
row_limit,
self.param_types(),
params)?;
conn.read_rows(acceptor)
}
@ -131,18 +131,18 @@ impl<'conn> Statement<'conn> {
pub fn execute(&self, params: &[&ToSql]) -> Result<u64> {
let mut conn = self.conn.0.borrow_mut();
check_desync!(conn);
try!(conn.raw_execute(&self.info.name, "", 0, self.param_types(), params));
conn.raw_execute(&self.info.name, "", 0, self.param_types(), params)?;
let num;
loop {
match try!(conn.read_message()) {
match conn.read_message()? {
backend::Message::DataRow(_) => {}
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
return Err(err(&mut body.fields()));
}
backend::Message::CommandComplete(body) => {
num = parse_update_count(try!(body.tag()));
num = parse_update_count(body.tag()?);
break;
}
backend::Message::EmptyQueryResponse => {
@ -150,19 +150,19 @@ impl<'conn> Statement<'conn> {
break;
}
backend::Message::CopyInResponse(_) => {
try!(conn.stream.write_message(|buf| {
conn.stream.write_message(|buf| {
frontend::copy_fail("COPY queries cannot be directly executed", buf)
}));
try!(conn.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(conn.stream.flush());
})?;
conn.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
conn.stream.flush()?;
}
backend::Message::CopyOutResponse(_) => {
loop {
match try!(conn.read_message()) {
match conn.read_message()? {
backend::Message::CopyDone => break,
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
return Err(err(&mut body.fields()));
}
_ => {}
@ -177,7 +177,7 @@ impl<'conn> Statement<'conn> {
}
}
}
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
Ok(num)
}
@ -204,7 +204,7 @@ impl<'conn> Statement<'conn> {
pub fn query<'a>(&'a self, params: &[&ToSql]) -> Result<Rows<'a>> {
check_desync!(self.conn);
let mut rows = vec![];
try!(self.inner_query("", 0, params, |row| rows.push(row)));
self.inner_query("", 0, params, |row| rows.push(row))?;
Ok(Rows::new(self, rows))
}
@ -245,10 +245,10 @@ impl<'conn> Statement<'conn> {
let portal_name = format!("{}p{}", self.info.name, id);
let mut rows = VecDeque::new();
let more_rows = try!(self.inner_query(&portal_name,
row_limit,
params,
|row| rows.push_back(row)));
let more_rows = self.inner_query(&portal_name,
row_limit,
params,
|row| rows.push_back(row))?;
Ok(LazyRows::new(self, rows, portal_name, row_limit, more_rows, false, trans))
}
@ -275,23 +275,23 @@ impl<'conn> Statement<'conn> {
/// ```
pub fn copy_in<R: ReadWithInfo>(&self, params: &[&ToSql], r: &mut R) -> Result<u64> {
let mut conn = self.conn.0.borrow_mut();
try!(conn.raw_execute(&self.info.name, "", 0, self.param_types(), params));
conn.raw_execute(&self.info.name, "", 0, self.param_types(), params)?;
let (format, column_formats) = match try!(conn.read_message()) {
let (format, column_formats) = match conn.read_message()? {
backend::Message::CopyInResponse(body) => {
let format = body.format();
let column_formats = try!(body.column_formats()
let column_formats = body.column_formats()
.map(|f| Format::from_u16(f))
.collect());
.collect()?;
(format, column_formats)
}
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
return Err(err(&mut body.fields()));
}
_ => {
loop {
if let backend::Message::ReadyForQuery(_) = try!(conn.read_message()) {
if let backend::Message::ReadyForQuery(_) = conn.read_message()? {
return Err(Error::Io(io::Error::new(io::ErrorKind::InvalidInput,
"called `copy_in` on a \
non-`COPY FROM STDIN` \
@ -311,19 +311,16 @@ impl<'conn> Statement<'conn> {
match fill_copy_buf(&mut buf, r, &info) {
Ok(0) => break,
Ok(len) => {
try!(conn
.stream.write_message(|out| frontend::copy_data(&buf[..len], out)));
conn.stream.write_message(|out| frontend::copy_data(&buf[..len], out))?;
}
Err(err) => {
try!(conn.stream.write_message(|buf| frontend::copy_fail("", buf)));
try!(conn
.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::copy_done(buf))));
try!(conn
.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(conn.stream.flush());
match try!(conn.read_message()) {
conn.stream.write_message(|buf| frontend::copy_fail("", buf))?;
conn.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::copy_done(buf)))?;
conn.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
conn.stream.flush()?;
match conn.read_message()? {
backend::Message::ErrorResponse(_) => {
// expected from the CopyFail
}
@ -332,20 +329,20 @@ impl<'conn> Statement<'conn> {
return Err(Error::Io(bad_response()));
}
}
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
return Err(Error::Io(err));
}
}
}
try!(conn.stream.write_message(|buf| Ok::<(), io::Error>(frontend::copy_done(buf))));
try!(conn.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(conn.stream.flush());
conn.stream.write_message(|buf| Ok::<(), io::Error>(frontend::copy_done(buf)))?;
conn.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
conn.stream.flush()?;
let num = match try!(conn.read_message()) {
backend::Message::CommandComplete(body) => parse_update_count(try!(body.tag())),
let num = match conn.read_message()? {
backend::Message::CommandComplete(body) => parse_update_count(body.tag()?),
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
return Err(err(&mut body.fields()));
}
_ => {
@ -354,7 +351,7 @@ impl<'conn> Statement<'conn> {
}
};
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
Ok(num)
}
@ -382,23 +379,23 @@ impl<'conn> Statement<'conn> {
/// ```
pub fn copy_out<'a, W: WriteWithInfo>(&'a self, params: &[&ToSql], w: &mut W) -> Result<u64> {
let mut conn = self.conn.0.borrow_mut();
try!(conn.raw_execute(&self.info.name, "", 0, self.param_types(), params));
conn.raw_execute(&self.info.name, "", 0, self.param_types(), params)?;
let (format, column_formats) = match try!(conn.read_message()) {
let (format, column_formats) = match conn.read_message()? {
backend::Message::CopyOutResponse(body) => {
let format = body.format();
let column_formats = try!(body.column_formats()
let column_formats = body.column_formats()
.map(|f| Format::from_u16(f))
.collect());
.collect()?;
(format, column_formats)
}
backend::Message::CopyInResponse(_) => {
try!(conn.stream.write_message(|buf| frontend::copy_fail("", buf)));
try!(conn.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::copy_done(buf))));
try!(conn.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf))));
try!(conn.stream.flush());
match try!(conn.read_message()) {
conn.stream.write_message(|buf| frontend::copy_fail("", buf))?;
conn.stream
.write_message(|buf| Ok::<(), io::Error>(frontend::copy_done(buf)))?;
conn.stream.write_message(|buf| Ok::<(), io::Error>(frontend::sync(buf)))?;
conn.stream.flush()?;
match conn.read_message()? {
backend::Message::ErrorResponse(_) => {
// expected from the CopyFail
}
@ -407,18 +404,18 @@ impl<'conn> Statement<'conn> {
return Err(Error::Io(bad_response()));
}
}
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
return Err(Error::Io(io::Error::new(io::ErrorKind::InvalidInput,
"called `copy_out` on a non-`COPY TO \
STDOUT` statement")));
}
backend::Message::ErrorResponse(body) => {
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
return Err(err(&mut body.fields()));
}
_ => {
loop {
if let backend::Message::ReadyForQuery(_) = try!(conn.read_message()) {
if let backend::Message::ReadyForQuery(_) = conn.read_message()? {
return Err(Error::Io(io::Error::new(io::ErrorKind::InvalidInput,
"called `copy_out` on a \
non-`COPY TO STDOUT` statement")));
@ -434,7 +431,7 @@ impl<'conn> Statement<'conn> {
let count;
loop {
match try!(conn.read_message()) {
match conn.read_message()? {
backend::Message::CopyData(body) => {
let mut data = body.data();
while !data.is_empty() {
@ -443,7 +440,7 @@ impl<'conn> Statement<'conn> {
Err(e) => {
loop {
if let backend::Message::ReadyForQuery(_) =
try!(conn.read_message()) {
conn.read_message()? {
return Err(Error::Io(e));
}
}
@ -453,13 +450,13 @@ impl<'conn> Statement<'conn> {
}
backend::Message::CopyDone => {}
backend::Message::CommandComplete(body) => {
count = parse_update_count(try!(body.tag()));
count = parse_update_count(body.tag()?);
break;
}
backend::Message::ErrorResponse(body) => {
loop {
if let backend::Message::ReadyForQuery(_) =
try!(conn.read_message()) {
conn.read_message()? {
return Err(err(&mut body.fields()));
}
}
@ -467,7 +464,7 @@ impl<'conn> Statement<'conn> {
_ => {
loop {
if let backend::Message::ReadyForQuery(_) =
try!(conn.read_message()) {
conn.read_message()? {
return Err(Error::Io(bad_response()));
}
}
@ -475,7 +472,7 @@ impl<'conn> Statement<'conn> {
}
}
try!(conn.wait_for_ready());
conn.wait_for_ready()?;
Ok(count)
}

6
postgres/src/tls/native_tls.rs
View File

@ -31,8 +31,8 @@ impl fmt::Debug for NativeTls {
impl NativeTls {
/// Creates a new `NativeTls` with its default configuration.
pub fn new() -> Result<NativeTls, native_tls::Error> {
let connector = try!(TlsConnector::builder());
let connector = try!(connector.build());
let connector = TlsConnector::builder()?;
let connector = connector.build()?;
Ok(NativeTls(connector))
}
@ -58,7 +58,7 @@ impl TlsHandshake for NativeTls {
domain: &str,
stream: Stream)
-> Result<Box<TlsStream>, Box<Error + Send + Sync>> {
let stream = try!(self.0.connect(domain, stream));
let stream = self.0.connect(domain, stream)?;
Ok(Box::new(stream))
}
}

6
postgres/src/tls/openssl.rs
View File

@ -35,7 +35,7 @@ impl fmt::Debug for OpenSsl {
impl OpenSsl {
/// Creates a `OpenSsl` with `SslConnector`'s default configuration.
pub fn new() -> Result<OpenSsl, ErrorStack> {
let connector = try!(SslConnectorBuilder::new(SslMethod::tls())).build();
let connector = SslConnectorBuilder::new(SslMethod::tls())?.build();
Ok(OpenSsl::from(connector))
}
@ -75,9 +75,9 @@ impl TlsHandshake for OpenSsl {
stream: Stream)
-> Result<Box<TlsStream>, Box<Error + Send + Sync>> {
let stream = if self.disable_verification {
try!(self.connector.danger_connect_without_providing_domain_for_certificate_verification_and_server_name_indication(stream))
self.connector.danger_connect_without_providing_domain_for_certificate_verification_and_server_name_indication(stream)?
} else {
try!(self.connector.connect(domain, stream))
self.connector.connect(domain, stream)?
};
Ok(Box::new(stream))
}

6
postgres/src/tls/schannel.rs
View File

@ -42,10 +42,10 @@ impl TlsHandshake for Schannel {
host: &str,
stream: Stream)
-> Result<Box<TlsStream>, Box<Error + Sync + Send>> {
let creds = try!(SchannelCred::builder().acquire(Direction::Outbound));
let stream = try!(tls_stream::Builder::new()
let creds = SchannelCred::builder().acquire(Direction::Outbound)?;
let stream = tls_stream::Builder::new()
.domain(host)
.connect(creds, stream));
.connect(creds, stream)?;
Ok(Box::new(stream))
}
}

2
postgres/src/tls/security_framework.rs
View File

@ -49,7 +49,7 @@ impl TlsHandshake for SecurityFramework {
domain: &str,
stream: Stream)
-> Result<Box<TlsStream>, Box<Error + Send + Sync>> {
let stream = try!(self.0.handshake(domain, stream));
let stream = self.0.handshake(domain, stream)?;
Ok(Box::new(stream))
}
}

10
postgres/src/transaction.rs
View File

@ -198,10 +198,10 @@ impl<'conn> Transaction<'conn> {
debug_assert!(self.depth == conn.trans_depth);
conn.trans_depth -= 1;
match (self.commit.get(), &self.savepoint_name) {
(false, &Some(ref sp)) => try!(conn.quick_query(&format!("ROLLBACK TO {}", sp))),
(false, &None) => try!(conn.quick_query("ROLLBACK")),
(true, &Some(ref sp)) => try!(conn.quick_query(&format!("RELEASE {}", sp))),
(true, &None) => try!(conn.quick_query("COMMIT")),
(false, &Some(ref sp)) => conn.quick_query(&format!("ROLLBACK TO {}", sp))?,
(false, &None) => conn.quick_query("ROLLBACK")?,
(true, &Some(ref sp)) => conn.quick_query(&format!("RELEASE {}", sp))?,
(true, &None) => conn.quick_query("COMMIT")?,
};
Ok(())
@ -258,7 +258,7 @@ impl<'conn> Transaction<'conn> {
check_desync!(conn);
assert!(conn.trans_depth == self.depth,
"`savepoint` may only be called on the active transaction");
try!(conn.quick_query(&format!("SAVEPOINT {}", name)));
conn.quick_query(&format!("SAVEPOINT {}", name))?;
conn.trans_depth += 1;
Ok(Transaction {
conn: self.conn,

26
postgres/src/url.rs
View File

@ -51,17 +51,17 @@ impl Url {
pub fn parse(rawurl: &str) -> DecodeResult<Url> {
// scheme
let (scheme, rest) = try!(get_scheme(rawurl));
let (scheme, rest) = get_scheme(rawurl)?;
// authority
let (userinfo, host, port, rest) = try!(get_authority(rest));
let (userinfo, host, port, rest) = get_authority(rest)?;
// path
let has_authority = !host.is_empty();
let (path, rest) = try!(get_path(rest, has_authority));
let (path, rest) = get_path(rest, has_authority)?;
// query and fragment
let (query, fragment) = try!(get_query_fragment(rest));
let (query, fragment) = get_query_fragment(rest)?;
let url = Url::new(scheme.to_owned(),
userinfo,
@ -84,10 +84,10 @@ impl Path {
}
pub fn parse(rawpath: &str) -> DecodeResult<Path> {
let (path, rest) = try!(get_path(rawpath, false));
let (path, rest) = get_path(rawpath, false)?;
// query and fragment
let (query, fragment) = try!(get_query_fragment(&rest));
let (query, fragment) = get_query_fragment(&rest)?;
Ok(Path {
path: path,
@ -177,7 +177,7 @@ fn query_from_str(rawquery: &str) -> DecodeResult<Query> {
if !rawquery.is_empty() {
for p in rawquery.split('&') {
let (k, v) = split_char_first(p, '=');
query.push((try!(decode_component(k)), try!(decode_component(v))));
query.push((decode_component(k)?, decode_component(v)?));
}
}
@ -316,13 +316,13 @@ fn get_authority(rawurl: &str) -> DecodeResult<(Option<UserInfo>, &str, Option<u
colon_count = 0; // reset count
match st {
State::Start => {
let user = try!(decode_component(&rawurl[begin..i]));
let user = decode_component(&rawurl[begin..i])?;
userinfo = Some(UserInfo::new(user, None));
st = State::InHost;
}
State::PassHostPort => {
let user = try!(decode_component(&rawurl[begin..pos]));
let pass = try!(decode_component(&rawurl[pos + 1..i]));
let user = decode_component(&rawurl[begin..pos])?;
let pass = decode_component(&rawurl[pos + 1..i])?;
userinfo = Some(UserInfo::new(user, Some(pass)));
st = State::InHost;
}
@ -392,7 +392,7 @@ fn get_path(rawurl: &str, is_authority: bool) -> DecodeResult<(String, &str)> {
if is_authority && end != 0 && !rawurl.starts_with('/') {
Err("Non-empty path must begin with '/' in presence of authority.".to_owned())
} else {
Ok((try!(decode_component(&rawurl[0..end])), &rawurl[end..len]))
Ok((decode_component(&rawurl[0..end])?, &rawurl[end..len]))
}
}
@ -403,11 +403,11 @@ fn get_query_fragment(rawurl: &str) -> DecodeResult<(Query, Option<String>)> {
// Parse the fragment if available
let fragment = match raw_fragment {
"" => None,
raw => Some(try!(decode_component(raw))),
raw => Some(decode_component(raw)?),
};
match before_fragment.chars().next() {
Some('?') => Ok((try!(query_from_str(&before_fragment[1..])), fragment)),
Some('?') => Ok((query_from_str(&before_fragment[1..])?, fragment)),
None => Ok((vec![], fragment)),
_ => Err(format!("Query didn't start with '?': '{}..'", before_fragment)),
}

2
tokio-postgres/src/error.rs
View File

@ -24,7 +24,7 @@ pub enum Error<C = Connection> {
impl fmt::Display for Error {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
try!(fmt.write_str(error::Error::description(self)));
fmt.write_str(error::Error::description(self))?;
match *self {
Error::Db(ref err, _) => write!(fmt, ": {}", err),
Error::Io(ref err) => write!(fmt, ": {}", err),

2
tokio-postgres/src/stream.rs
View File

@ -136,7 +136,7 @@ impl Codec for PostgresCodec {
// FIXME ideally we'd avoid re-copying the data
fn decode(&mut self, buf: &mut EasyBuf) -> io::Result<Option<Self::In>> {
match try!(backend::Message::parse_owned(buf.as_ref())) {
match backend::Message::parse_owned(buf.as_ref())? {
ParseResult::Complete { message, consumed } => {
buf.drain_to(consumed);
Ok(Some(message))

4
tokio-postgres/src/tls/openssl.rs
View File

@ -24,7 +24,7 @@ pub struct OpenSsl(SslConnector);
impl OpenSsl {
/// Creates a new `OpenSsl` with default settings.
pub fn new() -> Result<OpenSsl, ErrorStack> {
let connector = try!(SslConnectorBuilder::new(SslMethod::tls())).build();
let connector = SslConnectorBuilder::new(SslMethod::tls())?.build();
Ok(OpenSsl(connector))
}
}
@ -51,4 +51,4 @@ impl Handshake for OpenSsl {
})
.boxed()
}
}
}