Extract Connection::open into a new trait Connect::connect

Author: mehcode

sqlx-core/src/connection.rs

@@ -9,12 +9,6 @@ use std::convert::TryInto;
 /// Prefer running queries from [Pool] unless there is a specific need for a single, continuous
 /// connection.
 pub trait Connection: Executor + Send + 'static {
-    /// Establish a new database connection.
-    fn open<T>(url: T) -> BoxFuture<'static, crate::Result<Self>>
-    where
-        T: TryInto<Url, Error = crate::Error>,
-        Self: Sized;
-
     /// Close this database connection.
     fn close(self) -> BoxFuture<'static, crate::Result<()>>;
 
@@ -23,3 +17,14 @@ pub trait Connection: Executor + Send + 'static {
         Box::pin(self.execute("SELECT 1", Default::default()).map_ok(|_| ()))
     }
 }
+
+/// Represents a type that can directly establish a new connection.
+pub trait Connect {
+    type Connection: Connection;
+
+    /// Establish a new database connection.
+    fn connect<T>(url: T) -> BoxFuture<'static, crate::Result<Self::Connection>>
+    where
+        T: TryInto<Url, Error = crate::Error>,
+        Self: Sized;
+}

sqlx-core/src/lib.rs

@@ -43,7 +43,7 @@ pub use database::Database;
 #[doc(inline)]
 pub use error::{Error, Result};
 
-pub use connection::Connection;
+pub use connection::{Connect, Connection};
 pub use executor::Executor;
 pub use query::{query, Query};
 pub use query_as::{query_as, QueryAs};

sqlx-core/src/mysql/connection.rs

@@ -591,15 +591,30 @@ impl MySqlConnection {
     }
 }
 
-impl Connection for MySqlConnection {
-    fn open<T>(url: T) -> BoxFuture<'static, crate::Result<Self>>
+impl MySqlConnection {
+    #[deprecated(note = "please use 'connect' instead")]
+    pub fn open<T>(url: T) -> BoxFuture<'static, crate::Result<Self>>
     where
         T: TryInto<Url, Error = crate::Error>,
         Self: Sized,
     {
         Box::pin(MySqlConnection::open(url.try_into()))
     }
+}
+
+impl Connect for MySqlConnection {
+    type Connection = MySqlConnection;
+
+    fn connect<T>(url: T) -> BoxFuture<'static, Result<MySqlConnection>>
+    where
+        T: TryInto<Url, Error = crate::Error>,
+        Self: Sized,
+    {
+        Box::pin(PgConnection::open(url.try_into()))
+    }
+}
 
+impl Connection for MySqlConnection {
     fn close(self) -> BoxFuture<'static, crate::Result<()>> {
         Box::pin(self.close())
     }

sqlx-core/src/pool/executor.rs

@@ -1,34 +1,40 @@
 use futures_core::{future::BoxFuture, stream::BoxStream};
 use futures_util::StreamExt;
 
-use crate::{describe::Describe, executor::Executor, pool::Pool, Database};
+use crate::{
+    connection::{Connect, Connection},
+    describe::Describe,
+    executor::Executor,
+    pool::Pool,
+    Database,
+};
 
-impl<DB> Executor for Pool<DB>
+impl<C> Executor for Pool<C>
 where
-    DB: Database,
+    C: Connection + Connect<Connection = C>,
 {
-    type Database = DB;
+    type Database = <C as Executor>::Database;
 
     fn send<'e, 'q: 'e>(&'e mut self, commands: &'q str) -> BoxFuture<'e, crate::Result<()>> {
-        Box::pin(async move { <&Pool<DB> as Executor>::send(&mut &*self, commands).await })
+        Box::pin(async move { <&Pool<C> as Executor>::send(&mut &*self, commands).await })
     }
 
     fn execute<'e, 'q: 'e>(
         &'e mut self,
         query: &'q str,
-        args: DB::Arguments,
+        args: <<C as Executor>::Database as Database>::Arguments,
     ) -> BoxFuture<'e, crate::Result<u64>> {
-        Box::pin(async move { <&Pool<DB> as Executor>::execute(&mut &*self, query, args).await })
+        Box::pin(async move { <&Pool<C> as Executor>::execute(&mut &*self, query, args).await })
     }
 
     fn fetch<'e, 'q: 'e>(
         &'e mut self,
         query: &'q str,
-        args: DB::Arguments,
-    ) -> BoxStream<'e, crate::Result<DB::Row>> {
+        args: <<C as Executor>::Database as Database>::Arguments,
+    ) -> BoxStream<'e, crate::Result<<<C as Executor>::Database as Database>::Row>> {
         Box::pin(async_stream::try_stream! {
             let mut self_ = &*self;
-            let mut s = <&Pool<DB> as Executor>::fetch(&mut self_, query, args);
+            let mut s = <&Pool<C> as Executor>::fetch(&mut self_, query, args);
 
             while let Some(row) = s.next().await.transpose()? {
                 yield row;
@@ -39,26 +45,26 @@ where
     fn fetch_optional<'e, 'q: 'e>(
         &'e mut self,
         query: &'q str,
-        args: DB::Arguments,
-    ) -> BoxFuture<'e, crate::Result<Option<DB::Row>>> {
+        args: <<C as Executor>::Database as Database>::Arguments,
+    ) -> BoxFuture<'e, crate::Result<Option<<<C as Executor>::Database as Database>::Row>>> {
         Box::pin(
-            async move { <&Pool<DB> as Executor>::fetch_optional(&mut &*self, query, args).await },
+            async move { <&Pool<C> as Executor>::fetch_optional(&mut &*self, query, args).await },
         )
     }
 
     fn describe<'e, 'q: 'e>(
         &'e mut self,
         query: &'q str,
     ) -> BoxFuture<'e, crate::Result<Describe<Self::Database>>> {
-        Box::pin(async move { <&Pool<DB> as Executor>::describe(&mut &*self, query).await })
+        Box::pin(async move { <&Pool<C> as Executor>::describe(&mut &*self, query).await })
     }
 }
 
-impl<DB> Executor for &'_ Pool<DB>
+impl<C> Executor for &'_ Pool<C>
 where
-    DB: Database,
+    C: Connection + Connect<Connection = C>,
 {
-    type Database = DB;
+    type Database = <C as Executor>::Database;
 
     fn send<'e, 'q: 'e>(&'e mut self, commands: &'q str) -> BoxFuture<'e, crate::Result<()>> {
         Box::pin(async move { self.acquire().await?.send(commands).await })
@@ -67,16 +73,16 @@ where
     fn execute<'e, 'q: 'e>(
         &'e mut self,
         query: &'q str,
-        args: DB::Arguments,
+        args: <<C as Executor>::Database as Database>::Arguments,
     ) -> BoxFuture<'e, crate::Result<u64>> {
         Box::pin(async move { self.acquire().await?.execute(query, args).await })
     }
 
     fn fetch<'e, 'q: 'e>(
         &'e mut self,
         query: &'q str,
-        args: DB::Arguments,
-    ) -> BoxStream<'e, crate::Result<DB::Row>> {
+        args: <<C as Executor>::Database as Database>::Arguments,
+    ) -> BoxStream<'e, crate::Result<<<C as Executor>::Database as Database>::Row>> {
         Box::pin(async_stream::try_stream! {
             let mut live = self.acquire().await?;
             let mut s = live.fetch(query, args);
@@ -90,8 +96,8 @@ where
     fn fetch_optional<'e, 'q: 'e>(
         &'e mut self,
         query: &'q str,
-        args: DB::Arguments,
-    ) -> BoxFuture<'e, crate::Result<Option<DB::Row>>> {
+        args: <<C as Executor>::Database as Database>::Arguments,
+    ) -> BoxFuture<'e, crate::Result<Option<<<C as Executor>::Database as Database>::Row>>> {
         Box::pin(async move { self.acquire().await?.fetch_optional(query, args).await })
     }
 

sqlx-core/src/pool/inner.rs

@@ -8,24 +8,23 @@ use crossbeam_queue::{ArrayQueue, SegQueue};
 use futures_channel::oneshot::{channel, Sender};
 
 use super::{Idle, Live, Options};
-use crate::{error::Error, Connection, Database};
+use crate::{
+    connection::{Connect, Connection},
+    error::Error,
+};
 
-pub(super) struct SharedPool<DB>
-where
-    DB: Database,
-{
+pub(super) struct SharedPool<C> {
     url: String,
-    idle: ArrayQueue<Idle<DB>>,
-    waiters: SegQueue<Sender<Live<DB>>>,
+    idle: ArrayQueue<Idle<C>>,
+    waiters: SegQueue<Sender<Live<C>>>,
     size: AtomicU32,
     is_closed: AtomicBool,
     options: Options,
 }
 
-impl<DB> SharedPool<DB>
+impl<C> SharedPool<C>
 where
-    DB: Database,
-    DB::Connection: Connection<Database = DB>,
+    C: Connection + Connect<Connection = C>,
 {
     pub(super) async fn new_arc(url: &str, options: Options) -> crate::Result<Arc<Self>> {
         let pool = Arc::new(Self {
@@ -96,15 +95,15 @@ where
     }
 
     #[inline]
-    pub(super) fn try_acquire(&self) -> Option<Live<DB>> {
+    pub(super) fn try_acquire(&self) -> Option<Live<C>> {
         if self.is_closed.load(Ordering::Acquire) {
             return None;
         }
 
         Some(self.idle.pop().ok()?.live)
     }
 
-    pub(super) fn release(&self, mut live: Live<DB>) {
+    pub(super) fn release(&self, mut live: Live<C>) {
         // Try waiters in (FIFO) order until one is still waiting ..
         while let Ok(waiter) = self.waiters.pop() {
             live = match waiter.send(live) {
@@ -123,7 +122,7 @@ where
         });
     }
 
-    pub(super) async fn acquire(&self) -> crate::Result<Live<DB>> {
+    pub(super) async fn acquire(&self) -> crate::Result<Live<C>> {
         let start = Instant::now();
         let deadline = start + self.options.connect_timeout;
 
@@ -198,7 +197,7 @@ where
         Err(Error::PoolClosed)
     }
 
-    async fn eventually_connect(&self, deadline: Instant) -> crate::Result<Live<DB>> {
+    async fn eventually_connect(&self, deadline: Instant) -> crate::Result<Live<C>> {
         loop {
             // [connect] will raise an error when past deadline
             // [connect] returns None if its okay to retry
@@ -208,7 +207,7 @@ where
         }
     }
 
-    async fn connect(&self, deadline: Instant) -> crate::Result<Option<Live<DB>>> {
+    async fn connect(&self, deadline: Instant) -> crate::Result<Option<Live<C>>> {
         // FIXME: Code between `-` is duplicate with [acquire]
         // ---------------------------------
 
@@ -227,8 +226,8 @@ where
 
         // ---------------------------------
 
-        // result here is `Result<Result<DB, Error>, TimeoutError>`
-        match timeout(until, DB::Connection::open(&self.url)).await {
+        // result here is `Result<Result<C, Error>, TimeoutError>`
+        match timeout(until, C::connect(&self.url)).await {
             // successfully established connection
             Ok(Ok(raw)) => {
                 Ok(Some(Live {
@@ -260,18 +259,18 @@ where
     }
 }
 
-impl<DB: Database> Idle<DB>
+impl<C> Idle<C>
 where
-    DB::Connection: Connection<Database = DB>,
+    C: Connection,
 {
     async fn close(self) {
         self.live.close().await;
     }
 }
 
-impl<DB: Database> Live<DB>
+impl<C> Live<C>
 where
-    DB::Connection: Connection<Database = DB>,
+    C: Connection,
 {
     async fn close(self) {
         let _ = self.raw.close().await;
@@ -280,21 +279,24 @@ where
 
 // NOTE: Function names here are bizzare. Helpful help would be appreciated.
 
-fn is_beyond_lifetime<DB: Database>(live: &Live<DB>, options: &Options) -> bool {
+fn is_beyond_lifetime<C>(live: &Live<C>, options: &Options) -> bool {
     // check if connection was within max lifetime (or not set)
     options
         .max_lifetime
         .map_or(false, |max| live.created.elapsed() > max)
 }
 
-fn is_beyond_idle<DB: Database>(idle: &Idle<DB>, options: &Options) -> bool {
+fn is_beyond_idle<C>(idle: &Idle<C>, options: &Options) -> bool {
     // if connection wasn't idle too long (or not set)
     options
         .idle_timeout
         .map_or(false, |timeout| idle.since.elapsed() > timeout)
 }
 
-async fn check_live<DB: Database>(mut live: Live<DB>, options: &Options) -> Option<Live<DB>> {
+async fn check_live<C>(mut live: Live<C>, options: &Options) -> Option<Live<C>>
+where
+    C: Connection,
+{
     // If the connection we pulled has expired, close the connection and
     // immediately create a new connection
     if is_beyond_lifetime(&live, options) {
@@ -325,9 +327,9 @@ async fn check_live<DB: Database>(mut live: Live<DB>, options: &Options) -> Opti
 }
 
 /// if `max_lifetime` or `idle_timeout` is set, spawn a task that reaps senescent connections
-fn spawn_reaper<DB: Database>(pool: &Arc<SharedPool<DB>>)
+fn spawn_reaper<C>(pool: &Arc<SharedPool<C>>)
 where
-    DB::Connection: Connection<Database = DB>,
+    C: Connection,
 {
     let period = match (pool.options.max_lifetime, pool.options.idle_timeout) {
         (Some(it), None) | (None, Some(it)) => it,