refactor add proper timeout on TcpStream

engineio/src/transports/websocket_secure.rs

@@ -10,7 +10,6 @@ use std::convert::TryFrom;
 use std::io::ErrorKind;
 use std::str::from_utf8;
 use std::sync::{Arc, Mutex, RwLock};
-use std::thread::sleep;
 use std::time::Duration;
 use websocket::WebSocketError;
 use websocket::{
@@ -45,6 +44,11 @@ impl WebsocketSecureTransport {
             client_builder = client_builder.custom_headers(&headers);
         }
         let client = client_builder.connect_secure(tls_config)?;
+        // set a read timeout on the client so that the connection doesn't block on receive
+        client
+            .stream_ref()
+            .get_ref()
+            .set_read_timeout(Some(Duration::from_millis(200)))?;
 
         client.set_nonblocking(false)?;
 
@@ -104,23 +108,23 @@ impl Transport for WebsocketSecureTransport {
         let received_df: DataFrame;
         loop {
             let mut receiver = self.client.lock()?;
-            receiver.set_nonblocking(true)?;
             match receiver.recv_dataframe() {
                 Ok(payload) => {
                     received_df = payload;
                     break;
                 }
                 // Special case to fix https://github.com/1c3t3a/rust-socketio/issues/133
-                // This error occures when the websocket connection is set to nonblock,
-                // but receive is called on the socket without having a current package to handle.
-                // As a result we're going to sleep for 200ms and release the lock on the client,
+                // This error occures when the websocket connection times out on the receive method.
+                // The timeout is defined on the underlying TcpStream (see `WebsocketSecureTransport::new`).
+                // The error kind is platform specific, on Unix systems this errors with `ErrorKind::WouldBlock`,
+                // on Windows with `ErrorKind::TimedOut`.
+                // As a result we're going to release the lock on the client,
                 // so that other threads (especially emit) are able to access the client.
-                Err(WebSocketError::IoError(err)) if err.kind() == ErrorKind::WouldBlock => (),
-                _ => (),
+                Err(WebSocketError::IoError(err))
+                    if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {}
+                Err(err) => return Err(err.into()),
             }
-            receiver.set_nonblocking(false)?;
             drop(receiver);
-            sleep(Duration::from_millis(200));
         }
 
         // if this is a binary payload, we mark it as a message
@@ -210,7 +214,6 @@ mod test {
     #[test]
     fn websocket_secure_debug() -> Result<()> {
         let transport = new()?;
-        transport.poll()?;
         assert_eq!(
             format!("{:?}", transport),
             format!(