rustvani 0.1.2

Voice AI framework for Rust — real-time speech pipelines with STT, LLM, TTS, and Dhara conversation flows
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

use std::collections::VecDeque;
use std::sync::Arc;
use futures::future::BoxFuture;
use tokio::sync::Mutex;
use tokio::sync::Notify;

use super::direction::FrameDirection;
use super::Frame;

pub type QueueCallback = Box<dyn FnOnce() -> BoxFuture<'static, ()> + Send>;
pub type QueueItem = (Frame, FrameDirection, Option<QueueCallback>);

/// Priority queue: system frames go to the front, others to the back.
pub struct FrameProcessorQueue {
    system:     Arc<Mutex<VecDeque<QueueItem>>>,
    non_system: Arc<Mutex<VecDeque<QueueItem>>>,
    notify:     Arc<Notify>,
}

impl FrameProcessorQueue {
    pub fn new() -> Self {
        Self {
            system:     Arc::new(Mutex::new(VecDeque::new())),
            non_system: Arc::new(Mutex::new(VecDeque::new())),
            notify:     Arc::new(Notify::new()),
        }
    }

    pub async fn put(&self, item: QueueItem) {
        if item.0.is_system() {
            self.system.lock().await.push_back(item);
        } else {
            self.non_system.lock().await.push_back(item);
        }
        self.notify.notify_one();
    }

    /// Get next item — system frames take priority.
    pub async fn get(&self) -> QueueItem {
        loop {
            {
                if let Some(item) = self.system.lock().await.pop_front() {
                    return item;
                }
                if let Some(item) = self.non_system.lock().await.pop_front() {
                    return item;
                }
            }
            self.notify.notified().await;
        }
    }

    /// Drain all non-uninterruptible frames from the non-system queue.
    pub async fn drain_keep_uninterruptible(&self) {
        let mut q = self.non_system.lock().await;
        q.retain(|(frame, _, _)| frame.is_uninterruptible());
    }
}

/// Simple FIFO queue for the process task.
pub struct ProcessQueue {
    inner:  Arc<Mutex<VecDeque<QueueItem>>>,
    notify: Arc<Notify>,
}

impl ProcessQueue {
    pub fn new() -> Self {
        Self {
            inner:  Arc::new(Mutex::new(VecDeque::new())),
            notify: Arc::new(Notify::new()),
        }
    }

    pub async fn put(&self, item: QueueItem) {
        self.inner.lock().await.push_back(item);
        self.notify.notify_one();
    }

    pub async fn get(&self) -> QueueItem {
        loop {
            if let Some(item) = self.inner.lock().await.pop_front() {
                return item;
            }
            self.notify.notified().await;
        }
    }

    pub async fn drain_keep_uninterruptible(&self) {
        let mut q = self.inner.lock().await;
        q.retain(|(frame, _, _)| frame.is_uninterruptible());
    }
}