Merge branch 'quick-reconnect'

Author: manio

src/bluetooth.rs

@@ -1,4 +1,5 @@
 use crate::btle;
+use crate::config::Action;
 use crate::config::WifiConfig;
 use crate::config::IDENTITY_NAME;
 use crate::config_types::BluetoothAddressList;
@@ -11,10 +12,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;
 
@@ -514,25 +518,12 @@ impl Bluetooth {
         Ok(())
     }
 
-    pub async fn aa_handshake(
-        &mut self,
-        dongle_mode: bool,
-        connect: BluetoothAddressList,
-        wifi_config: WifiConfig,
-        tcp_start: Arc<Notify>,
-        bt_timeout: Duration,
-        stopped: bool,
-    ) -> Result<()> {
+    async fn send_params(wifi_config: WifiConfig, stream: &mut Stream) -> Result<()> {
         use WifiInfoResponse::WifiInfoResponse;
         use WifiStartRequest::WifiStartRequest;
         let mut stage = 1;
         let mut started;
 
-        // Use the provided session and adapter instead of creating new ones
-        let (address, mut stream) = self
-            .get_aa_profile_connection(dongle_mode, connect, bt_timeout, stopped)
-            .await?;
-
         info!("{} 📲 Sending parameters via bluetooth to phone...", NAME);
         let mut start_req = WifiStartRequest::new();
         info!(
@@ -541,10 +532,10 @@ impl Bluetooth {
         );
         start_req.set_ip_address(wifi_config.ip_addr);
         start_req.set_port(wifi_config.port);
-        send_message(&mut stream, stage, MessageId::WifiStartRequest, start_req).await?;
+        send_message(stream, stage, MessageId::WifiStartRequest, start_req).await?;
         stage += 1;
         started = Instant::now();
-        read_message(&mut stream, stage, MessageId::WifiInfoRequest, started).await?;
+        read_message(stream, stage, MessageId::WifiInfoRequest, started).await?;
 
         let mut info = WifiInfoResponse::new();
         info!(
@@ -557,19 +548,81 @@ impl Bluetooth {
         info.set_security_mode(SecurityMode::WPA2_PERSONAL);
         info.set_access_point_type(AccessPointType::DYNAMIC);
         stage += 1;
-        send_message(&mut stream, stage, MessageId::WifiInfoResponse, info).await?;
+        send_message(stream, stage, MessageId::WifiInfoResponse, info).await?;
         stage += 1;
         started = Instant::now();
-        read_message(&mut stream, stage, MessageId::WifiStartResponse, started).await?;
+        read_message(stream, stage, MessageId::WifiStartResponse, started).await?;
         stage += 1;
         started = Instant::now();
-        read_message(&mut stream, stage, MessageId::WifiConnectStatus, started).await?;
+        read_message(stream, stage, MessageId::WifiConnectStatus, started).await?;
+
+        Ok(())
+    }
+
+    pub async fn aa_handshake(
+        &mut self,
+        dongle_mode: bool,
+        connect: BluetoothAddressList,
+        wifi_config: WifiConfig,
+        tcp_start: Arc<Notify>,
+        bt_timeout: Duration,
+        stopped: bool,
+        quick_reconnect: bool,
+        mut need_restart: BroadcastReceiver<Option<Action>>,
+        profile_connected: Arc<AtomicBool>,
+    ) -> Result<()> {
+        // Use the provided session and adapter instead of creating new ones
+        let (address, mut stream) = self
+            .get_aa_profile_connection(dongle_mode, connect, bt_timeout, stopped)
+            .await?;
+        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 adapter_cloned = self.adapter.clone();
+            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 action = need_restart.recv().await;
+                    if let Ok(Some(action)) = action {
+                        // check if we need to stop now
+                        if action == Action::Stop {
+                            // disconnect and break
+                            if let Ok(device) = adapter_cloned.device(bluer::Address(*address)) {
+                                let _ = device.disconnect().await;
+                            }
+                            break;
+                        }
+                    }
+
+                    // 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/config.rs

@@ -84,6 +84,7 @@ pub struct AppConfig {
     pub hexdump_level: HexdumpLevel,
     pub disable_console_debug: bool,
     pub legacy: bool,
+    pub quick_reconnect: bool,
     pub connect: BluetoothAddressList,
     pub logfile: PathBuf,
     pub stats_interval: u16,
@@ -224,6 +225,7 @@ impl Default for AppConfig {
             hexdump_level: HexdumpLevel::Disabled,
             disable_console_debug: false,
             legacy: true,
+            quick_reconnect: false,
             connect: BluetoothAddressList::default(),
             logfile: "/var/log/aa-proxy-rs.log".into(),
             stats_interval: 0,
@@ -315,6 +317,7 @@ impl AppConfig {
         doc["hexdump_level"] = value(format!("{:?}", self.hexdump_level));
         doc["disable_console_debug"] = value(self.disable_console_debug);
         doc["legacy"] = value(self.legacy);
+        doc["quick_reconnect"] = value(self.quick_reconnect);
         doc["connect"] = value(self.connect.to_string());
         doc["logfile"] = value(self.logfile.display().to_string());
         doc["stats_interval"] = value(self.stats_interval as i64);

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<Option<Action>>,
     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(None);
+                    }
                     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(None);
+                }
                 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(None);
+                }
                 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(None);
+                    }
                     continue;
                 }
             }
@@ -542,8 +552,10 @@ pub async fn io_loop(
             NAME,
             format_duration(started.elapsed()).to_string()
         );
+        // obtain action for passing it to broadcast sender
+        let action = shared_config.read().await.action_requested.clone();
         // stream(s) closed, notify main loop to restart
-        need_restart.notify_one();
+        let _ = need_restart.send(action);
     }
 
     #[allow(unreachable_code)]