remove all usage of explict pipeline

Author: fakeshadow

frameworks/Rust/xitca-web/Cargo.lock

@@ -95,17 +95,17 @@ codegen-units = 1
 panic = "abort"
 
 [patch.crates-io]
-xitca-postgres-diesel = { git = "https://github.com/fakeshadow/xitca-postgres-diesel", rev = "ffd222a" }
+xitca-postgres-diesel = { git = "https://github.com/fakeshadow/xitca-postgres-diesel", rev = "fb5dcba" }
 xitca-postgres-toasty = { git = "https://github.com/fakeshadow/xitca-postgres-toasty", rev = "04bedb8" }
 
 # personal fork for efficient toasty engine fine tuned with pipelined xitca-postgres client
 toasty = { git = "https://github.com/fakeshadow/toasty", branch = "engine" }
 toasty-core = { git = "https://github.com/fakeshadow/toasty", branch = "engine" }
 toasty-sql = { git = "https://github.com/fakeshadow/toasty", branch = "engine" }
 
-xitca-codegen = { git = "http://github.com/HFQR/xitca-web", rev = "91c6c31" }
-xitca-http = { git = "http://github.com/HFQR/xitca-web", rev = "91c6c31" }
-xitca-postgres = { git = "http://github.com/HFQR/xitca-web", rev = "91c6c31" }
-xitca-server = { git = "http://github.com/HFQR/xitca-web", rev = "91c6c31" }
-xitca-service = { git = "http://github.com/HFQR/xitca-web", rev = "91c6c31" }
-xitca-web = { git = "http://github.com/HFQR/xitca-web", rev = "91c6c31" }
+xitca-codegen = { git = "http://github.com/HFQR/xitca-web", rev = "244a06d" }
+xitca-http = { git = "http://github.com/HFQR/xitca-web", rev = "244a06d" }
+xitca-postgres = { git = "http://github.com/HFQR/xitca-web", rev = "244a06d" }
+xitca-server = { git = "http://github.com/HFQR/xitca-web", rev = "244a06d" }
+xitca-service = { git = "http://github.com/HFQR/xitca-web", rev = "244a06d" }
+xitca-web = { git = "http://github.com/HFQR/xitca-web", rev = "244a06d" }

frameworks/Rust/xitca-web/Cargo.toml

@@ -1,10 +1,7 @@
 #[path = "./db_util.rs"]
 mod db_util;
 
-use core::cell::RefCell;
-
-use xitca_io::bytes::BytesMut;
-use xitca_postgres::{Execute, iter::AsyncLendingIterator, pipeline::Pipeline, pool::Pool};
+use xitca_postgres::{Execute, iter::AsyncLendingIterator, pool::Pool};
 
 use super::{
     ser::{Fortune, Fortunes, World},
@@ -15,21 +12,21 @@ use db_util::{FORTUNE_STMT, UPDATE_STMT, WORLD_STMT, not_found};
 
 pub struct Client {
     pool: Pool,
-    shared: RefCell<(Rand, BytesMut)>,
+    rng: core::cell::RefCell<Rand>,
 }
 
 pub async fn create() -> HandleResult<Client> {
     Ok(Client {
         pool: Pool::builder(DB_URL).capacity(1).build()?,
-        shared: Default::default(),
+        rng: Default::default(),
     })
 }
 
 impl Client {
     pub async fn get_world(&self) -> HandleResult<World> {
         let mut conn = self.pool.get().await?;
         let stmt = WORLD_STMT.execute(&mut conn).await?;
-        let id = self.shared.borrow_mut().0.gen_id();
+        let id = self.rng.borrow_mut().gen_id();
         let mut res = stmt.bind([id]).query(&conn.consume()).await?;
         let row = res.try_next().await?.ok_or_else(not_found)?;
         Ok(World::new(row.get(0), row.get(1)))
@@ -39,19 +36,21 @@ impl Client {
         let mut conn = self.pool.get().await?;
         let stmt = WORLD_STMT.execute(&mut conn).await?;
 
-        let mut res = {
-            let (ref mut rng, ref mut buf) = *self.shared.borrow_mut();
-            let mut pipe = Pipeline::with_capacity_from_buf(num as _, buf);
-            rng.gen_multi()
-                .take(num as _)
-                .try_for_each(|id| stmt.bind([id]).query(&mut pipe))?;
-            pipe.query(&conn.consume())?
-        };
+        let get = self
+            .rng
+            .borrow_mut()
+            .gen_multi()
+            .take(num as _)
+            .map(|id| stmt.bind([id]).query(&conn))
+            .collect::<Vec<_>>();
+
+        drop(conn);
 
         let mut worlds = Vec::with_capacity(num as _);
 
-        while let Some(mut item) = res.try_next().await? {
-            let row = item.try_next().await?.ok_or_else(not_found)?;
+        for get in get {
+            let mut res = get.await?;
+            let row = res.try_next().await?.ok_or_else(not_found)?;
             worlds.push(World::new(row.get(0), row.get(1)));
         }
 
@@ -63,32 +62,32 @@ impl Client {
         let world_stmt = WORLD_STMT.execute(&mut conn).await?;
         let update_stmt = UPDATE_STMT.execute(&mut conn).await?;
 
-        let (mut res, worlds) = {
-            let (ref mut rng, ref mut buf) = *self.shared.borrow_mut();
-            let mut pipe = Pipeline::with_capacity_from_buf((num + 1) as _, buf);
-
+        let (get, update, worlds) = {
+            let mut rng = self.rng.borrow_mut();
             let mut ids = rng.gen_multi().take(num as _).collect::<Vec<_>>();
             ids.sort();
 
-            let (rngs, worlds) = ids
+            let (get, rngs, worlds) = ids
                 .iter()
                 .cloned()
                 .zip(rng.gen_multi())
                 .map(|(id, rand)| {
-                    world_stmt.bind([id]).query(&mut pipe)?;
-                    HandleResult::Ok((rand, World::new(id, rand)))
+                    let get = world_stmt.bind([id]).query(&conn);
+                    (get, rand, World::new(id, rand))
                 })
-                .collect::<HandleResult<(Vec<_>, Vec<_>)>>()?;
-            update_stmt.bind([&ids, &rngs]).query(&mut pipe)?;
-            (pipe.query(&conn.consume())?, worlds)
+                .collect::<(Vec<_>, Vec<_>, Vec<_>)>();
+
+            let update = update_stmt.bind([&ids, &rngs]).query(&conn.consume());
+
+            (get, update, worlds)
         };
 
-        while let Some(mut item) = res.try_next().await? {
-            while let Some(row) = item.try_next().await? {
-                let _rand = row.get::<i32>(1);
-            }
+        for fut in get {
+            let _rand = fut.await?.try_next().await?.ok_or_else(not_found)?.get::<i32>(1);
         }
 
+        update.await?;
+
         Ok(worlds)
     }