Replace hand written future impls

This commit replaces the hand written future impls with such ones that
wrap the types from futures-util. This hopefully reduces the complexity
and also allows using the well tested types instead of hand written
ones.

I also removed the dependency on pin-project-light in favour of two
lines of unsafe code as pin-project would do the same thing internally
as well and these two instances are trivial to reason about.

Author: weiznich

Cargo.toml

@@ -36,7 +36,6 @@ deadpool = { version = "0.12", optional = true, default-features = false, featur
 ] }
 mobc = { version = ">=0.7,<0.10", optional = true }
 scoped-futures = { version = "0.1", features = ["std"] }
-pin-project-lite = "0.2.16"
 
 [dependencies.diesel]
 version = "~2.3.0"

src/pg/mod.rs

@@ -1175,22 +1175,22 @@ mod tests {
 
         async fn fn37(
             mut conn: &AsyncPgConnection,
-        ) -> QueryResult<(usize, Vec<i32>, i32, Vec<i32>, i32)> {
+        ) -> QueryResult<(usize, (Vec<i32>, (i32, (Vec<i32>, i32))))> {
             let f3 = diesel::select(0_i32.into_sql::<Integer>()).execute(&mut conn);
             let f4 = diesel::select(4_i32.into_sql::<Integer>()).load::<i32>(&mut conn);
             let f5 = diesel::select(5_i32.into_sql::<Integer>()).get_result::<i32>(&mut conn);
             let f6 = diesel::select(6_i32.into_sql::<Integer>()).get_results::<i32>(&mut conn);
             let f7 = diesel::select(7_i32.into_sql::<Integer>()).first::<i32>(&mut conn);
 
-            try_join!(f3, f4, f5, f6, f7)
+            try_join(f3, try_join(f4, try_join(f5, try_join(f6, f7)))).await
         }
 
         conn.transaction(|conn| {
             async move {
                 let f12 = fn12(conn);
                 let f37 = fn37(conn);
 
-                let ((r1, r2), (r3, r4, r5, r6, r7)) = try_join!(f12, f37).unwrap();
+                let ((r1, r2), (r3, (r4, (r5, (r6, r7))))) = try_join(f12, f37).await.unwrap();
 
                 assert_eq!(r1, 1);
                 assert_eq!(r2, 2);

src/run_query_dsl/mod.rs

@@ -99,7 +99,7 @@ pub mod methods {
         ST: 'static,
     {
         type LoadFuture<'conn>
-            = utils::Map<Conn::LoadFuture<'conn, 'query>, Self::Stream<'conn>>
+            = utils::MapOk<Conn::LoadFuture<'conn, 'query>, Self::Stream<'conn>>
         where
             Conn: 'conn;
 
@@ -112,10 +112,10 @@ pub mod methods {
             Conn: 'conn;
 
         fn internal_load(self, conn: &mut Conn) -> Self::LoadFuture<'_> {
-            utils::Map::new(conn.load(self), |stream| {
-                Ok(stream?.map(|row| {
+            utils::MapOk::new(conn.load(self), |stream| {
+                stream.map(|row| {
                     U::build_from_row(&row?).map_err(diesel::result::Error::DeserializationError)
-                }))
+                })
             })
         }
     }
@@ -313,7 +313,7 @@ pub trait RunQueryDsl<Conn>: Sized {
         Conn: AsyncConnectionCore,
         Self: methods::LoadQuery<'query, Conn, U> + 'query,
     {
-        utils::AndThen::new(self.internal_load(conn), |stream| Ok(stream?.try_collect()))
+        utils::AndThen::new(self.internal_load(conn), |stream| stream.try_collect())
     }
 
     /// Executes the given query, returning a [`Stream`] with the returned rows.
@@ -509,7 +509,7 @@ pub trait RunQueryDsl<Conn>: Sized {
         Self: methods::LoadQuery<'query, Conn, U> + 'query,
     {
         utils::AndThen::new(self.internal_load(conn), |stream| {
-            Ok(utils::LoadNext::new(Box::pin(stream?)))
+            utils::LoadNext::new(Box::pin(stream))
         })
     }
 

src/run_query_dsl/utils.rs

@@ -4,95 +4,77 @@ use std::task::{Context, Poll};
 
 use diesel::QueryResult;
 use futures_core::{ready, TryFuture, TryStream};
-use futures_util::TryStreamExt;
-use pin_project_lite::pin_project;
+use futures_util::{TryFutureExt, TryStreamExt};
 
-pin_project! {
-    /// Reimplementation of [`futures_util::future::Map`] without the generic closure argument
-    #[project = MapProj]
-    #[project_replace = MapProjReplace]
-    pub enum Map<Fut: Future, T> {
-         Incomplete {
-            #[pin]
-            future: Fut,
-            f: fn(Fut::Output) -> QueryResult<T>,
-        },
-        Complete,
-    }
+// We use a custom future implementation here to erase some lifetimes
+// that otherwise need to be specified explicitly
+//
+// Specifying these lifetimes results in the compiler not beeing
+// able to look through the generic code and emit
+// lifetime erros for pipelined queries. See
+// https://github.com/weiznich/diesel_async/issues/249 for more context
+#[repr(transparent)]
+pub struct MapOk<F: TryFutureExt, T> {
+    future: futures_util::future::MapOk<F, fn(F::Ok) -> T>,
 }
 
-impl<Fut: Future, T> Map<Fut, T> {
-    pub(crate) fn new(future: Fut, f: fn(Fut::Output) -> QueryResult<T>) -> Self {
-        Self::Incomplete { future, f }
+impl<F, T> Future for MapOk<F, T>
+where
+    F: TryFuture,
+    futures_util::future::MapOk<F, fn(F::Ok) -> T>: Future<Output = Result<T, F::Error>>,
+{
+    type Output = Result<T, F::Error>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+        unsafe {
+            // SAFETY: This projects pinning to the only inner field, so it
+            // should be safe
+            self.map_unchecked_mut(|s| &mut s.future)
+        }
+        .poll(cx)
     }
 }
 
-impl<Fut: Future, T> Future for Map<Fut, T> {
-    type Output = QueryResult<T>;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<QueryResult<T>> {
-        match self.as_mut().project() {
-            MapProj::Incomplete { future, .. } => {
-                let output = ready!(future.poll(cx));
-                match self.as_mut().project_replace(Map::Complete) {
-                    MapProjReplace::Incomplete { f, .. } => Poll::Ready(f(output)),
-                    MapProjReplace::Complete => unreachable!(),
-                }
-            }
-            MapProj::Complete => panic!("Map polled after completion"),
+impl<Fut: TryFutureExt, T> MapOk<Fut, T> {
+    pub(crate) fn new(future: Fut, f: fn(Fut::Ok) -> T) -> Self {
+        Self {
+            future: future.map_ok(f),
         }
     }
 }
 
-pin_project! {
-    /// Reimplementation of [`futures_util::future::AndThen`] without the generic closure argument
-    #[project = AndThenProj]
-    pub enum AndThen<Fut1: Future, Fut2> {
-        First {
-            #[pin]
-            future1: Map<Fut1, Fut2>,
-        },
-        Second {
-            #[pin]
-            future2: Fut2,
-        },
-        Empty,
-    }
+// similar to `MapOk` above this mainly exists to hide the lifetime
+#[repr(transparent)]
+pub struct AndThen<F1: TryFuture, F2> {
+    future: futures_util::future::AndThen<F1, F2, fn(F1::Ok) -> F2>,
 }
 
-impl<Fut1: Future, Fut2> AndThen<Fut1, Fut2> {
-    pub(crate) fn new(fut1: Fut1, f: fn(Fut1::Output) -> QueryResult<Fut2>) -> AndThen<Fut1, Fut2> {
-        Self::First {
-            future1: Map::new(fut1, f),
+impl<Fut1, Fut2> AndThen<Fut1, Fut2>
+where
+    Fut1: TryFuture,
+    Fut2: TryFuture<Error = Fut1::Error>,
+{
+    pub(crate) fn new(fut1: Fut1, f: fn(Fut1::Ok) -> Fut2) -> AndThen<Fut1, Fut2> {
+        Self {
+            future: fut1.and_then(f),
         }
     }
 }
 
-impl<Fut1, Fut2> Future for AndThen<Fut1, Fut2>
+impl<F1, F2> Future for AndThen<F1, F2>
 where
-    Fut1: TryFuture<Error = diesel::result::Error>,
-    Fut2: TryFuture<Error = diesel::result::Error>,
+    F1: TryFuture,
+    F2: TryFuture<Error = F1::Error>,
 {
-    type Output = QueryResult<Fut2::Ok>;
+    type Output = Result<F2::Ok, F2::Error>;
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        loop {
-            match self.as_mut().project() {
-                AndThenProj::First { future1 } => match ready!(future1.try_poll(cx)) {
-                    Ok(future2) => self.set(Self::Second { future2 }),
-                    Err(error) => {
-                        self.set(Self::Empty);
-                        break Poll::Ready(Err(error));
-                    }
-                },
-                AndThenProj::Second { future2 } => {
-                    let output = ready!(future2.try_poll(cx));
-                    self.set(Self::Empty);
-                    break Poll::Ready(output);
-                }
-                AndThenProj::Empty => panic!("AndThen polled after completion"),
-            }
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        unsafe {
+            // SAFETY: This projects pinning to the only inner field, so it
+            // should be safe
+            self.map_unchecked_mut(|s| &mut s.future)
         }
+        .poll(cx)
     }
 }