oxidite_queue/
worker.rs

1use std::sync::Arc;
2use std::time::Duration;
3use tokio::time::sleep;
4use crate::queue::Queue;
5
6
7/// Worker for processing jobs
8pub struct Worker {
9    queue: Arc<Queue>,
10    worker_count: usize,
11    poll_interval: Duration,
12}
13
14impl Worker {
15    /// Create a new worker with the given queue
16    pub fn new(queue: Arc<Queue>) -> Self {
17        Self {
18            queue,
19            worker_count: 4,
20            poll_interval: Duration::from_secs(1),
21        }
22    }
23
24    /// Set the number of concurrent workers
25    pub fn worker_count(mut self, count: usize) -> Self {
26        self.worker_count = count;
27        self
28    }
29
30    /// Set the poll interval when no jobs are available
31    pub fn poll_interval(mut self, interval: Duration) -> Self {
32        self.poll_interval = interval;
33        self
34    }
35
36    /// Start the worker (runs forever)
37    pub async fn start(self) {
38        println!("Starting {} workers...", self.worker_count);
39        
40        let mut handles = vec![];
41        
42        for i in 0..self.worker_count {
43            let queue = self.queue.clone();
44            let poll_interval = self.poll_interval;
45            
46            let handle = tokio::spawn(async move {
47                loop {
48                    match queue.dequeue().await {
49                        Ok(Some(job)) => {
50                            println!("Worker {}: Processing job {}", i, job.id);
51                            
52                            // In a real implementation, deserialize and execute the job
53                            // For now, just mark as complete
54                            sleep(Duration::from_millis(100)).await;
55                            
56                            if let Err(e) = queue.complete(&job.id).await {
57                                eprintln!("Worker {}: Failed to mark job as complete: {}", i, e);
58                            }
59                        }
60                        Ok(None) => {
61                            // No jobs available, sleep
62                            sleep(poll_interval).await;
63                        }
64                        Err(e) => {
65                            eprintln!("Worker {}: Error dequeuing job: {}", i, e);
66                            sleep(poll_interval).await;
67                        }
68                    }
69                }
70            });
71            
72            handles.push(handle);
73        }
74
75        // Wait for all workers (they run forever)
76        for handle in handles {
77            let _ = handle.await;
78        }
79    }
80}
oxidite_queue/worker.rs

oxidite_queue/
worker.rs
