1use crate::error::{Result, VoxRtcError};
2use crate::socket::RawSocketChannel;
3use crate::types::*;
4use serde_json::Value;
5use std::ops::ControlFlow;
6use std::sync::{Arc, Mutex};
7use std::time::{SystemTime, UNIX_EPOCH};
8use tokio::sync::broadcast::error::RecvError;
9use tokio::task::JoinHandle;
10use tokio::time::{Duration, timeout};
11
12#[derive(Clone)]
13pub struct VoxRtcControlSession {
14 channel: RawSocketChannel,
15 session_id: String,
16 channel_name: String,
17 join_timeout: Duration,
18 response_generation: Arc<Mutex<ResponseGeneration>>,
19}
20
21#[derive(Default)]
22struct ResponseGeneration {
23 counter: u64,
24 id: Option<String>,
25}
26
27pub struct Listener {
28 handle: JoinHandle<()>,
29}
30
31impl Drop for Listener {
32 fn drop(&mut self) {
33 self.handle.abort();
34 }
35}
36
37impl VoxRtcControlSession {
38 pub(crate) fn new(
39 channel: RawSocketChannel,
40 session_id: String,
41 join_timeout: Duration,
42 ) -> Self {
43 let channel_name = format!("/rtc/{session_id}");
44 Self {
45 channel,
46 session_id,
47 channel_name,
48 join_timeout,
49 response_generation: Arc::new(Mutex::new(ResponseGeneration::default())),
50 }
51 }
52
53 pub fn session_id(&self) -> &str {
54 &self.session_id
55 }
56
57 pub fn channel_name(&self) -> &str {
58 &self.channel_name
59 }
60
61 pub async fn join(&self) -> Result<()> {
62 let mut states = self.channel.subscribe_state();
63 self.channel.join().await?;
64 let channel_name = self.channel.name().to_owned();
65 let channel = self.channel.clone();
66 timeout(self.join_timeout, async move {
67 loop {
68 let state = *states.borrow_and_update();
69 match state {
70 ChannelState::Joined => return Ok(()),
71 ChannelState::Closed | ChannelState::Declined => {
72 let reason = join_decline_reason(channel.decline_reason().await);
73 return Err(VoxRtcError::JoinFailed {
74 channel: channel_name,
75 state: format!("{state:?}"),
76 reason,
77 });
78 }
79 _ => {}
80 }
81 if states.changed().await.is_err() {
82 return Err(VoxRtcError::Disconnected);
83 }
84 }
85 })
86 .await
87 .map_err(|_| VoxRtcError::JoinTimeout(self.channel_name.clone()))?
88 }
89
90 pub async fn close(&self) -> Result<()> {
91 self.channel.leave().await
92 }
93
94 pub fn on_event<F>(&self, handler: F) -> Listener
95 where
96 F: Fn(WireEvent) + Send + Sync + 'static,
97 {
98 let mut messages = self.channel.subscribe_messages();
99 let session_id = self.session_id.clone();
100 let channel_name = self.channel_name.clone();
101 Listener {
102 handle: tokio::spawn(async move {
103 loop {
104 match next_message(messages.recv().await) {
105 ControlFlow::Break(()) => break,
106 ControlFlow::Continue(None) => continue,
107 ControlFlow::Continue(Some((event, payload))) => handler(WireEvent {
108 r#type: event,
109 data: payload,
110 session_id: session_id.clone(),
111 channel_name: channel_name.clone(),
112 }),
113 }
114 }
115 }),
116 }
117 }
118
119 pub fn on<F>(&self, event_name: impl Into<String>, handler: F) -> Listener
120 where
121 F: Fn(EventData) + Send + Sync + 'static,
122 {
123 let event_name = event_name.into();
124 let mut messages = self.channel.subscribe_messages();
125 Listener {
126 handle: tokio::spawn(async move {
127 loop {
128 match next_message(messages.recv().await) {
129 ControlFlow::Break(()) => break,
130 ControlFlow::Continue(None) => continue,
131 ControlFlow::Continue(Some((event, payload))) => {
132 if event == event_name {
133 handler(payload);
134 }
135 }
136 }
137 }
138 }),
139 }
140 }
141
142 pub fn on_session_attached<F>(&self, handler: F) -> Listener
143 where
144 F: Fn(SessionAttachedEvent) + Send + Sync + 'static,
145 {
146 let session_id = self.session_id.clone();
147 let channel_name = self.channel_name.clone();
148 self.on(EVENT_RTC_SESSION_ATTACHED, move |payload| {
149 handler(SessionAttachedEvent {
150 session_id: base_session_id(&payload, &session_id),
151 channel_name: channel_name.clone(),
152 data: payload,
153 })
154 })
155 }
156
157 pub fn on_session_created<F>(&self, handler: F) -> Listener
158 where
159 F: Fn(SessionCreatedEvent) + Send + Sync + 'static,
160 {
161 let session_id = self.session_id.clone();
162 let channel_name = self.channel_name.clone();
163 self.on(EVENT_SESSION_CREATED, move |payload| {
164 let session = payload.get("session").and_then(Value::as_object).cloned();
165 handler(SessionCreatedEvent {
166 session_id: base_session_id(&payload, &session_id),
167 channel_name: channel_name.clone(),
168 data: payload,
169 session,
170 });
171 })
172 }
173
174 pub fn on_transcript<F>(&self, handler: F) -> Listener
175 where
176 F: Fn(TranscriptEvent) + Send + Sync + 'static,
177 {
178 let session_id = self.session_id.clone();
179 let channel_name = self.channel_name.clone();
180 self.on(EVENT_TRANSCRIPT_COMPLETED, move |payload| {
181 handler(TranscriptEvent {
182 session_id: base_session_id(&payload, &session_id),
183 channel_name: channel_name.clone(),
184 transcript: required_string(&payload, "transcript", ""),
185 language: optional_string(&payload, "language"),
186 start_ms: optional_number(&payload, "start_ms"),
187 end_ms: optional_number(&payload, "end_ms"),
188 eou_probability: optional_number(&payload, "eou_probability"),
189 topics: optional_string_vec(&payload, "topics"),
190 data: payload,
191 });
192 })
193 }
194
195 pub fn on_turn_state_changed<F>(&self, handler: F) -> Listener
196 where
197 F: Fn(TurnStateEvent) + Send + Sync + 'static,
198 {
199 let session_id = self.session_id.clone();
200 let channel_name = self.channel_name.clone();
201 self.on(EVENT_TURN_STATE_CHANGED, move |payload| {
202 handler(TurnStateEvent {
203 session_id: base_session_id(&payload, &session_id),
204 channel_name: channel_name.clone(),
205 state: required_string(&payload, "state", "unknown"),
206 previous_state: optional_string(&payload, "previous_state"),
207 data: payload,
208 });
209 })
210 }
211
212 pub fn on_speech_started<F>(&self, handler: F) -> Listener
213 where
214 F: Fn(SpeechStartedEvent) + Send + Sync + 'static,
215 {
216 let session_id = self.session_id.clone();
217 let channel_name = self.channel_name.clone();
218 self.on(EVENT_SPEECH_STARTED, move |payload| {
219 handler(SpeechStartedEvent {
220 session_id: base_session_id(&payload, &session_id),
221 channel_name: channel_name.clone(),
222 timestamp_ms: optional_number(&payload, "timestamp_ms"),
223 data: payload,
224 });
225 })
226 }
227
228 pub fn on_speech_stopped<F>(&self, handler: F) -> Listener
229 where
230 F: Fn(SpeechStoppedEvent) + Send + Sync + 'static,
231 {
232 let session_id = self.session_id.clone();
233 let channel_name = self.channel_name.clone();
234 self.on(EVENT_SPEECH_STOPPED, move |payload| {
235 handler(SpeechStoppedEvent {
236 session_id: base_session_id(&payload, &session_id),
237 channel_name: channel_name.clone(),
238 timestamp_ms: optional_number(&payload, "timestamp_ms"),
239 data: payload,
240 });
241 })
242 }
243
244 pub fn on_transcript_delta<F>(&self, handler: F) -> Listener
245 where
246 F: Fn(TranscriptDeltaEvent) + Send + Sync + 'static,
247 {
248 let session_id = self.session_id.clone();
249 let channel_name = self.channel_name.clone();
250 self.on(EVENT_TRANSCRIPT_DELTA, move |payload| {
251 handler(TranscriptDeltaEvent {
252 session_id: base_session_id(&payload, &session_id),
253 channel_name: channel_name.clone(),
254 delta: required_string(&payload, "delta", ""),
255 start_ms: optional_number(&payload, "start_ms"),
256 end_ms: optional_number(&payload, "end_ms"),
257 data: payload,
258 });
259 })
260 }
261
262 pub fn on_turn_eou_predicted<F>(&self, handler: F) -> Listener
263 where
264 F: Fn(TurnEouPredictedEvent) + Send + Sync + 'static,
265 {
266 let session_id = self.session_id.clone();
267 let channel_name = self.channel_name.clone();
268 self.on(EVENT_TURN_EOU_PREDICTED, move |payload| {
269 handler(TurnEouPredictedEvent {
270 session_id: base_session_id(&payload, &session_id),
271 channel_name: channel_name.clone(),
272 probability: optional_number(&payload, "probability"),
273 threshold: optional_number(&payload, "threshold"),
274 delay_ms: optional_number(&payload, "delay_ms"),
275 start_ms: optional_number(&payload, "start_ms"),
276 end_ms: optional_number(&payload, "end_ms"),
277 decision: optional_string(&payload, "decision"),
278 action: optional_string(&payload, "action"),
279 turn_detector: optional_string(&payload, "turn_detector"),
280 data: payload,
281 });
282 })
283 }
284
285 pub fn on_response_created<F>(&self, handler: F) -> Listener
286 where
287 F: Fn(ResponseEvent) + Send + Sync + 'static,
288 {
289 self.on_response_event(EVENT_RESPONSE_CREATED, handler)
290 }
291
292 pub fn on_response_committed<F>(&self, handler: F) -> Listener
293 where
294 F: Fn(ResponseEvent) + Send + Sync + 'static,
295 {
296 self.on_response_event(EVENT_RESPONSE_COMMITTED, handler)
297 }
298
299 pub fn on_response_done<F>(&self, handler: F) -> Listener
300 where
301 F: Fn(ResponseEvent) + Send + Sync + 'static,
302 {
303 self.on_response_event(EVENT_RESPONSE_DONE, handler)
304 }
305
306 pub fn on_response_cancelled<F>(&self, handler: F) -> Listener
307 where
308 F: Fn(ResponseEvent) + Send + Sync + 'static,
309 {
310 self.on_response_event(EVENT_RESPONSE_CANCELLED, handler)
311 }
312
313 pub fn on_response_audio_clear<F>(&self, handler: F) -> Listener
314 where
315 F: Fn(ResponseEvent) + Send + Sync + 'static,
316 {
317 self.on_response_event(EVENT_RESPONSE_AUDIO_CLEAR, handler)
318 }
319
320 fn on_response_event<F>(&self, event_name: &'static str, handler: F) -> Listener
321 where
322 F: Fn(ResponseEvent) + Send + Sync + 'static,
323 {
324 let session_id = self.session_id.clone();
325 let channel_name = self.channel_name.clone();
326 self.on(event_name, move |payload| {
327 handler(response_event(payload, &session_id, &channel_name));
328 })
329 }
330
331 pub fn on_interruption_detected<F>(&self, handler: F) -> Listener
332 where
333 F: Fn(InterruptionEvent) + Send + Sync + 'static,
334 {
335 self.on_interruption_event(EVENT_INTERRUPTION_DETECTED, handler)
336 }
337
338 pub fn on_interruption_false_positive<F>(&self, handler: F) -> Listener
339 where
340 F: Fn(InterruptionEvent) + Send + Sync + 'static,
341 {
342 self.on_interruption_event(EVENT_INTERRUPTION_FALSE_POSITIVE, handler)
343 }
344
345 fn on_interruption_event<F>(&self, event_name: &'static str, handler: F) -> Listener
346 where
347 F: Fn(InterruptionEvent) + Send + Sync + 'static,
348 {
349 let session_id = self.session_id.clone();
350 let channel_name = self.channel_name.clone();
351 self.on(event_name, move |payload| {
352 handler(InterruptionEvent {
353 response: response_event(payload.clone(), &session_id, &channel_name),
354 vad_active_ms: optional_number(&payload, "vad_active_ms"),
355 partial_transcript: optional_string(&payload, "partial_transcript"),
356 });
357 })
358 }
359
360 pub fn on_browser_event<F>(&self, handler: F) -> Listener
361 where
362 F: Fn(BrowserEvent) + Send + Sync + 'static,
363 {
364 let session_id = self.session_id.clone();
365 let channel_name = self.channel_name.clone();
366 self.on(EVENT_BROWSER_EVENT, move |payload| {
367 handler(BrowserEvent {
368 session_id: base_session_id(&payload, &session_id),
369 channel_name: channel_name.clone(),
370 event: required_string(&payload, "event", ""),
371 payload: payload.get("payload").cloned().unwrap_or(Value::Null),
372 data: payload,
373 });
374 })
375 }
376
377 pub fn on_close<F>(&self, handler: F) -> Listener
378 where
379 F: Fn(CloseEvent) + Send + Sync + 'static,
380 {
381 let session_id = self.session_id.clone();
382 let channel_name = self.channel_name.clone();
383 self.on(EVENT_RTC_CLIENT_DISCONNECTED, move |payload| {
384 handler(CloseEvent {
385 session_id: base_session_id(&payload, &session_id),
386 channel_name: channel_name.clone(),
387 reason: required_string(&payload, "reason", "unknown"),
388 connection_state: optional_string(&payload, "connection_state"),
389 ice_connection_state: optional_string(&payload, "ice_connection_state"),
390 data_channel_state: optional_string(&payload, "data_channel_state"),
391 data: payload,
392 });
393 })
394 }
395
396 pub fn on_error<F>(&self, handler: F) -> Listener
397 where
398 F: Fn(ErrorEvent) + Send + Sync + 'static,
399 {
400 let session_id = self.session_id.clone();
401 let channel_name = self.channel_name.clone();
402 self.on(EVENT_ERROR, move |payload| {
403 handler(ErrorEvent {
404 session_id: base_session_id(&payload, &session_id),
405 channel_name: channel_name.clone(),
406 message: optional_string(&payload, "message"),
407 code: optional_string(&payload, "code"),
408 data: payload,
409 });
410 })
411 }
412
413 pub async fn send_control(&self, event: &str, payload: EventData) -> Result<()> {
414 self.channel.send_message(event, payload).await
415 }
416
417 pub async fn configure(&self, config: SessionConfig) -> Result<()> {
418 let mut session = config.extra;
419 insert_opt(&mut session, "stt_model", config.stt_model);
420 insert_opt(&mut session, "tts_model", config.tts_model);
421 insert_opt(&mut session, "voice", config.voice);
422 insert_opt(&mut session, "turn_profile", config.turn_profile);
423 insert_opt(&mut session, "vad_backend", config.vad_backend);
424 insert_opt(&mut session, "turn_detector", config.turn_detector);
425
426 let mut payload = EventData::new();
427 payload.insert("session".to_owned(), Value::Object(session));
428 self.send_control("session.update", payload).await
429 }
430
431 pub async fn start_response(&self, options: Option<ResponseOptions>) -> Result<()> {
432 let mut payload = response_options_payload(options);
433 let generation_id = self.next_response_generation();
434 payload.insert("generation_id".to_owned(), Value::String(generation_id));
435 self.send_control("response.start", payload).await
436 }
437
438 pub async fn append_response_text(
439 &self,
440 delta: impl Into<String>,
441 options: Option<ResponseOptions>,
442 ) -> Result<()> {
443 let mut payload = response_options_payload(options);
444 payload.insert("delta".to_owned(), Value::String(delta.into()));
445 self.add_response_generation(&mut payload);
446 self.send_control("response.delta", payload).await
447 }
448
449 pub async fn commit_response(&self) -> Result<()> {
450 let mut payload = EventData::new();
451 self.add_response_generation(&mut payload);
452 self.send_control("response.commit", payload).await
453 }
454
455 pub async fn cancel_response(&self) -> Result<()> {
456 let mut payload = EventData::new();
457 self.add_response_generation(&mut payload);
458 self.clear_response_generation();
459 self.send_control("response.cancel", payload).await
460 }
461
462 pub async fn replace_response_text(
463 &self,
464 text: impl Into<String>,
465 options: Option<ResponseOptions>,
466 ) -> Result<()> {
467 self.clear_response_generation();
468 let mut payload = response_options_payload(options);
469 payload.insert("text".to_owned(), Value::String(text.into()));
470 self.send_control("response.replace_text", payload).await
471 }
472
473 pub async fn send_text_response(
474 &self,
475 text: impl Into<String>,
476 options: Option<ResponseOptions>,
477 cancel_first: bool,
478 ) -> Result<()> {
479 let text = text.into();
480 if cancel_first {
481 return self.replace_response_text(text, options).await;
482 }
483 self.start_response(options.clone()).await?;
484 self.append_response_text(text, options).await?;
485 self.commit_response().await
486 }
487
488 pub async fn send_client_event(&self, envelope: ClientEventEnvelope) -> Result<()> {
489 let mut payload = EventData::new();
490 payload.insert("event".to_owned(), Value::String(envelope.event));
491 payload.insert("payload".to_owned(), envelope.payload);
492 self.send_control(EVENT_CLIENT_EVENT, payload).await
493 }
494
495 fn next_response_generation(&self) -> String {
496 let mut state = self
497 .response_generation
498 .lock()
499 .expect("response generation mutex poisoned");
500 state.counter += 1;
501 let nonce = SystemTime::now()
502 .duration_since(UNIX_EPOCH)
503 .unwrap_or_default()
504 .as_nanos();
505 let generation_id = format!("generation_{}_{}", state.counter, nonce);
506 state.id = Some(generation_id.clone());
507 generation_id
508 }
509
510 fn add_response_generation(&self, payload: &mut EventData) {
511 let state = self
512 .response_generation
513 .lock()
514 .expect("response generation mutex poisoned");
515 if let Some(generation_id) = &state.id {
516 payload.insert(
517 "generation_id".to_owned(),
518 Value::String(generation_id.clone()),
519 );
520 }
521 }
522
523 fn clear_response_generation(&self) {
524 self.response_generation
525 .lock()
526 .expect("response generation mutex poisoned")
527 .id = None;
528 }
529}
530
531fn next_message(
532 result: std::result::Result<(String, EventData), RecvError>,
533) -> ControlFlow<(), Option<(String, EventData)>> {
534 match result {
535 Ok(message) => ControlFlow::Continue(Some(message)),
536 Err(RecvError::Lagged(_)) => ControlFlow::Continue(None),
537 Err(RecvError::Closed) => ControlFlow::Break(()),
538 }
539}
540
541fn join_decline_reason(reason: Option<EventData>) -> Option<String> {
542 let reason = reason?;
543 for key in ["message", "reason", "error"] {
544 if let Some(value) = reason.get(key).and_then(Value::as_str)
545 && !value.is_empty()
546 {
547 return Some(value.to_owned());
548 }
549 }
550 if reason.is_empty() {
551 None
552 } else {
553 Some(Value::Object(reason).to_string())
554 }
555}
556
557fn insert_opt(session: &mut EventData, key: &str, value: Option<String>) {
558 if let Some(value) = value {
559 session.insert(key.to_owned(), Value::String(value));
560 }
561}
562
563fn response_options_payload(options: Option<ResponseOptions>) -> EventData {
564 let mut payload = EventData::new();
565 if let Some(options) = options
566 && let Some(allow) = options.allow_interruptions
567 {
568 payload.insert("allow_interruptions".to_owned(), Value::Bool(allow));
569 }
570 payload
571}
572
573fn base_session_id(payload: &EventData, fallback: &str) -> String {
574 required_string(payload, "session_id", fallback)
575}
576
577fn response_event(payload: EventData, session_id: &str, channel_name: &str) -> ResponseEvent {
578 ResponseEvent {
579 session_id: base_session_id(&payload, session_id),
580 channel_name: channel_name.to_owned(),
581 response_id: optional_string(&payload, "response_id"),
582 data: payload,
583 }
584}
585
586#[cfg(test)]
587mod tests {
588 use super::*;
589 use crate::socket::test_channel;
590 use serde_json::json;
591 use tokio::sync::broadcast;
592 use tokio::sync::mpsc;
593
594 async fn session() -> (VoxRtcControlSession, broadcast::Sender<(String, EventData)>) {
595 let (channel, sender) = test_channel().await;
596 let session =
597 VoxRtcControlSession::new(channel, "sess-1".to_owned(), Duration::from_secs(1));
598 (session, sender)
599 }
600
601 fn payload(value: Value) -> EventData {
602 value.as_object().cloned().expect("object payload")
603 }
604
605 #[test]
606 fn join_decline_reason_prefers_structured_message_fields() {
607 assert_eq!(
608 join_decline_reason(Some(payload(json!({ "message": "expired" })))),
609 Some("expired".to_owned())
610 );
611 assert_eq!(
612 join_decline_reason(Some(payload(json!({ "reason": "missing" })))),
613 Some("missing".to_owned())
614 );
615 assert_eq!(
616 join_decline_reason(Some(payload(json!({ "channel": "/rtc/abc" })))),
617 Some(r#"{"channel":"/rtc/abc"}"#.to_owned())
618 );
619 assert_eq!(join_decline_reason(Some(EventData::new())), None);
620 assert_eq!(join_decline_reason(None), None);
621 }
622
623 async fn recv<T>(rx: &mut mpsc::UnboundedReceiver<T>) -> T {
624 timeout(Duration::from_secs(1), rx.recv())
625 .await
626 .expect("handler fired within timeout")
627 .expect("handler produced an event")
628 }
629
630 #[test]
631 fn next_message_classifies_lag_close_and_ok() {
632 assert!(matches!(
633 next_message(Ok(("e".to_owned(), EventData::new()))),
634 ControlFlow::Continue(Some(_))
635 ));
636 assert!(matches!(
637 next_message(Err(RecvError::Lagged(7))),
638 ControlFlow::Continue(None)
639 ));
640 assert!(matches!(
641 next_message(Err(RecvError::Closed)),
642 ControlFlow::Break(())
643 ));
644 }
645
646 #[tokio::test]
647 async fn response_commands_share_one_generation_id() {
648 let (session, _) = session().await;
649 let generation_id = session.next_response_generation();
650 let mut delta = payload(json!({ "delta": "hello" }));
651 session.add_response_generation(&mut delta);
652 let mut commit = EventData::new();
653 session.add_response_generation(&mut commit);
654
655 assert_eq!(
656 delta.get("generation_id"),
657 Some(&Value::String(generation_id.clone()))
658 );
659 assert_eq!(
660 commit.get("generation_id"),
661 Some(&Value::String(generation_id))
662 );
663 }
664
665 #[tokio::test]
666 async fn on_speech_started_fires_with_timestamp() {
667 let (session, sender) = session().await;
668 let (tx, mut rx) = mpsc::unbounded_channel();
669 let _listener = session.on_speech_started(move |event| {
670 tx.send(event).unwrap();
671 });
672 sender
673 .send((
674 EVENT_SPEECH_STARTED.to_owned(),
675 payload(json!({ "session_id": "sess-1", "timestamp_ms": 1234 })),
676 ))
677 .unwrap();
678 let event = recv(&mut rx).await;
679 assert_eq!(event.session_id, "sess-1");
680 assert_eq!(event.channel_name, "/rtc/sess-1");
681 assert_eq!(event.timestamp_ms, Some(1234.0));
682 }
683
684 #[tokio::test]
685 async fn on_speech_stopped_fires_with_timestamp() {
686 let (session, sender) = session().await;
687 let (tx, mut rx) = mpsc::unbounded_channel();
688 let _listener = session.on_speech_stopped(move |event| {
689 tx.send(event).unwrap();
690 });
691 sender
692 .send((
693 EVENT_SPEECH_STOPPED.to_owned(),
694 payload(json!({ "timestamp_ms": 5678 })),
695 ))
696 .unwrap();
697 let event = recv(&mut rx).await;
698 assert_eq!(event.timestamp_ms, Some(5678.0));
699 }
700
701 #[tokio::test]
702 async fn on_transcript_delta_fires_with_fields() {
703 let (session, sender) = session().await;
704 let (tx, mut rx) = mpsc::unbounded_channel();
705 let _listener = session.on_transcript_delta(move |event| {
706 tx.send(event).unwrap();
707 });
708 sender
709 .send((
710 EVENT_TRANSCRIPT_DELTA.to_owned(),
711 payload(json!({ "delta": "hel", "start_ms": 10, "end_ms": 20 })),
712 ))
713 .unwrap();
714 let event = recv(&mut rx).await;
715 assert_eq!(event.delta, "hel");
716 assert_eq!(event.start_ms, Some(10.0));
717 assert_eq!(event.end_ms, Some(20.0));
718 }
719
720 #[tokio::test]
721 async fn on_turn_eou_predicted_fires_with_fields() {
722 let (session, sender) = session().await;
723 let (tx, mut rx) = mpsc::unbounded_channel();
724 let _listener = session.on_turn_eou_predicted(move |event| {
725 tx.send(event).unwrap();
726 });
727 sender
728 .send((
729 EVENT_TURN_EOU_PREDICTED.to_owned(),
730 payload(json!({
731 "probability": 0.82,
732 "threshold": 0.5,
733 "delay_ms": 120,
734 "start_ms": 0,
735 "end_ms": 300,
736 "decision": "end",
737 "action": "commit",
738 "turn_detector": "smart"
739 })),
740 ))
741 .unwrap();
742 let event = recv(&mut rx).await;
743 assert_eq!(event.probability, Some(0.82));
744 assert_eq!(event.threshold, Some(0.5));
745 assert_eq!(event.delay_ms, Some(120.0));
746 assert_eq!(event.start_ms, Some(0.0));
747 assert_eq!(event.end_ms, Some(300.0));
748 assert_eq!(event.decision.as_deref(), Some("end"));
749 assert_eq!(event.action.as_deref(), Some("commit"));
750 assert_eq!(event.turn_detector.as_deref(), Some("smart"));
751 }
752
753 #[tokio::test]
754 async fn handler_survives_a_lagged_broadcast() {
755 let (session, sender) = session().await;
756 let (tx, mut rx) = mpsc::unbounded_channel();
757 let _listener = session.on_speech_started(move |event| {
758 tx.send(event.timestamp_ms).unwrap();
759 });
760
761 for index in 0..2100u32 {
762 let _ = sender.send((
763 EVENT_SPEECH_STARTED.to_owned(),
764 payload(json!({ "timestamp_ms": index })),
765 ));
766 }
767 let _ = sender.send((
768 EVENT_SPEECH_STARTED.to_owned(),
769 payload(json!({ "timestamp_ms": 9999 })),
770 ));
771
772 let mut saw_final = false;
773 while let Ok(Some(value)) = timeout(Duration::from_secs(1), rx.recv()).await {
774 if value == Some(9999.0) {
775 saw_final = true;
776 break;
777 }
778 }
779 assert!(
780 saw_final,
781 "loop must keep delivering events after a broadcast lag"
782 );
783 }
784}