Move HTTP client into separate async task and communicate over channels

Author: stevepryde

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "thirtyfour"
-version = "0.21.0"
+version = "0.22.0"
 authors = ["Steve Pryde <steve@stevepryde.com>"]
 edition = "2018"
 license = "MIT OR Apache-2.0"

src/session.rs

@@ -1,24 +1,65 @@
 use crate::common::command::FormatRequestData;
 use crate::common::config::WebDriverConfig;
-use crate::error::WebDriverResult;
+use crate::error::{WebDriverError, WebDriverResult};
 use crate::http::connection_async::WebDriverHttpClientAsync;
 use crate::webdrivercommands::WebDriverCommands;
-use crate::SessionId;
+use crate::{RequestData, SessionId};
 use async_trait::async_trait;
-use std::sync::Arc;
+use futures::channel::mpsc::{unbounded, UnboundedReceiver, UnboundedSender};
+use futures::channel::oneshot;
+use futures::SinkExt;
+use futures::StreamExt;
+
+#[derive(Debug)]
+pub enum SessionMessage {
+    Request(RequestData, oneshot::Sender<WebDriverResult<serde_json::Value>>),
+}
+
+pub fn spawn_session_task(
+    conn: Box<dyn WebDriverHttpClientAsync>,
+) -> UnboundedSender<SessionMessage> {
+    let (tx, rx) = unbounded();
+
+    #[cfg(feature = "async-std-runtime")]
+    {
+        async_std::task::spawn(session_runner(rx, conn));
+    }
+
+    #[cfg(all(feature = "tokio-runtime", not(feature = "async-std-runtime")))]
+    {
+        tokio::spawn(session_runner(rx, conn));
+    }
+
+    tx
+}
+
+async fn session_runner(
+    mut rx: UnboundedReceiver<SessionMessage>,
+    conn: Box<dyn WebDriverHttpClientAsync>,
+) {
+    // This will return None when the sender hangs up.
+    while let Some(msg) = rx.next().await {
+        match msg {
+            SessionMessage::Request(data, tx) => {
+                let ret = conn.execute(data).await;
+                tx.send(ret).expect("Failed to send response");
+            }
+        }
+    }
+}
 
 #[derive(Debug)]
 pub struct WebDriverSession {
     session_id: SessionId,
-    conn: Arc<dyn WebDriverHttpClientAsync>,
+    tx: UnboundedSender<SessionMessage>,
     config: WebDriverConfig,
 }
 
 impl WebDriverSession {
-    pub fn new(session_id: SessionId, conn: Arc<dyn WebDriverHttpClientAsync>) -> Self {
+    pub fn new(session_id: SessionId, tx: UnboundedSender<SessionMessage>) -> Self {
         Self {
             session_id,
-            conn,
+            tx,
             config: WebDriverConfig::new(),
         }
     }
@@ -39,7 +80,21 @@ impl WebDriverSession {
         &self,
         request: Box<dyn FormatRequestData + Send + Sync>,
     ) -> WebDriverResult<serde_json::Value> {
-        self.conn.execute(request.format_request(&self.session_id)).await
+        let (ret_tx, ret_rx) = oneshot::channel();
+        self.tx
+            .clone()
+            .send(SessionMessage::Request(request.format_request(&self.session_id), ret_tx))
+            .await
+            .map_err(|e| {
+                WebDriverError::UnknownResponse(format!("Failed to send request to server: {}", e))
+            })?;
+
+        match ret_rx.await {
+            Ok(x) => x,
+            Err(oneshot::Canceled) => Err(WebDriverError::UnknownResponse(
+                "Failed to get response from server".to_string(),
+            )),
+        }
     }
 }
 

src/support.rs

@@ -13,7 +13,7 @@ where
     rt.block_on(future)
 }
 
-#[cfg(all(feature = "async-std-runtime"))]
+#[cfg(feature = "async-std-runtime")]
 pub fn block_on<F, T>(future: F) -> WebDriverResult<T>
 where
     F: Future<Output = WebDriverResult<T>>,
@@ -26,7 +26,7 @@ pub async fn sleep(duration: std::time::Duration) {
     tokio::time::delay_for(duration).await;
 }
 
-#[cfg(all(feature = "async-std-runtime", not(feature = "tokio-runtime")))]
+#[cfg(feature = "async-std-runtime")]
 pub async fn sleep(duration: std::time::Duration) {
     async_std::task::sleep(duration).await;
 }

src/webdriver.rs

@@ -1,11 +1,8 @@
-use std::marker::PhantomData;
-use std::sync::Arc;
-
 use async_trait::async_trait;
-use futures::executor::block_on;
 use log::error;
 use serde::Serialize;
 use serde_json::Value;
+use std::marker::PhantomData;
 
 use crate::common::config::WebDriverConfig;
 use crate::http::connection_async::WebDriverHttpClientAsync;
@@ -15,11 +12,13 @@ use crate::http::nulldriver_async::NullDriverAsync;
 use crate::http::reqwest_async::ReqwestDriverAsync;
 #[cfg(feature = "async-std-runtime")]
 use crate::http::surf_async::SurfDriverAsync;
+use crate::session::spawn_session_task;
 use crate::webdrivercommands::{start_session, WebDriverCommands};
 use crate::{
     common::command::Command, error::WebDriverResult, session::WebDriverSession,
     DesiredCapabilities, SessionId,
 };
+use futures::executor::block_on;
 
 #[cfg(not(any(feature = "tokio-runtime", feature = "async-std-runtime")))]
 /// The WebDriver struct represents a browser session.
@@ -105,10 +104,12 @@ where
     where
         C: Serialize,
     {
-        let conn = Arc::new(T::create(remote_server_addr)?);
-        let (session_id, session_capabilities) = start_session(conn.clone(), capabilities).await?;
+        let conn = T::create(remote_server_addr)?;
+        let (session_id, session_capabilities) = start_session(&conn, capabilities).await?;
+        let tx = spawn_session_task(Box::new(conn));
+
         let driver = GenericWebDriver {
-            session: WebDriverSession::new(session_id, conn),
+            session: WebDriverSession::new(session_id, tx),
             capabilities: session_capabilities,
             quit_on_drop: true,
             phantom: PhantomData,

src/webdrivercommands.rs

@@ -1,6 +1,5 @@
 #[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))]
 use std::path::Path;
-use std::sync::Arc;
 use std::time::Duration;
 
 #[cfg(feature = "async-std-runtime")]
@@ -29,7 +28,7 @@ use crate::{
 /// Start a new WebDriver session, returning the session id and the
 /// capabilities JSON that was received back from the server.
 pub async fn start_session<C>(
-    conn: Arc<dyn WebDriverHttpClientAsync>,
+    conn: &dyn WebDriverHttpClientAsync,
     capabilities: C,
 ) -> WebDriverResult<(SessionId, serde_json::Value)>
 where