handle lack of TCP connection from phone

Before this commit described situation was possible, leading to stuck
and not trying to reconnect.

The phone is mainly connecting in several seconds, once I had a 16 secs
measured time, so set this timeout to 30 secs for now including a marigin.

Author: manio

src/bluetooth.rs

@@ -8,9 +8,11 @@ use bluer::{
 };
 use futures::StreamExt;
 use simplelog::*;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
+use tokio::sync::Notify;
 use tokio::task::JoinHandle;
 use tokio::time::sleep;
 use tokio::time::timeout;
@@ -302,6 +304,7 @@ pub async fn bluetooth_stop(state: BluetoothState) -> Result<()> {
 pub async fn bluetooth_setup_connection(
     advertise: bool,
     connect: Option<Address>,
+    tcp_start: Arc<Notify>,
 ) -> Result<BluetoothState> {
     use WifiInfoResponse::WifiInfoResponse;
     use WifiStartRequest::WifiStartRequest;
@@ -326,6 +329,7 @@ pub async fn bluetooth_setup_connection(
     info.set_security_mode(SecurityMode::WPA2_PERSONAL);
     info.set_access_point_type(AccessPointType::DYNAMIC);
     send_message(&mut stream, MessageId::WifiInfoResponse, info).await?;
+    tcp_start.notify_one();
     read_message(&mut stream, MessageId::WifiStartResponse).await?;
     read_message(&mut stream, MessageId::WifiConnectStatus).await?;
     let _ = stream.shutdown().await?;

src/io_uring.rs

@@ -14,6 +14,7 @@ const NAME: &str = "<i><bright-black> proxy: </>";
 const USB_ACCESSORY_PATH: &str = "/dev/usb_accessory";
 const BUFFER_LEN: usize = 16 * 1024;
 const READ_TIMEOUT: Duration = Duration::new(5, 0);
+const TCP_CLIENT_TIMEOUT: Duration = Duration::new(30, 0);
 
 async fn copy_file_to_stream(
     from: Rc<tokio_uring::fs::File>,
@@ -152,6 +153,7 @@ async fn copy_stream_to_file(
 pub async fn io_loop(
     stats_interval: Option<Duration>,
     need_restart: Arc<Notify>,
+    tcp_start: Arc<Notify>,
 ) -> Result<(), Box<dyn std::error::Error>> {
     info!("{} 🛰️ Starting TCP server...", NAME);
     let bind_addr = format!("0.0.0.0:{}", TCP_SERVER_PORT).parse().unwrap();
@@ -161,8 +163,23 @@ pub async fn io_loop(
         NAME, bind_addr
     );
     loop {
+        // wait for bluetooth handshake
+        tcp_start.notified().await;
+
         // Asynchronously wait for an inbound TCP connection
-        let (stream, addr) = listener.accept().await?;
+        let retval = listener.accept();
+        let (stream, addr) = match timeout(TCP_CLIENT_TIMEOUT, retval).await? {
+            Ok((stream, addr)) => (stream, addr),
+            Err(_) => {
+                error!(
+                    "{} 📵 TCP server: timed out waiting for phone connection, restarting...",
+                    NAME
+                );
+                // notify main loop to restart
+                need_restart.notify_one();
+                continue;
+            }
+        };
         info!(
             "{} 📳 TCP server: new client connected: <b>{:?}</b>",
             NAME, addr

src/main.rs

@@ -106,6 +106,7 @@ async fn tokio_main(
     legacy: bool,
     connect: Option<Address>,
     need_restart: Arc<Notify>,
+    tcp_start: Arc<Notify>,
 ) {
     let accessory_started = Arc::new(Notify::new());
     let accessory_started_cloned = accessory_started.clone();
@@ -122,7 +123,7 @@ async fn tokio_main(
         }
 
         loop {
-            match bluetooth_setup_connection(advertise, connect).await {
+            match bluetooth_setup_connection(advertise, connect, tcp_start.clone()).await {
                 Ok(state) => {
                     // we're ready, gracefully shutdown bluetooth in task
                     tokio::spawn(async move { bluetooth_stop(state).await });
@@ -143,7 +144,10 @@ async fn tokio_main(
         need_restart.notified().await;
 
         // TODO: make proper main loop with cancelation
-        info!("{} 📵 TCP/USB connection closed, trying again...", NAME);
+        info!(
+            "{} 📵 TCP/USB connection closed or not started, trying again...",
+            NAME
+        );
         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
     }
 }
@@ -183,13 +187,26 @@ fn main() {
     // notify for syncing threads
     let need_restart = Arc::new(Notify::new());
     let need_restart_cloned = need_restart.clone();
+    let tcp_start = Arc::new(Notify::new());
+    let tcp_start_cloned = tcp_start.clone();
 
     // build and spawn main tokio runtime
     let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
     runtime.spawn(async move {
-        tokio_main(args.advertise, args.legacy, args.connect, need_restart).await
+        tokio_main(
+            args.advertise,
+            args.legacy,
+            args.connect,
+            need_restart,
+            tcp_start,
+        )
+        .await
     });
 
     // start tokio_uring runtime simultaneously
-    let _ = tokio_uring::start(io_loop(stats_interval, need_restart_cloned));
+    let _ = tokio_uring::start(io_loop(
+        stats_interval,
+        need_restart_cloned,
+        tcp_start_cloned,
+    ));
 }