1 use ffi;
2 use libc;
3 use super::Connection;
4
5 use std::mem;
6 use std::sync::{Mutex, RwLock};
7 use std::os::unix::io::{RawFd, AsRawFd};
8 use std::os::raw::{c_void, c_uint};
9
10 /// A file descriptor to watch for incoming events (for async I/O).
11 ///
12 /// # Example
13 /// ```
14 /// extern crate libc;
15 /// extern crate dbus;
16 /// fn main() {
17 /// use dbus::{Connection, BusType, WatchEvent};
18 /// let c = Connection::get_private(BusType::Session).unwrap();
19 ///
20 /// // Get a list of fds to poll for
21 /// let mut fds: Vec<_> = c.watch_fds().iter().map(|w| w.to_pollfd()).collect();
22 ///
23 /// // Poll them with a 1 s timeout
24 /// let r = unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as libc::c_ulong, 1000) };
25 /// assert!(r >= 0);
26 ///
27 /// // And handle incoming events
28 /// for pfd in fds.iter().filter(|pfd| pfd.revents != 0) {
29 /// for item in c.watch_handle(pfd.fd, WatchEvent::from_revents(pfd.revents)) {
30 /// // Handle item
31 /// println!("Received ConnectionItem: {:?}", item);
32 /// }
33 /// }
34 /// }
35 /// ```
36
37 #[repr(C)]
38 #[derive(Debug, PartialEq, Copy, Clone)]
39 /// The enum is here for backwards compatibility mostly.
40 ///
41 /// It should really be bitflags instead.
42 pub enum WatchEvent {
43 /// The fd is readable
44 Readable = ffi::DBUS_WATCH_READABLE as isize,
45 /// The fd is writable
46 Writable = ffi::DBUS_WATCH_WRITABLE as isize,
47 /// An error occured on the fd
48 Error = ffi::DBUS_WATCH_ERROR as isize,
49 /// The fd received a hangup.
50 Hangup = ffi::DBUS_WATCH_HANGUP as isize,
51 }
52
53 impl WatchEvent {
54 /// After running poll, this transforms the revents into a parameter you can send into `Connection::watch_handle`
from_revents(revents: libc::c_short) -> c_uint55 pub fn from_revents(revents: libc::c_short) -> c_uint {
56 0 +
57 if (revents & libc::POLLIN) != 0 { WatchEvent::Readable as c_uint } else { 0 } +
58 if (revents & libc::POLLOUT) != 0 { WatchEvent::Writable as c_uint } else { 0 } +
59 if (revents & libc::POLLERR) != 0 { WatchEvent::Error as c_uint } else { 0 } +
60 if (revents & libc::POLLHUP) != 0 { WatchEvent::Hangup as c_uint } else { 0 }
61 }
62 }
63
64 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
65 /// A file descriptor, and an indication whether it should be read from, written to, or both.
66 pub struct Watch {
67 fd: RawFd,
68 read: bool,
69 write: bool,
70 }
71
72 impl Watch {
73 /// Get the RawFd this Watch is for
fd(&self) -> RawFd74 pub fn fd(&self) -> RawFd { self.fd }
75 /// Add POLLIN to events to listen for
readable(&self) -> bool76 pub fn readable(&self) -> bool { self.read }
77 /// Add POLLOUT to events to listen for
writable(&self) -> bool78 pub fn writable(&self) -> bool { self.write }
79 /// Returns the current watch as a libc::pollfd, to use with libc::poll
to_pollfd(&self) -> libc::pollfd80 pub fn to_pollfd(&self) -> libc::pollfd {
81 libc::pollfd { fd: self.fd, revents: 0, events: libc::POLLERR + libc::POLLHUP +
82 if self.readable() { libc::POLLIN } else { 0 } +
83 if self.writable() { libc::POLLOUT } else { 0 },
84 }
85 }
86 }
87
88 impl AsRawFd for Watch {
as_raw_fd(&self) -> RawFd89 fn as_raw_fd(&self) -> RawFd { self.fd }
90 }
91
92 /// Note - internal struct, not to be used outside API. Moving it outside its box will break things.
93 pub struct WatchList {
94 watches: RwLock<Vec<*mut ffi::DBusWatch>>,
95 enabled_fds: Mutex<Vec<Watch>>,
96 on_update: Mutex<Box<Fn(Watch) + Send>>,
97 }
98
99 impl WatchList {
new(c: &Connection, on_update: Box<Fn(Watch) + Send>) -> Box<WatchList>100 pub fn new(c: &Connection, on_update: Box<Fn(Watch) + Send>) -> Box<WatchList> {
101 let w = Box::new(WatchList { on_update: Mutex::new(on_update), watches: RwLock::new(vec!()), enabled_fds: Mutex::new(vec!()) });
102 if unsafe { ffi::dbus_connection_set_watch_functions(super::connection::conn_handle(c),
103 Some(add_watch_cb), Some(remove_watch_cb), Some(toggled_watch_cb), &*w as *const _ as *mut _, None) } == 0 {
104 panic!("dbus_connection_set_watch_functions failed");
105 }
106 w
107 }
108
set_on_update(&self, on_update: Box<Fn(Watch) + Send>)109 pub fn set_on_update(&self, on_update: Box<Fn(Watch) + Send>) { *self.on_update.lock().unwrap() = on_update; }
110
watch_handle(&self, fd: RawFd, flags: c_uint)111 pub fn watch_handle(&self, fd: RawFd, flags: c_uint) {
112 // println!("watch_handle {} flags {}", fd, flags);
113 for &q in self.watches.read().unwrap().iter() {
114 let w = self.get_watch(q);
115 if w.fd != fd { continue };
116 if unsafe { ffi::dbus_watch_handle(q, flags) } == 0 {
117 panic!("dbus_watch_handle failed");
118 }
119 self.update(q);
120 };
121 }
122
get_enabled_fds(&self) -> Vec<Watch>123 pub fn get_enabled_fds(&self) -> Vec<Watch> {
124 self.enabled_fds.lock().unwrap().clone()
125 }
126
get_watch(&self, watch: *mut ffi::DBusWatch) -> Watch127 fn get_watch(&self, watch: *mut ffi::DBusWatch) -> Watch {
128 let mut w = Watch { fd: unsafe { ffi::dbus_watch_get_unix_fd(watch) }, read: false, write: false};
129 let enabled = self.watches.read().unwrap().contains(&watch) && unsafe { ffi::dbus_watch_get_enabled(watch) != 0 };
130 let flags = unsafe { ffi::dbus_watch_get_flags(watch) };
131 if enabled {
132 w.read = (flags & WatchEvent::Readable as c_uint) != 0;
133 w.write = (flags & WatchEvent::Writable as c_uint) != 0;
134 }
135 // println!("Get watch fd {:?} ptr {:?} enabled {:?} flags {:?}", w, watch, enabled, flags);
136 w
137 }
138
update(&self, watch: *mut ffi::DBusWatch)139 fn update(&self, watch: *mut ffi::DBusWatch) {
140 let mut w = self.get_watch(watch);
141
142 for &q in self.watches.read().unwrap().iter() {
143 if q == watch { continue };
144 let ww = self.get_watch(q);
145 if ww.fd != w.fd { continue };
146 w.read |= ww.read;
147 w.write |= ww.write;
148 }
149 // println!("Updated sum: {:?}", w);
150
151 {
152 let mut fdarr = self.enabled_fds.lock().unwrap();
153
154 if w.write || w.read {
155 if fdarr.contains(&w) { return; } // Nothing changed
156 }
157 else if !fdarr.iter().any(|q| w.fd == q.fd) { return; } // Nothing changed
158
159 fdarr.retain(|f| f.fd != w.fd);
160 if w.write || w.read { fdarr.push(w) };
161 }
162 let func = self.on_update.lock().unwrap();
163 (*func)(w);
164 }
165 }
166
add_watch_cb(watch: *mut ffi::DBusWatch, data: *mut c_void) -> u32167 extern "C" fn add_watch_cb(watch: *mut ffi::DBusWatch, data: *mut c_void) -> u32 {
168 let wlist: &WatchList = unsafe { mem::transmute(data) };
169 // println!("Add watch {:?}", watch);
170 wlist.watches.write().unwrap().push(watch);
171 wlist.update(watch);
172 1
173 }
174
remove_watch_cb(watch: *mut ffi::DBusWatch, data: *mut c_void)175 extern "C" fn remove_watch_cb(watch: *mut ffi::DBusWatch, data: *mut c_void) {
176 let wlist: &WatchList = unsafe { mem::transmute(data) };
177 // println!("Removed watch {:?}", watch);
178 wlist.watches.write().unwrap().retain(|w| *w != watch);
179 wlist.update(watch);
180 }
181
toggled_watch_cb(watch: *mut ffi::DBusWatch, data: *mut c_void)182 extern "C" fn toggled_watch_cb(watch: *mut ffi::DBusWatch, data: *mut c_void) {
183 let wlist: &WatchList = unsafe { mem::transmute(data) };
184 // println!("Toggled watch {:?}", watch);
185 wlist.update(watch);
186 }
187
188 #[cfg(test)]
189 mod test {
190 use libc;
191 use super::super::{Connection, Message, BusType, WatchEvent, ConnectionItem, MessageType};
192
193 #[test]
async()194 fn async() {
195 let c = Connection::get_private(BusType::Session).unwrap();
196 c.register_object_path("/test").unwrap();
197 let m = Message::new_method_call(&c.unique_name(), "/test", "com.example.asynctest", "AsyncTest").unwrap();
198 let serial = c.send(m).unwrap();
199 println!("Async: sent serial {}", serial);
200
201 let mut fds: Vec<_> = c.watch_fds().iter().map(|w| w.to_pollfd()).collect();
202 let mut new_fds = None;
203 let mut i = 0;
204 let mut success = false;
205 while !success {
206 i += 1;
207 if let Some(q) = new_fds { fds = q; new_fds = None };
208
209 for f in fds.iter_mut() { f.revents = 0 };
210 assert!(unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as libc::c_ulong, 1000) } > 0);
211
212 for f in fds.iter().filter(|pfd| pfd.revents != 0) {
213 let m = WatchEvent::from_revents(f.revents);
214 println!("Async: fd {}, revents {} -> {}", f.fd, f.revents, m);
215 assert!(f.revents & libc::POLLIN != 0 || f.revents & libc::POLLOUT != 0);
216
217 for e in c.watch_handle(f.fd, m) {
218 println!("Async: got {:?}", e);
219 match e {
220 ConnectionItem::MethodCall(m) => {
221 assert_eq!(m.headers(), (MessageType::MethodCall, Some("/test".to_string()),
222 Some("com.example.asynctest".into()), Some("AsyncTest".to_string())));
223 let mut mr = Message::new_method_return(&m).unwrap();
224 mr.append_items(&["Goodies".into()]);
225 c.send(mr).unwrap();
226 }
227 ConnectionItem::MethodReturn(m) => {
228 assert_eq!(m.headers().0, MessageType::MethodReturn);
229 assert_eq!(m.get_reply_serial().unwrap(), serial);
230 let i = m.get_items();
231 let s: &str = i[0].inner().unwrap();
232 assert_eq!(s, "Goodies");
233 success = true;
234 }
235 _ => (),
236 }
237 }
238 if i > 100 { panic!() };
239 }
240 }
241 }
242 }
243