fix(listener): handle shutdown during realtime audio and await control sends (#1878)

- Prevent blocking when initiating realtime audio by selecting on a 
shutdown   receiver. If shutdown is triggered while calling 
client.from_realtime_audio,   return early instead of awaiting 
indefinitely. This is added in two places   where outbound streams are 
created. - Unify result handling by matching on the selected result 
(res) and preserve   existing error reporting when from_realtime_audio 
fails. - Make control message forwarding asynchronous: replace try_send 
with awaitable   send calls so control messages are properly delivered 
to mic and speaker   channels.
- - Ensure graceful stream termination by sending shutdown signals to 
c and   speaker shutdown channels when the inbound stream ends.
- These changes improve responsiveness to shutdown and ensure control 
- ssages
are reliably forwarded.

Author: yujonglee

plugins/listener/src/actors/listener.rs

@@ -259,7 +259,15 @@ async fn run_single_stream(
 
     let outbound = tokio_stream::wrappers::ReceiverStream::new(rx);
 
-    let (listen_stream, handle) = match client.from_realtime_audio(outbound).await {
+    let mut shutdown_rx = shutdown_rx;
+    let res = tokio::select! {
+        res = client.from_realtime_audio(outbound) => res,
+        _ = &mut shutdown_rx => {
+            return;
+        }
+    };
+
+    let (listen_stream, handle) = match res {
         Ok(res) => res,
         Err(e) => {
             let _ = myself.send_message(ListenerMsg::StreamStartFailed(format!("{:?}", e)));
@@ -296,7 +304,15 @@ async fn run_dual_stream(
 
     let outbound = tokio_stream::wrappers::ReceiverStream::new(rx);
 
-    let (listen_stream, handle) = match client.from_realtime_audio(outbound).await {
+    let mut shutdown_rx = shutdown_rx;
+    let res = tokio::select! {
+        res = client.from_realtime_audio(outbound) => res,
+        _ = &mut shutdown_rx => {
+            return;
+        }
+    };
+
+    let (listen_stream, handle) = match res {
         Ok(res) => res,
         Err(e) => {
             let _ = myself.send_message(ListenerMsg::StreamStartFailed(format!("{:?}", e)));
@@ -448,10 +464,12 @@ async fn spawn_rx_task_dual_split(
                                 let _ = spk_tx.try_send(MixedMessage::Audio(spk));
                             }
                             Some(MixedMessage::Control(ctrl)) => {
-                                let _ = mic_tx.try_send(MixedMessage::Control(ctrl.clone()));
-                                let _ = spk_tx.try_send(MixedMessage::Control(ctrl));
+                                let _ = mic_tx.send(MixedMessage::Control(ctrl.clone())).await;
+                                let _ = spk_tx.send(MixedMessage::Control(ctrl)).await;
                             }
                             None => {
+                                let _ = shutdown_tx_mic.send(());
+                                let _ = shutdown_tx_spk.send(());
                                 break;
                             }
                         }