tuio_rs/
client.rs

1use std::{time::{Instant, Duration}, sync::{RwLock, Arc, Mutex}, collections::HashSet, cell::{RefCell, Cell}, thread};
2
3use indexmap::IndexMap;
4use ringbuffer::{ConstGenericRingBuffer, RingBufferWrite, RingBufferRead};
5use rosc::{OscPacket};
6
7use crate::{osc_receiver::{UdpReceiver, RoscReceiver}, cursor::{Cursor}, object::Object, blob::Blob, errors::{TuioError, OscReceiverError}, osc_encode_decode::{OscDecoder, DecodeOsc, self, Set}};
8
9#[derive(Default)]
10pub struct TuioEvents {
11    pub cursor_events: Vec<CursorEvent>,
12    pub object_events: Vec<ObjectEvent>,
13    pub blob_events: Vec<BlobEvent>,
14}
15
16pub struct CursorData {
17    pub source_name: String,
18    pub cursor: Cursor
19}
20
21pub struct ObjectData {
22    pub source_name: String,
23    pub object: Object
24}
25
26pub struct BlobData {
27    pub source_name: String,
28    pub blob: Blob
29}
30
31pub enum CursorEvent {
32    New(CursorData),
33    Update(CursorData),
34    Remove(CursorData),
35}
36
37pub enum ObjectEvent {
38    New(ObjectData),
39    Update(ObjectData),
40    Remove(ObjectData),
41}
42
43pub enum BlobEvent {
44    New(BlobData),
45    Update(BlobData),
46    Remove(BlobData),
47}
48
49#[derive(Default)]
50pub struct SourceCollection {
51    pub object_map: IndexMap<i32, Object>,
52    pub blob_map: IndexMap<i32, Blob>,
53    pub cursor_map: IndexMap<i32, Cursor>
54}
55
56pub struct Client {
57    current_frame: Cell<i32>,
58    instant: Instant,
59    current_time: Cell<Duration>,
60    pub source_list: RefCell<IndexMap<String, SourceCollection>>,
61    osc_receivers: Vec<Arc<RoscReceiver>>,
62    packet_buffer: Arc<Mutex<ConstGenericRingBuffer<OscPacket, 128>>>,
63    local_receiver: bool,
64    listen: Arc<RwLock<bool>>
65}
66
67/// Keeps the entries whose keys are contained in a [HashSet]
68/// 
69/// Returns a [Vec<i32>] of removed ids
70/// 
71/// # Arguments
72/// * `index_map` - an [IndexMap<i32, T>] to filter
73/// * `to_keep` - an [HashSet<i32>] containing the keys to retain
74fn retain_by_ids<T>(index_map: &mut IndexMap<i32, T>, to_keep: HashSet<i32>) -> Vec<T> {
75    let mut removed: Vec<T> = Vec::with_capacity(index_map.len());
76    let mut to_remove: Vec<i32> = Vec::with_capacity(index_map.len());
77
78    for id in index_map.keys() {
79        if !to_keep.contains(id) {
80            to_remove.push(*id);
81        }
82    }
83
84    for id in to_remove {
85        removed.push(index_map.remove(&id).unwrap());
86    }
87
88    removed
89}
90
91impl Client {
92    pub fn new() -> Result<Self, std::io::Error> {
93        Self::from_port(3333)
94    }
95
96    pub fn from_port(port: u16) -> Result<Self, std::io::Error> {
97        Ok(Self {
98            instant: Instant::now(),
99            osc_receivers: vec![Arc::new(UdpReceiver::from_port(port)?)],
100            current_frame: (-1).into(),
101            current_time: Cell::new(Duration::default()),
102            source_list: RefCell::new(IndexMap::new()),
103            local_receiver: true,
104            listen: Arc::new(RwLock::new(false)),
105            packet_buffer: Default::default()
106        })
107    }
108
109    pub fn connect(&self) -> Result<(), OscReceiverError> {
110        if *self.listen.read().unwrap() {
111            return Err(OscReceiverError::AlreadyConnected());
112        }
113
114        *self.listen.write().unwrap() = true;
115
116        for receiver in &self.osc_receivers {
117            receiver.connect().map_err(OscReceiverError::Connect)?;
118
119            let listen = Arc::clone(&self.listen);
120            let receiver = Arc::clone(receiver);
121            let buffer = Arc::clone(&self.packet_buffer);
122
123            thread::spawn(move || loop {
124                if !*listen.read().unwrap() {
125                    break;
126                }
127                
128                match receiver.recv() {
129                    Ok(packet) => {
130                        buffer.lock().unwrap().push(packet);
131                    }
132                    Err(err) => {
133                        match err {
134                            OscReceiverError::Receive(err) => if err.raw_os_error().unwrap() != 10004 {
135                                println!("Error receiving from socket: {}", err);
136                            },
137                            _ => println!("Error receiving from socket: {}", err)
138                        }
139                        
140                        break;
141                    }
142                }
143            });
144        }
145
146        Ok(())
147    }
148
149    pub fn disconnect(&self) {
150        for receiver in &self.osc_receivers {
151            receiver.disconnect();
152        }
153    }
154
155    /// Refreshes the client's state
156    /// 
157    /// On success, returns an [Option] containing the evnts of all new, updated and removed TUIO inputs
158    pub fn refresh(&self) -> Result<Option<TuioEvents>, TuioError> {
159        let mut updated = false;
160        let mut events = TuioEvents::default();
161
162        for packet in self.packet_buffer.lock().unwrap().drain() {
163            if self.process_osc_packet(packet, &mut events)? {
164                updated = true;
165            }
166        };
167
168        if updated {
169            Ok(Some(events))
170        }
171        else {
172            Ok(None)
173        }
174    }
175
176    /// Update frame parameters based on a frame number
177    /// 
178    /// Returns true if the frame is a new frame
179    /// # Argument
180    /// * `frame` - the new frame number
181    fn update_frame(&self, frame: i32) -> bool {
182        if frame >= 0 {
183            let current_frame = self.current_frame.get();
184            
185            if frame > current_frame {
186                self.current_time.set(self.instant.elapsed());
187            }
188            
189            if frame >= current_frame || current_frame - frame > 100 {
190                self.current_frame.set(frame);
191                return true;
192            }
193            else if self.instant.elapsed() - self.current_time.get() > Duration::from_millis(100){
194                self.current_time.set(self.instant.elapsed());
195                return false;
196            }
197        }
198        false
199    }
200
201    fn process_osc_packet(&self, packet: OscPacket, events: &mut TuioEvents) -> Result<bool, TuioError> {
202        if let OscPacket::Bundle(bundle) = packet {
203            let decoded_bundle = OscDecoder::decode_bundle(bundle)?;
204            
205            let to_keep: HashSet<i32> = HashSet::from_iter(decoded_bundle.alive);
206            
207            if self.update_frame(decoded_bundle.fseq) {
208                let mut source_list = self.source_list.borrow_mut();
209                let source_collection = source_list.entry(decoded_bundle.source.clone()).or_default(); 
210                match decoded_bundle.tuio_type {
211                    osc_encode_decode::TuioBundleType::Cursor => {
212                        let cursor_map = &mut source_collection.cursor_map;
213
214                        for cursor in retain_by_ids(cursor_map, to_keep).into_iter() {
215                            events.cursor_events.push(CursorEvent::Remove(CursorData{source_name: decoded_bundle.source.clone(), cursor: cursor.clone()}));
216                        }
217
218                        if let Some(Set::Cursor(cursor_collection)) = decoded_bundle.set {
219                            for cursor in cursor_collection {
220                                match cursor_map.entry(cursor.get_session_id()) {
221                                    indexmap::map::Entry::Occupied(mut entry) => {
222                                        events.cursor_events.push(CursorEvent::Update(CursorData{source_name: decoded_bundle.source.clone(), cursor: cursor.clone()}));
223                                        entry.insert(cursor);
224                                    },
225                                    indexmap::map::Entry::Vacant(entry) => {
226                                        events.cursor_events.push(CursorEvent::New(CursorData{source_name: decoded_bundle.source.clone(), cursor: cursor.clone()}));
227                                        entry.insert(cursor);
228                                    },
229                                }
230                            }
231                        }
232                    },
233                    osc_encode_decode::TuioBundleType::Object => {
234                        let object_map = &mut source_collection.object_map;
235
236                        for object in retain_by_ids(object_map, to_keep).into_iter() {
237                            events.object_events.push(ObjectEvent::Remove(ObjectData { source_name: decoded_bundle.source.clone(), object: object.clone() }));
238                        }
239
240                        if let Some(Set::Object(object_collection)) = decoded_bundle.set {
241                            for object in object_collection {
242                                match object_map.entry(object.get_session_id()) {
243                                    indexmap::map::Entry::Occupied(mut entry) => {
244                                        events.object_events.push(ObjectEvent::Update(ObjectData { source_name: decoded_bundle.source.clone(), object: object.clone() }));
245                                        entry.insert(object);
246                                    },
247                                    indexmap::map::Entry::Vacant(entry) => {
248                                        events.object_events.push(ObjectEvent::New(ObjectData { source_name: decoded_bundle.source.clone(), object: object.clone() }));
249                                        entry.insert(object);
250                                    },
251                                }
252                            }
253                        }
254                    },
255                    osc_encode_decode::TuioBundleType::Blob => {
256                        let blob_map = &mut source_collection.blob_map;
257
258                        for blob in retain_by_ids(blob_map, to_keep).into_iter() {
259                            events.blob_events.push(BlobEvent::Remove(BlobData { source_name: decoded_bundle.source.clone(), blob: blob.clone() }));
260                        }
261
262                        if let Some(Set::Blob(blob_collection)) = decoded_bundle.set {
263                            for blob in blob_collection {
264                                match blob_map.entry(blob.get_session_id()) {
265                                    indexmap::map::Entry::Occupied(mut entry) => {
266                                        events.blob_events.push(BlobEvent::Update(BlobData { source_name: decoded_bundle.source.clone(), blob: blob.clone() }));
267                                        entry.insert(blob);
268                                    },
269                                    indexmap::map::Entry::Vacant(entry) => {
270                                        events.blob_events.push(BlobEvent::New(BlobData { source_name: decoded_bundle.source.clone(), blob: blob.clone() }));
271                                        entry.insert(blob);
272                                    },
273                                }
274                            }
275                        }
276                    },
277                    osc_encode_decode::TuioBundleType::Unknown => (),
278                }
279                Ok(true)
280            }
281            else {
282                Ok(false)
283            }
284        }
285        else {
286            Err(TuioError::NotABundle(packet))
287        }
288    }
289
290    pub fn local_receiver(&self) -> bool {
291        self.local_receiver
292    }
293}
tuio_rs/client.rs

tuio_rs/
client.rs
