Use `Weak` for PooledConnection's Pool ref? about bb8 HOT 12 CLOSED

impl<M: crate::ManageConnection> Pool<M> {
    pub fn new(
        builder: bb8::Builder<ManageConnectionContainer<M>>,
        manager: M,
    ) -> Result<Self, BuildError<M::Error>> {
        use futures::future::{Future, TryFutureExt};

        let (tx, mut rx) = mpsc::unbounded_channel::<
            oneshot::Sender<
                Result<
                    bb8::PooledConnection<ManageConnectionContainer<M>>,
                    bb8::RunError<M::Error>,
                >,
            >,
        >();
        let (tx_b, rx_b) = oneshot::channel::<Result<(), BuildError<M::Error>>>();

        let handle = std::thread::Builder::new()
            .name("bb8-sync-runtime".into())
            .spawn(move || {
                let mut rt = match runtime::Builder::new()
                    .basic_scheduler()
                    .enable_all()
                    .build()
                    .map_err(BuildError::TokioIo)
                {
                    Ok(rt) => rt,
                    Err(e) => {
                        tx_b.send(Err(e)).unwrap();
                        return;
                    }
                };

                rt.block_on(async move {
                    let pool = match builder
                        .build(ManageConnectionContainer(manager))
                        .map_err(BuildError::ManagedConnectionError)
                        .await
                    {
                        Ok(pool) => pool,
                        Err(e) => {
                            tx_b.send(Err(e)).unwrap();
                            return;
                        }
                    };

                    tx_b.send(Ok(())).unwrap();

                    while let Some(mut tx_get) = rx.recv().await {
                        let pool = pool.clone();
                        tokio::spawn(async move {
                            use std::task::Poll;

                            let fut = pool.get();

                            futures::pin_mut!(fut);

                            let res =
                                futures::future::poll_fn(|ctx| match fut.as_mut().poll(ctx) {
                                    Poll::Ready(val) => Poll::Ready(Some(val)),
                                    Poll::Pending => {
                                        futures::ready!(tx_get.poll_closed(ctx));
                                        Poll::Ready(None)
                                    }
                                })
                                .await;

                            if let Some(res) = res {
                                let _ = tx_get.send(res);
                            }
                        });
                    }
                });
            })
            .unwrap();

        futures::executor::block_on(async move { rx_b.await.unwrap() })?;

        Ok(Pool { tx, handle })
    }

    pub fn get(
        &self,
    ) -> Result<bb8::PooledConnection<ManageConnectionContainer<M>>, bb8::RunError<M::Error>> {
        let (tx, rx) = oneshot::channel();

        self.tx.send(tx).ok().unwrap();

        futures::executor::block_on(async move { rx.await.unwrap() })
    }
}

#[async_trait]
pub trait FromRequest {
    async fn from_request(req: &mut http::Request<hyper::Body>) -> Result<Self, Error>;
}