Implement quick_reconnect

This switch makes the Bluetooth connection persistent across HU restarts
when possible, speeding up reconnections because Bluetooth and Wi-Fi
are effectively not reconnected (same Wi-Fi parameters are reused).

Author: manio

src/bluetooth.rs

@@ -11,10 +11,13 @@ use bluer::{
 };
 use futures::StreamExt;
 use simplelog::*;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::io::AsyncReadExt;
 use tokio::io::AsyncWriteExt;
+use tokio::sync::broadcast::Receiver as BroadcastReceiver;
 use tokio::sync::Notify;
 use tokio::time::timeout;
 
@@ -563,6 +566,9 @@ impl Bluetooth {
         tcp_start: Arc<Notify>,
         bt_timeout: Duration,
         stopped: bool,
+        quick_reconnect: bool,
+        mut need_restart: BroadcastReceiver<()>,
+        profile_connected: Arc<AtomicBool>,
     ) -> Result<()> {
         // Use the provided session and adapter instead of creating new ones
         let (address, mut stream) = self
@@ -571,10 +577,40 @@ impl Bluetooth {
         Self::send_params(wifi_config.clone(), &mut stream).await?;
         tcp_start.notify_one();
 
-        // handshake complete, now disconnect the device so it should
-        // connect to real HU for calls
-        let device = self.adapter.device(bluer::Address(*address))?;
-        let _ = device.disconnect().await;
+        if quick_reconnect {
+            // keep the bluetooth profile connection alive
+            // and use it in a loop to restart handshake when necessary
+            let _ = Some(tokio::spawn(async move {
+                profile_connected.store(true, Ordering::Relaxed);
+                loop {
+                    // wait for restart notification from main loop (eg when HU disconnected)
+                    let _ = need_restart.recv().await;
+
+                    // now restart handshake with the same params
+                    match Self::send_params(wifi_config.clone(), &mut stream).await {
+                        Ok(_) => {
+                            tcp_start.notify_one();
+                            continue;
+                        }
+                        Err(e) => {
+                            error!(
+                                "{} handshake restart error: {}, doing full restart!",
+                                NAME, e
+                            );
+                            // this break should end this task
+                            break;
+                        }
+                    }
+                }
+                // we are now disconnected, redo bluetooth connection
+                profile_connected.store(false, Ordering::Relaxed);
+            }));
+        } else {
+            // handshake complete, now disconnect the device so it should
+            // connect to real HU for calls
+            let device = self.adapter.device(bluer::Address(*address))?;
+            let _ = device.disconnect().await;
+        }
 
         info!("{} 🚀 Bluetooth launch sequence completed", NAME);
 

src/io_uring.rs

@@ -4,11 +4,12 @@ use simplelog::*;
 use std::cell::RefCell;
 use std::marker::PhantomData;
 use std::rc::Rc;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::io::copy_bidirectional;
 use tokio::net::TcpStream as TokioTcpStream;
+use tokio::sync::broadcast::Sender as BroadcastSender;
 use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::sync::{mpsc, Mutex, Notify};
 use tokio::task::JoinHandle;
@@ -284,11 +285,12 @@ async fn tcp_wait_for_connection(listener: &mut TcpListener) -> Result<TcpStream
 }
 
 pub async fn io_loop(
-    need_restart: Arc<Notify>,
+    need_restart: BroadcastSender<()>,
     tcp_start: Arc<Notify>,
     config: SharedConfig,
     tx: Arc<Mutex<Option<Sender<Packet>>>>,
     sensor_channel: Arc<Mutex<Option<u8>>>,
+    profile_connected: Arc<AtomicBool>,
 ) -> Result<()> {
     let shared_config = config.clone();
     #[allow(unused_variables)]
@@ -331,7 +333,9 @@ pub async fn io_loop(
                 Err(e) => {
                     error!("{} 🔴 Enabling Android Auto: {}", NAME, e);
                     // notify main loop to restart
-                    need_restart.notify_one();
+                    if !profile_connected.load(Ordering::Relaxed) {
+                        let _ = need_restart.send(());
+                    }
                     continue;
                 }
                 Ok(s) => {
@@ -350,7 +354,9 @@ pub async fn io_loop(
                 md_tcp = Some(s);
             } else {
                 // notify main loop to restart
-                need_restart.notify_one();
+                if !profile_connected.load(Ordering::Relaxed) {
+                    let _ = need_restart.send(());
+                }
                 continue;
             }
         }
@@ -364,7 +370,9 @@ pub async fn io_loop(
                 hu_tcp = Some(s);
             } else {
                 // notify main loop to restart
-                need_restart.notify_one();
+                if !profile_connected.load(Ordering::Relaxed) {
+                    let _ = need_restart.send(());
+                }
                 continue;
             }
         } else {
@@ -383,7 +391,9 @@ pub async fn io_loop(
                 Err(e) => {
                     error!("{} 🔴 Error opening USB accessory: {}", NAME, e);
                     // notify main loop to restart
-                    need_restart.notify_one();
+                    if !profile_connected.load(Ordering::Relaxed) {
+                        let _ = need_restart.send(());
+                    }
                     continue;
                 }
             }
@@ -543,7 +553,7 @@ pub async fn io_loop(
             format_duration(started.elapsed()).to_string()
         );
         // stream(s) closed, notify main loop to restart
-        need_restart.notify_one();
+        let _ = need_restart.send(());
     }
 
     #[allow(unreachable_code)]

src/main.rs

@@ -19,10 +19,14 @@ use std::fs;
 use std::fs::OpenOptions;
 use std::path::PathBuf;
 use std::process::Command;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::thread;
 use std::time::Duration;
 use tokio::runtime::Builder;
+use tokio::sync::broadcast;
+use tokio::sync::broadcast::Sender as BroadcastSender;
 use tokio::sync::mpsc::Sender;
 use tokio::sync::Mutex;
 use tokio::sync::Notify;
@@ -179,12 +183,13 @@ async fn action_handler(config: &mut SharedConfig) {
 async fn tokio_main(
     config: SharedConfig,
     config_json: SharedConfigJson,
-    need_restart: Arc<Notify>,
+    restart_tx: BroadcastSender<()>,
     tcp_start: Arc<Notify>,
     config_file: PathBuf,
     tx: Arc<Mutex<Option<Sender<Packet>>>>,
     sensor_channel: Arc<Mutex<Option<u8>>>,
     led_support: bool,
+    profile_connected: Arc<AtomicBool>,
 ) -> Result<()> {
     let accessory_started = Arc::new(Notify::new());
     let accessory_started_cloned = accessory_started.clone();
@@ -275,6 +280,7 @@ async fn tokio_main(
 
     // main connection loop
     let change_usb_order = cfg.change_usb_order;
+    let mut need_restart = restart_tx.subscribe();
     loop {
         if let Some(ref mut leds) = led_manager {
             leds.set_led(LedColor::Green, LedMode::Heartbeat).await;
@@ -289,21 +295,27 @@ async fn tokio_main(
             enable_usb_if_present(&mut usb, accessory_started.clone()).await;
         }
 
-        // bluetooth handshake
-        if let Err(e) = bluetooth
-            .aa_handshake(
-                cfg.dongle_mode,
-                cfg.connect.clone(),
-                wifi_conf.clone().unwrap(),
-                tcp_start.clone(),
-                Duration::from_secs(cfg.bt_timeout_secs.into()),
-                cfg.action_requested == Some(Action::Stop),
-            )
-            .await
-        {
-            error!("{} bluetooth AA handshake error: {}", NAME, e);
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-            continue;
+        // run only if not handling this in handshake task
+        if !(cfg.quick_reconnect && profile_connected.load(Ordering::Relaxed)) {
+            // bluetooth handshake
+            if let Err(e) = bluetooth
+                .aa_handshake(
+                    cfg.dongle_mode,
+                    cfg.connect.clone(),
+                    wifi_conf.clone().unwrap(),
+                    tcp_start.clone(),
+                    Duration::from_secs(cfg.bt_timeout_secs.into()),
+                    cfg.action_requested == Some(Action::Stop),
+                    cfg.quick_reconnect,
+                    restart_tx.subscribe(),
+                    profile_connected.clone(),
+                )
+                .await
+            {
+                error!("{} bluetooth AA handshake error: {}", NAME, e);
+                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                continue;
+            }
         }
 
         if !change_usb_order {
@@ -315,14 +327,21 @@ async fn tokio_main(
             leds.set_led(LedColor::Blue, LedMode::On).await;
         }
         // wait for restart notification
-        need_restart.notified().await;
+        let _ = need_restart.recv().await;
+        if !(cfg.quick_reconnect && profile_connected.load(Ordering::Relaxed)) {
+            info!(
+                "{} 📵 TCP/USB connection closed or not started, trying again...",
+                NAME
+            );
+            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+        } else {
+            info!(
+                "{} 📵 TCP/USB connection closed or not started, quick restart...",
+                NAME
+            );
+        }
 
         // TODO: make proper main loop with cancelation
-        info!(
-            "{} 📵 TCP/USB connection closed or not started, trying again...",
-            NAME
-        );
-        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
         // re-read config
         cfg = config.read().await.clone();
     }
@@ -525,8 +544,7 @@ fn main() -> Result<()> {
     }
 
     // notify for syncing threads
-    let need_restart = Arc::new(Notify::new());
-    let need_restart_cloned = need_restart.clone();
+    let (restart_tx, _) = broadcast::channel(1);
     let tcp_start = Arc::new(Notify::new());
     let tcp_start_cloned = tcp_start.clone();
     let config = Arc::new(RwLock::new(config));
@@ -536,30 +554,35 @@ fn main() -> Result<()> {
     let tx_cloned = tx.clone();
     let sensor_channel = Arc::new(Mutex::new(None));
     let sensor_channel_cloned = sensor_channel.clone();
+    let profile_connected = Arc::new(AtomicBool::new(false));
 
     // build and spawn main tokio runtime
     let runtime = Builder::new_multi_thread().enable_all().build().unwrap();
+    let restart_tx_cloned = restart_tx.clone();
+    let profile_connected_cloned = profile_connected.clone();
     runtime.spawn(async move {
         tokio_main(
             config_cloned,
             config_json.clone(),
-            need_restart,
+            restart_tx_cloned,
             tcp_start,
             args.config.clone(),
             tx_cloned,
             sensor_channel_cloned,
             led_support,
+            profile_connected_cloned,
         )
         .await
     });
 
     // start tokio_uring runtime simultaneously
     let _ = tokio_uring::start(io_loop(
-        need_restart_cloned,
+        restart_tx,
         tcp_start_cloned,
         config,
         tx,
         sensor_channel,
+        profile_connected,
     ));
 
     info!(