Add Client::client() method, returns available client from the pool (#47)

Author: fafhrd91

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [1.8.0] - 2025-01-31
+
+* Add Client::client() method, returns available client from the pool
+
 ## [1.7.0] - 2025-01-30
 
 * Add disconnect on drop request for client

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "ntex-h2"
-version = "1.7.0"
+version = "1.8.0"
 license = "MIT OR Apache-2.0"
 authors = ["Nikolay Kim <[email protected]>"]
 description = "An HTTP/2 client and server"

src/client/pool.rs

@@ -56,7 +56,6 @@ impl Client {
         ClientBuilder::with_default(addr)
     }
 
-    #[inline]
     /// Send request to the peer
     pub async fn send(
         &self,
@@ -65,41 +64,46 @@ impl Client {
         headers: HeaderMap,
         eof: bool,
     ) -> Result<(SendStream, RecvStream), ClientError> {
+        self.client()
+            .await?
+            .send(method, path, headers, eof)
+            .await
+            .map_err(From::from)
+    }
+
+    /// Get client from the pool
+    pub async fn client(&self) -> Result<SimpleClient, ClientError> {
         loop {
             let (client, num) = self.get_client();
 
             if let Some(client) = client {
-                return client
-                    .send(method, path, headers, eof)
-                    .await
-                    .map_err(From::from);
+                return Ok(client);
+            } else {
+                self.connect(num).await?;
             }
+        }
+    }
 
-            // can create new connection
-            if !self.inner.connecting.get()
-                && (num < self.inner.maxconn
-                    || (self.inner.minconn > 0 && num < self.inner.minconn))
-            {
-                // create new connection
-                self.inner.connecting.set(true);
-
-                return self
-                    .create_connection()
-                    .await?
-                    .send(method, path, headers, eof)
-                    .await
-                    .map_err(From::from);
-            } else {
-                log::debug!(
-                    "New connection is being established {:?} or number of existing cons {} greater than allowed {}",
-                    self.inner.connecting.get(), num, self.inner.maxconn);
+    async fn connect(&self, num: usize) -> Result<(), ClientError> {
+        // can create new connection
+        if !self.inner.connecting.get()
+            && (num < self.inner.maxconn || (self.inner.minconn > 0 && num < self.inner.minconn))
+        {
+            // create new connection
+            self.inner.connecting.set(true);
 
-                // wait for available connection
-                let (tx, rx) = oneshot::channel();
-                self.waiters.borrow_mut().push_back(tx);
-                let _ = rx.await;
-            }
+            self.create_connection().await?;
+        } else {
+            log::debug!(
+                "New connection is being established {:?} or number of existing cons {} greater than allowed {}",
+                self.inner.connecting.get(), num, self.inner.maxconn);
+
+            // wait for available connection
+            let (tx, rx) = oneshot::channel();
+            self.waiters.borrow_mut().push_back(tx);
+            let _ = rx.await?;
         }
+        Ok(())
     }
 
     fn get_client(&self) -> (Option<SimpleClient>, usize) {
@@ -113,9 +117,10 @@ impl Client {
             } else if connections[idx].is_disconnecting() {
                 let con = connections.remove(idx);
                 let timeout = self.inner.disconnect_timeout;
-                ntex_util::spawn(async move {
+                let f = ntex_util::spawn(async move {
                     let _ = con.disconnect().disconnect_timeout(timeout).await;
                 });
+                drop(f);
             } else {
                 idx += 1;
             }
@@ -151,7 +156,7 @@ impl Client {
         }
     }
 
-    async fn create_connection(&self) -> Result<SimpleClient, ClientError> {
+    async fn create_connection(&self) -> Result<(), ClientError> {
         let (tx, rx) = oneshot::channel();
 
         let inner = self.inner.clone();
@@ -173,11 +178,11 @@ impl Client {
                         inner.authority.clone(),
                         storage,
                     );
-                    inner.connections.borrow_mut().push(client.clone());
+                    inner.connections.borrow_mut().push(client);
                     inner
                         .total_connections
                         .set(inner.total_connections.get() + 1);
-                    Ok(client)
+                    Ok(())
                 }
                 Ok(Err(err)) => Err(ClientError::from(err)),
                 Err(_) => Err(ClientError::HandshakeTimeout),

src/dispatcher.rs

@@ -136,11 +136,12 @@ where
         if let Err(e) = self.inner.publish.poll(cx) {
             let inner = self.inner.clone();
             let con = self.connection.connection();
-            ntex_util::spawn(async move {
+            let f = ntex_util::spawn(async move {
                 if inner.control.call_nowait(Control::error(e)).await.is_ok() {
                     con.close();
                 }
             });
+            drop(f);
         }
         self.inner.control.poll(cx).map_err(|_| ())
     }