diff --git a/Cargo.toml b/Cargo.toml index d176488d..c6493bb0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,23 +19,25 @@ postgres = [] mariadb = [] [dependencies] -bitflags = "1.1.0" -byteorder = "1.3.2" +# bitflags = "1.1.0" +byteorder = { version = "1.3.2", default-features = false } bytes = "0.4.12" crossbeam-queue = "0.1.2" -crossbeam-utils = "0.6.6" -enum-tryfrom = "0.2.1" -enum-tryfrom-derive = "0.2.1" -failure = "0.1.5" -futures-preview = "=0.3.0-alpha.18" -hex = "0.3.2" -itoa = "0.4.4" +crossbeam-utils = { version = "0.6.6", default-features = false } +# enum-tryfrom = "0.2.1" +# enum-tryfrom-derive = "0.2.1" +# failure = "0.1.5" +futures-util-preview = "=0.3.0-alpha.18" +futures-channel-preview = "=0.3.0-alpha.18" +futures-core-preview = "=0.3.0-alpha.18" +hex = { version = "0.3.2", default-features = false } +# itoa = "0.4.4" log = "0.4.8" md-5 = "0.8.0" url = "2.1.0" memchr = "2.2.1" async-stream = "0.1.0" -tokio = { version = "=0.2.0-alpha.2" } +tokio = { version = "=0.2.0-alpha.2", default-features = false, features = [ "tcp" ] } [dev-dependencies] criterion = "0.2" diff --git a/rust-toolchain b/rust-toolchain index 80e14c00..44134ce4 100644 --- a/rust-toolchain +++ b/rust-toolchain @@ -1 +1 @@ -nightly-2019-08-19 +nightly-2019-08-15 diff --git a/src/connection.rs b/src/connection.rs index 1a373f0d..5ea6d8a0 100644 --- a/src/connection.rs +++ b/src/connection.rs @@ -1,11 +1,9 @@ use crate::{backend::Backend, executor::Executor, query::RawQuery, row::FromRow}; use crossbeam_queue::SegQueue; use crossbeam_utils::atomic::AtomicCell; -use futures::{ - channel::oneshot::{channel, Sender}, - future::BoxFuture, - stream::{BoxStream, StreamExt}, -}; +use futures_channel::oneshot::{channel, Sender}; +use futures_core::{future::BoxFuture, stream::BoxStream}; +use futures_util::stream::StreamExt; use std::{ io, ops::{Deref, DerefMut}, @@ -77,10 +75,7 @@ where } async fn get(&self) -> ConnectionFairy<'_, DB> { - let raw = self.0.acquire().await; - let conn = ConnectionFairy::new(&self.0, raw); - - conn + ConnectionFairy::new(&self.0, self.0.acquire().await) } } diff --git a/src/executor.rs b/src/executor.rs index 3c7b50b7..db3c5deb 100644 --- a/src/executor.rs +++ b/src/executor.rs @@ -1,5 +1,5 @@ use crate::{backend::Backend, query::RawQuery, row::FromRow}; -use futures::{future::BoxFuture, stream::BoxStream}; +use futures_core::{future::BoxFuture, stream::BoxStream}; use std::io; pub trait Executor: Send { diff --git a/src/lib.rs b/src/lib.rs index 0bbc9c51..3989a080 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,16 +1,12 @@ #![feature(async_await)] -#![cfg_attr(test, feature(test))] #![allow(clippy::needless_lifetimes)] -// FIXME: Remove this once API has matured -#![allow(dead_code, unused_imports, unused_variables)] +#![allow(unused)] -#[cfg(test)] -extern crate test; +// #[macro_use] +// extern crate bitflags; -#[macro_use] -extern crate bitflags; -#[macro_use] -extern crate enum_tryfrom_derive; +// #[macro_use] +// extern crate enum_tryfrom_derive; #[macro_use] mod macros; @@ -41,5 +37,4 @@ pub use self::{ query::{query, SqlQuery}, }; -// TODO: Remove after Mariadb transitions to URIs -mod options; +// mod options; diff --git a/src/pg/connection/fetch.rs b/src/pg/connection/fetch.rs index 93537136..8a4f89c7 100644 --- a/src/pg/connection/fetch.rs +++ b/src/pg/connection/fetch.rs @@ -1,6 +1,6 @@ use super::{PgRawConnection, PgRow}; use crate::pg::protocol::Message; -use futures::stream::Stream; +use futures_core::stream::Stream; use std::io; pub fn fetch<'a>( diff --git a/src/pg/connection/mod.rs b/src/pg/connection/mod.rs index 5df762e6..4727cdb3 100644 --- a/src/pg/connection/mod.rs +++ b/src/pg/connection/mod.rs @@ -1,24 +1,16 @@ use super::{ - protocol::{Authentication, Encode, Message, PasswordMessage, StartupMessage, Terminate}, - Pg, PgRawQuery, PgRow, + protocol::{Encode, Message, Terminate}, + Pg, PgRow, }; -use crate::{connection::RawConnection, query::RawQuery, row::FromRow}; +use crate::{connection::RawConnection, query::RawQuery}; use bytes::{BufMut, BytesMut}; -use futures::{ - future::BoxFuture, - ready, - stream::{self, BoxStream, Stream}, - task::{Context, Poll}, - Future, -}; +use futures_core::{future::BoxFuture, stream::BoxStream}; use std::{ - fmt::Debug, io, net::{IpAddr, Shutdown, SocketAddr}, - pin::Pin, }; use tokio::{ - io::{AsyncReadExt, AsyncWrite, AsyncWriteExt}, + io::{AsyncReadExt, AsyncWriteExt}, net::TcpStream, }; use url::Url; diff --git a/src/pg/protocol/authentication.rs b/src/pg/protocol/authentication.rs index 84b69ba6..e8a88ceb 100644 --- a/src/pg/protocol/authentication.rs +++ b/src/pg/protocol/authentication.rs @@ -1,6 +1,4 @@ use super::Decode; -use bytes::Bytes; -use std::io; #[derive(Debug)] pub enum Authentication { diff --git a/src/pg/protocol/backend_key_data.rs b/src/pg/protocol/backend_key_data.rs index 0e074eb3..e0dbd0cc 100644 --- a/src/pg/protocol/backend_key_data.rs +++ b/src/pg/protocol/backend_key_data.rs @@ -1,6 +1,5 @@ use super::Decode; -use bytes::Bytes; -use std::{convert::TryInto, io}; +use std::convert::TryInto; #[derive(Debug)] pub struct BackendKeyData { diff --git a/src/pg/protocol/command_complete.rs b/src/pg/protocol/command_complete.rs index 39a8d475..4bc52094 100644 --- a/src/pg/protocol/command_complete.rs +++ b/src/pg/protocol/command_complete.rs @@ -1,7 +1,6 @@ use super::Decode; -use bytes::Bytes; use memchr::memrchr; -use std::{io, str}; +use std::str; #[derive(Debug)] pub struct CommandComplete { @@ -26,7 +25,6 @@ impl Decode for CommandComplete { #[cfg(test)] mod tests { use super::{CommandComplete, Decode}; - use bytes::Bytes; const COMMAND_COMPLETE_INSERT: &[u8] = b"INSERT 0 1\0"; const COMMAND_COMPLETE_UPDATE: &[u8] = b"UPDATE 512\0"; diff --git a/src/pg/protocol/data_row.rs b/src/pg/protocol/data_row.rs index 5dd8199e..e4fdbbaa 100644 --- a/src/pg/protocol/data_row.rs +++ b/src/pg/protocol/data_row.rs @@ -1,10 +1,7 @@ use super::Decode; -use bytes::Bytes; use std::{ convert::TryInto, fmt::{self, Debug}, - io, - ops::Range, pin::Pin, ptr::NonNull, }; @@ -84,7 +81,6 @@ impl Debug for DataRow { #[cfg(test)] mod tests { use super::{DataRow, Decode}; - use bytes::Bytes; use std::io; const DATA_ROW: &[u8] = b"\0\x03\0\0\0\x011\0\0\0\x012\0\0\0\x013"; diff --git a/src/pg/protocol/decode.rs b/src/pg/protocol/decode.rs index 51d2005b..76dc6b80 100644 --- a/src/pg/protocol/decode.rs +++ b/src/pg/protocol/decode.rs @@ -1,6 +1,5 @@ -use bytes::Bytes; use memchr::memchr; -use std::{io, str}; +use std::str; pub trait Decode { fn decode(src: &[u8]) -> Self diff --git a/src/pg/protocol/encode.rs b/src/pg/protocol/encode.rs index 8c1ed043..6687936c 100644 --- a/src/pg/protocol/encode.rs +++ b/src/pg/protocol/encode.rs @@ -1,5 +1,3 @@ -use std::io; - pub trait Encode { fn encode(&self, buf: &mut Vec); } diff --git a/src/pg/protocol/message.rs b/src/pg/protocol/message.rs index fc7a4187..fb9ae18d 100644 --- a/src/pg/protocol/message.rs +++ b/src/pg/protocol/message.rs @@ -41,14 +41,14 @@ impl Message { let token = src[0]; if token == 0 { // FIXME: Handle end-of-stream - return Err(io::ErrorKind::InvalidData)?; + return Err(io::ErrorKind::InvalidData.into()); } // FIXME: What happens if len(u32) < len(usize) ? let len = BigEndian::read_u32(&src[1..5]) as usize; if src.len() >= (len + 1) { - let window = &src[5..(len + 1)]; + let window = &src[5..=len]; let message = match token { b'N' | b'E' => Message::Response(Box::new(Response::decode(window))), diff --git a/src/pg/protocol/notification_response.rs b/src/pg/protocol/notification_response.rs index 4582855d..1af1afc0 100644 --- a/src/pg/protocol/notification_response.rs +++ b/src/pg/protocol/notification_response.rs @@ -1,8 +1,6 @@ use super::{decode::get_str, Decode}; use byteorder::{BigEndian, ByteOrder}; -use bytes::Bytes; - -use std::{fmt, io, pin::Pin, ptr::NonNull}; +use std::{fmt, pin::Pin, ptr::NonNull}; pub struct NotificationResponse { #[used] @@ -72,7 +70,6 @@ impl Decode for NotificationResponse { #[cfg(test)] mod tests { use super::{Decode, NotificationResponse}; - use bytes::Bytes; use std::io; const NOTIFICATION_RESPONSE: &[u8] = b"\x34\x20\x10\x02TEST-CHANNEL\0THIS IS A TEST\0"; diff --git a/src/pg/protocol/parameter_description.rs b/src/pg/protocol/parameter_description.rs index 58462d3e..7fc85245 100644 --- a/src/pg/protocol/parameter_description.rs +++ b/src/pg/protocol/parameter_description.rs @@ -1,6 +1,5 @@ use super::Decode; use byteorder::{BigEndian, ByteOrder}; -use bytes::Bytes; use std::mem::size_of; type ObjectId = u32; @@ -29,7 +28,6 @@ impl Decode for ParameterDescription { #[cfg(test)] mod test { use super::{Decode, ParameterDescription}; - use bytes::Bytes; use std::io; #[test] diff --git a/src/pg/protocol/parameter_status.rs b/src/pg/protocol/parameter_status.rs index 4ae69dff..d82226a4 100644 --- a/src/pg/protocol/parameter_status.rs +++ b/src/pg/protocol/parameter_status.rs @@ -1,5 +1,5 @@ use super::decode::{get_str, Decode}; -use std::{io, pin::Pin, ptr::NonNull, str}; +use std::{pin::Pin, ptr::NonNull, str}; // FIXME: Use &str functions for a custom Debug #[derive(Debug)] diff --git a/src/pg/protocol/ready_for_query.rs b/src/pg/protocol/ready_for_query.rs index d12046be..a351a425 100644 --- a/src/pg/protocol/ready_for_query.rs +++ b/src/pg/protocol/ready_for_query.rs @@ -1,6 +1,4 @@ use super::Decode; -use bytes::Bytes; -use std::io; #[derive(Debug, PartialEq, Clone, Copy)] #[repr(u8)] @@ -46,7 +44,6 @@ impl Decode for ReadyForQuery { #[cfg(test)] mod tests { use super::{Decode, ReadyForQuery, TransactionStatus}; - use bytes::Bytes; const READY_FOR_QUERY: &[u8] = b"E"; diff --git a/src/pg/protocol/response.rs b/src/pg/protocol/response.rs index 6e7d30a6..b86058f7 100644 --- a/src/pg/protocol/response.rs +++ b/src/pg/protocol/response.rs @@ -1,5 +1,4 @@ use super::{decode::get_str, Decode}; -use bytes::Bytes; use std::{ fmt, io, pin::Pin, @@ -68,7 +67,7 @@ impl FromStr for Severity { "LOG" => Severity::Log, _ => { - return Err(io::ErrorKind::InvalidData)?; + return Err(io::ErrorKind::InvalidData.into()); } }) } @@ -387,7 +386,6 @@ impl Decode for Response { #[cfg(test)] mod tests { use super::{Decode, Response, Severity}; - use bytes::Bytes; const RESPONSE: &[u8] = b"SNOTICE\0VNOTICE\0C42710\0Mextension \"uuid-ossp\" already exists, \ skipping\0Fextension.c\0L1656\0RCreateExtension\0\0"; diff --git a/src/pg/protocol/row_description.rs b/src/pg/protocol/row_description.rs index 5cfa9772..5eb63c41 100644 --- a/src/pg/protocol/row_description.rs +++ b/src/pg/protocol/row_description.rs @@ -1,9 +1,7 @@ use super::Decode; use byteorder::{BigEndian, ByteOrder}; -use bytes::Bytes; use memchr::memchr; use std::{ - io, mem::size_of_val, num::{NonZeroI16, NonZeroU32}, str, @@ -146,7 +144,6 @@ impl<'a> ExactSizeIterator for FieldDescriptions<'a> {} #[cfg(test)] mod tests { use super::{Decode, RowDescription}; - use bytes::Bytes; const ROW_DESC: &[u8] = b"\0\x03?column?\0\0\0\0\0\0\0\0\0\0\x17\0\x04\xff\xff\xff\xff\0\0?column?\0\0\0\0\0\0\0\0\0\0\x17\0\x04\xff\xff\xff\xff\0\0?column?\0\0\0\0\0\0\0\0\0\0\x17\0\x04\xff\xff\xff\xff\0\0"; diff --git a/src/pg/protocol/startup_message.rs b/src/pg/protocol/startup_message.rs index 2cb06657..ba64ace1 100644 --- a/src/pg/protocol/startup_message.rs +++ b/src/pg/protocol/startup_message.rs @@ -11,7 +11,7 @@ impl Encode for StartupMessage<'_> { buf.put_int_32(0); // skip over len // protocol version number (3.0) - buf.put_int_32(196608); + buf.put_int_32(196_608); for (name, value) in self.params { buf.put_str(name); diff --git a/src/pg/query.rs b/src/pg/query.rs index 855ba349..12b63488 100644 --- a/src/pg/query.rs +++ b/src/pg/query.rs @@ -1,20 +1,13 @@ use super::{ - protocol::{self, BufMut, Message}, - Pg, PgRawConnection, PgRow, + protocol::{self, BufMut}, + Pg, PgRawConnection, }; use crate::{ query::RawQuery, - row::FromRow, serialize::{IsNull, ToSql}, types::HasSqlType, }; use byteorder::{BigEndian, ByteOrder}; -use futures::{ - future::BoxFuture, - stream::{self, BoxStream}, - Stream, -}; -use std::io; pub struct PgRawQuery<'q> { limit: i32, diff --git a/src/pool.rs b/src/pool.rs index 44c279dd..e357cc83 100644 --- a/src/pool.rs +++ b/src/pool.rs @@ -1,16 +1,11 @@ use crate::{ backend::Backend, connection::RawConnection, executor::Executor, query::RawQuery, row::FromRow, - Connection, }; use crossbeam_queue::{ArrayQueue, SegQueue}; -use futures::{ - channel::oneshot, - future::BoxFuture, - stream::{self, BoxStream, Stream, StreamExt}, - TryFutureExt, -}; +use futures_channel::oneshot; +use futures_core::{future::BoxFuture, stream::BoxStream}; +use futures_util::stream::StreamExt; use std::{ - future::Future, io, ops::{Deref, DerefMut}, sync::{ @@ -19,7 +14,6 @@ use std::{ }, time::{Duration, Instant}, }; -use url::Url; pub struct PoolOptions { pub max_size: usize, @@ -48,7 +42,7 @@ where DB: Backend, { // TODO: PoolBuilder - pub fn new<'a>(url: &str, max_size: usize) -> Self { + pub fn new(url: &str, max_size: usize) -> Self { Self(Arc::new(SharedPool { url: url.to_owned(), idle: ArrayQueue::new(max_size), diff --git a/src/query.rs b/src/query.rs index 7f8296db..11b0f200 100644 --- a/src/query.rs +++ b/src/query.rs @@ -1,12 +1,11 @@ use crate::{ backend::{Backend, BackendAssocRawQuery}, executor::Executor, - pool::Pool, row::FromRow, serialize::ToSql, types::{AsSqlType, HasSqlType}, }; -use futures::{future::BoxFuture, stream::BoxStream}; +use futures_core::{future::BoxFuture, stream::BoxStream}; use std::io; pub trait RawQuery<'q>: Sized + Send + Sync {