rill_osc/
server.rs

1use std::net::SocketAddr;
2use std::sync::Arc;
3
4use tokio::net::UdpSocket;
5
6use crate::error::Error;
7use crate::osc::{self, OscMessage, OscPacket};
8
9type Handler = Arc<dyn Fn(OscMessage, SocketAddr) + Send + Sync + 'static>;
10
11pub struct OscServer {
12    socket: Arc<UdpSocket>,
13    handlers: Vec<(String, Handler)>,
14    buffer_size: usize,
15}
16
17impl OscServer {
18    pub async fn bind(addr: impl tokio::net::ToSocketAddrs) -> Result<Self, Error> {
19        let socket = UdpSocket::bind(addr).await?;
20        Ok(Self {
21            socket: Arc::new(socket),
22            handlers: Vec::new(),
23            buffer_size: 65536,
24        })
25    }
26
27    pub fn handle<F>(&mut self, pattern: impl Into<String>, f: F)
28    where
29        F: Fn(OscMessage, SocketAddr) + Send + Sync + 'static,
30    {
31        self.handlers.push((pattern.into(), Arc::new(f)));
32    }
33
34    pub fn set_buffer_size(&mut self, size: usize) {
35        self.buffer_size = size;
36    }
37
38    pub fn local_addr(&self) -> Result<SocketAddr, Error> {
39        Ok(self.socket.local_addr()?)
40    }
41
42    pub async fn run(&self) -> Result<(), Error> {
43        let mut buf = vec![0u8; self.buffer_size];
44        loop {
45            let (n, src) = self.socket.recv_from(&mut buf).await?;
46            let data = &buf[..n];
47            match osc::decode(data) {
48                Ok(packet) => {
49                    if let Err(e) = self.dispatch(packet, src) {
50                        log::warn!("dispatch error: {}", e);
51                    }
52                }
53                Err(e) => {
54                    log::warn!("failed to decode OSC packet from {}: {}", src, e);
55                }
56            }
57        }
58    }
59
60    fn dispatch(&self, packet: OscPacket, src: SocketAddr) -> Result<(), Error> {
61        match packet {
62            OscPacket::Message(msg) => {
63                for (pattern, handler) in &self.handlers {
64                    if osc::pattern_match(pattern, &msg.addr) {
65                        handler(msg.clone(), src);
66                    }
67                }
68                Ok(())
69            }
70            OscPacket::Bundle(bundle) => {
71                for p in bundle.packets {
72                    self.dispatch(p, src)?;
73                }
74                Ok(())
75            }
76        }
77    }
78}
79
80#[cfg(test)]
81mod tests {
82    use super::*;
83    use crate::osc::{OscBundle, OscMessage, OscPacket, OscType, TimeTag};
84    use std::sync::atomic::{AtomicBool, Ordering};
85    use std::sync::Arc;
86
87    #[test]
88    fn test_pattern_registration() {
89        let server = OscServer::bind("127.0.0.1:0");
90        let rt = tokio::runtime::Builder::new_current_thread()
91            .enable_io()
92            .build()
93            .unwrap();
94        let mut server = rt.block_on(server).unwrap();
95
96        let called = Arc::new(AtomicBool::new(false));
97        let called_clone = called.clone();
98
99        server.handle("/test", move |_msg, _src| {
100            called_clone.store(true, Ordering::SeqCst);
101        });
102
103        assert_eq!(server.handlers.len(), 1);
104
105        let msg = OscMessage {
106            addr: "/test".into(),
107            args: vec![OscType::Int(42)],
108        };
109
110        server
111            .dispatch(OscPacket::Message(msg), "127.0.0.1:0".parse().unwrap())
112            .unwrap();
113
114        assert!(called.load(Ordering::SeqCst));
115    }
116
117    #[test]
118    fn test_dispatch_bundle_unwraps() {
119        let server = OscServer::bind("127.0.0.1:0");
120        let rt = tokio::runtime::Builder::new_current_thread()
121            .enable_io()
122            .build()
123            .unwrap();
124        let mut server = rt.block_on(server).unwrap();
125
126        let count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
127        let count_clone = count.clone();
128
129        server.handle("/*", move |_msg, _src| {
130            count_clone.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
131        });
132
133        let bundle = OscBundle {
134            timetag: TimeTag::immediate(),
135            packets: vec![
136                OscPacket::Message(OscMessage {
137                    addr: "/a".into(),
138                    args: vec![],
139                }),
140                OscPacket::Message(OscMessage {
141                    addr: "/b".into(),
142                    args: vec![],
143                }),
144            ],
145        };
146
147        server
148            .dispatch(OscPacket::Bundle(bundle), "127.0.0.1:0".parse().unwrap())
149            .unwrap();
150
151        assert_eq!(count.load(std::sync::atomic::Ordering::SeqCst), 2);
152    }
153
154    #[test]
155    fn test_no_handler_match() {
156        let server = OscServer::bind("127.0.0.1:0");
157        let rt = tokio::runtime::Builder::new_current_thread()
158            .enable_io()
159            .build()
160            .unwrap();
161        let mut server = rt.block_on(server).unwrap();
162
163        server.handle("/audio/volume", |_msg, _src| {});
164
165        let msg = OscMessage {
166            addr: "/mixer/pan".into(),
167            args: vec![],
168        };
169
170        let result = server.dispatch(OscPacket::Message(msg), "127.0.0.1:0".parse().unwrap());
171        assert!(result.is_ok());
172    }
173
174    #[test]
175    fn test_set_buffer_size() {
176        let server = OscServer::bind("127.0.0.1:0");
177        let rt = tokio::runtime::Builder::new_current_thread()
178            .enable_io()
179            .build()
180            .unwrap();
181        let mut server = rt.block_on(server).unwrap();
182        server.set_buffer_size(1024);
183        assert_eq!(server.buffer_size, 1024);
184    }
185
186    #[test]
187    fn test_local_addr() {
188        let server = OscServer::bind("127.0.0.1:0");
189        let rt = tokio::runtime::Builder::new_current_thread()
190            .enable_io()
191            .build()
192            .unwrap();
193        let server = rt.block_on(server).unwrap();
194        let addr = server.local_addr().unwrap();
195        assert!(addr.port() > 0);
196    }
197}
rill_osc/server.rs

rill_osc/
server.rs
