| // Copyright 2018 The Fuchsia Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| //! An implementation of a client for a fidl interface. |
| |
| #[allow(unused_imports)] // for AsHandleRef (see https://github.com/rust-lang/rust/issues/53682) |
| use { |
| crate::{ |
| encoding::{ |
| Encodable, |
| Decodable, |
| Encoder, |
| Decoder, |
| decode_transaction_header, |
| TransactionHeader, |
| TransactionMessage |
| }, |
| Error, |
| }, |
| fuchsia_async::{ |
| self as fasync, |
| temp::{TempFutureExt, Either}, |
| }, |
| fuchsia_zircon::{self as zx, AsHandleRef}, |
| futures::{ |
| future::{self, Future, Ready, AndThen, TryFutureExt}, |
| ready, |
| stream::{FusedStream, Stream}, |
| task::{LocalWaker, Poll, Waker}, |
| }, |
| parking_lot::Mutex, |
| slab::Slab, |
| std::{ |
| collections::VecDeque, |
| marker::Unpin, |
| mem, |
| ops::Deref, |
| pin::Pin, |
| sync::Arc, |
| }, |
| }; |
| |
| /// Decode a new value of a decodable type from a transaction. |
| fn decode_transaction_body<D: Decodable>(mut buf: zx::MessageBuf) -> Result<D, Error> { |
| let (bytes, handles) = buf.split_mut(); |
| let header_len = <TransactionHeader as Decodable>::inline_size(); |
| if bytes.len() < header_len { return Err(Error::OutOfRange); } |
| let (_header_bytes, body_bytes) = bytes.split_at(header_len); |
| |
| let mut output = D::new_empty(); |
| Decoder::decode_into(body_bytes, handles, &mut output)?; |
| Ok(output) |
| } |
| |
| fn decode_transaction_body_fut<D: Decodable>(buf: zx::MessageBuf) -> Ready<Result<D, Error>> { |
| future::ready(decode_transaction_body(buf)) |
| } |
| |
| /// A FIDL client which can be used to send buffers and receive responses via a channel. |
| #[derive(Debug, Clone)] |
| pub struct Client { |
| inner: Arc<ClientInner>, |
| } |
| |
| /// A future representing the raw response to a FIDL query. |
| pub type RawQueryResponseFut = Either<Ready<Result<zx::MessageBuf, Error>>, MessageResponse>; |
| |
| /// A future representing the decoded response to a FIDL query. |
| pub type QueryResponseFut<D> = |
| AndThen< |
| RawQueryResponseFut, |
| Ready<Result<D, Error>>, |
| fn(zx::MessageBuf) -> Ready<Result<D, Error>>>; |
| |
| /// A FIDL transaction id. Will not be zero for a message that includes a response. |
| #[derive(Debug, Copy, Clone, PartialEq, Eq)] |
| pub struct Txid(u32); |
| /// A message interest id. |
| #[derive(Debug, Copy, Clone, PartialEq, Eq)] |
| struct InterestId(usize); |
| |
| impl InterestId { |
| fn from_txid(txid: Txid) -> Self { |
| InterestId(txid.0 as usize - 1) |
| } |
| |
| fn as_raw_id(&self) -> usize { |
| self.0 |
| } |
| } |
| |
| impl Txid { |
| fn from_interest_id(int_id: InterestId) -> Self { |
| Txid((int_id.0 + 1) as u32) |
| } |
| |
| fn as_raw_id(&self) -> u32 { |
| self.0 |
| } |
| } |
| |
| impl Client { |
| /// Create a new client. |
| /// |
| /// `channel` is the asynchronous channel over which data is sent and received. |
| /// `event_ordinals` are the ordinals on which events will be received. |
| pub fn new(channel: fasync::Channel) -> Client { |
| Client { |
| inner: Arc::new(ClientInner { |
| channel: channel, |
| message_interests: Mutex::new(Slab::<MessageInterest>::new()), |
| event_channel: Mutex::<EventChannel>::default(), |
| }) |
| } |
| } |
| |
| /// Attempt to convert the `Client` back into a channel. |
| /// |
| /// This will only succeed if there are no active clones of this `Client` |
| /// and no currently-alive `EventReceiver` or `MessageResponse`s that |
| /// came from this `Client`. |
| pub fn into_channel(self) -> Result<fasync::Channel, Self> { |
| match Arc::try_unwrap(self.inner) { |
| Ok(ClientInner { channel, .. }) => Ok(channel), |
| Err(inner) => Err(Self { inner }) |
| } |
| } |
| |
| /// Retrieve the stream of event messages for the `Client`. |
| /// Panics if the stream was already taken. |
| pub fn take_event_receiver(&self) -> EventReceiver { |
| { |
| let mut lock = self.inner.event_channel.lock(); |
| |
| if let EventListener::None = lock.listener { |
| lock.listener = EventListener::New; |
| } else { |
| panic!("Event stream was already taken"); |
| } |
| } |
| |
| EventReceiver { |
| inner: Some(self.inner.clone()), |
| } |
| } |
| |
| /// Send an encodable message without expecting a response. |
| pub fn send<T: Encodable>(&self, msg: &mut T, ordinal: u32) -> Result<(), Error> { |
| let (buf, handles) = (&mut vec![], &mut vec![]); |
| let msg = &mut TransactionMessage { |
| header: TransactionHeader { |
| tx_id: 0, |
| flags: 0, |
| ordinal, |
| }, |
| body: msg, |
| }; |
| Encoder::encode(buf, handles, msg)?; |
| self.send_raw_msg(&**buf, handles) |
| } |
| |
| /// Send an encodable query and receive a decodable response. |
| pub fn send_query<E: Encodable, D: Decodable>(&self, msg: &mut E, ordinal: u32) |
| -> QueryResponseFut<D> |
| { |
| let (buf, handles) = (&mut vec![], &mut vec![]); |
| let res_fut = self.send_raw_query(|tx_id| { |
| let msg = &mut TransactionMessage { |
| header: TransactionHeader { |
| tx_id: tx_id.as_raw_id(), |
| flags: 0, |
| ordinal, |
| }, |
| body: msg, |
| }; |
| Encoder::encode(buf, handles, msg)?; |
| Ok((buf, handles)) |
| }); |
| |
| res_fut.and_then(decode_transaction_body_fut::<D>) |
| } |
| |
| /// Send a raw message without expecting a response. |
| pub fn send_raw_msg(&self, buf: &[u8], handles: &mut Vec<zx::Handle>) -> Result<(), Error> { |
| Ok(self.inner.channel.write(buf, handles).map_err(Error::ClientWrite)?) |
| } |
| |
| /// Send a raw query and receive a response future. |
| pub fn send_raw_query<'a, F>(&'a self, msg_from_id: F) |
| -> RawQueryResponseFut |
| where F: FnOnce(Txid) -> Result<(&'a mut [u8], &'a mut Vec<zx::Handle>), Error> |
| { |
| let id = self.inner.register_msg_interest(); |
| let (out_buf, handles) = match msg_from_id(Txid::from_interest_id(id)) { |
| Ok(x) => x, |
| Err(e) => return future::ready(Err(e)).left_future(), |
| }; |
| if let Err(e) = self.inner.channel.write(out_buf, handles) { |
| return future::ready(Err(Error::ClientWrite(e))).left_future(); |
| } |
| |
| MessageResponse { |
| id: Txid::from_interest_id(id), |
| client: Some(self.inner.clone()), |
| }.right_future() |
| } |
| } |
| |
| #[must_use] |
| /// A future which polls for the response to a client message. |
| #[derive(Debug)] |
| pub struct MessageResponse { |
| id: Txid, |
| // `None` if the message response has been recieved |
| client: Option<Arc<ClientInner>>, |
| } |
| |
| impl Unpin for MessageResponse {} |
| |
| impl Future for MessageResponse { |
| type Output = Result<zx::MessageBuf, Error>; |
| fn poll(mut self: Pin<&mut Self>, lw: &LocalWaker) -> Poll<Self::Output> { |
| let this = &mut *self; |
| let res; |
| { |
| let client = this.client.as_ref().ok_or(Error::PollAfterCompletion)?; |
| res = client.poll_recv_msg_response(this.id, lw); |
| } |
| |
| // Drop the client reference if the response has been received |
| if let Poll::Ready(Ok(_)) = res { |
| let client = this.client.take() |
| .expect("MessageResponse polled after completion"); |
| client.wake_any(); |
| } |
| |
| res |
| } |
| } |
| |
| impl Drop for MessageResponse { |
| fn drop(&mut self) { |
| if let Some(client) = &self.client { |
| client.deregister_msg_interest(InterestId::from_txid(self.id)); |
| client.wake_any(); |
| } |
| } |
| } |
| |
| |
| /// An enum reprenting either a resolved message interest or a task on which to alert |
| /// that a response message has arrived. |
| #[derive(Debug)] |
| enum MessageInterest { |
| /// A new `MessageInterest` |
| WillPoll, |
| /// A task is waiting to receive a response, and can be awoken with `Waker`. |
| Waiting(Waker), |
| /// A message has been received, and a task will poll to receive it. |
| Received(zx::MessageBuf), |
| /// A message has not been received, but the person interested in the response |
| /// no longer cares about it, so the message should be discared upon arrival. |
| Discard, |
| } |
| |
| impl MessageInterest { |
| /// Check if a message has been received. |
| fn is_received(&self) -> bool { |
| if let MessageInterest::Received(_) = *self { |
| true |
| } else { |
| false |
| } |
| } |
| |
| fn unwrap_received(self) -> zx::MessageBuf { |
| if let MessageInterest::Received(buf) = self { |
| buf |
| } else { |
| panic!("EXPECTED received message") |
| } |
| } |
| } |
| |
| /// A stream of events as `MessageBuf`s. |
| #[derive(Debug)] |
| pub struct EventReceiver { |
| inner: Option<Arc<ClientInner>>, |
| } |
| |
| impl Unpin for EventReceiver {} |
| |
| impl FusedStream for EventReceiver { |
| fn is_terminated(&self) -> bool { |
| self.inner.is_none() |
| } |
| } |
| |
| impl Stream for EventReceiver { |
| type Item = Result<zx::MessageBuf, Error>; |
| |
| fn poll_next(mut self: Pin<&mut Self>, lw: &LocalWaker) -> Poll<Option<Self::Item>> { |
| let inner = self.inner.as_ref().expect("polled EventReceiver after `None`"); |
| Poll::Ready(match ready!(inner.poll_recv_event(lw)) { |
| Ok(x) => Some(Ok(x)), |
| Err(Error::ClientRead(zx::Status::PEER_CLOSED)) => { |
| self.inner = None; |
| None |
| }, |
| Err(e) => Some(Err(e)), |
| }) |
| } |
| } |
| |
| impl Drop for EventReceiver { |
| fn drop(&mut self) { |
| if let Some(inner) = &self.inner { |
| inner.event_channel.lock().listener = EventListener::None; |
| inner.wake_any(); |
| } |
| } |
| } |
| |
| #[derive(Debug, Default)] |
| struct EventChannel { |
| listener: EventListener, |
| queue: VecDeque<zx::MessageBuf>, |
| } |
| |
| #[derive(Debug)] |
| enum EventListener { |
| /// No one is listening for the event |
| None, |
| /// Someone is listening for the event but has not yet polled |
| New, |
| /// Someone is listening for the event and can be woken via the `Waker` |
| Some(Waker), |
| } |
| |
| impl Default for EventListener { |
| fn default() -> Self { EventListener::None } |
| } |
| |
| /// A shared client channel which tracks EXPECTED and received responses |
| #[derive(Debug)] |
| struct ClientInner { |
| channel: fasync::Channel, |
| |
| /// A map of message interests to either `None` (no message received yet) |
| /// or `Some(DecodeBuf)` when a message has been received. |
| /// An interest is registered with `register_msg_interest` and deregistered |
| /// by either receiving a message via a call to `poll_recv` or manually |
| /// deregistering with `deregister_msg_interest` |
| message_interests: Mutex<Slab<MessageInterest>>, |
| |
| /// A queue of received events and a waker for the task to receive them. |
| event_channel: Mutex<EventChannel>, |
| } |
| |
| impl Deref for Client { |
| type Target = fasync::Channel; |
| |
| fn deref(&self) -> &Self::Target { |
| &self.inner.channel |
| } |
| } |
| |
| impl ClientInner { |
| /// Registers interest in a response message. |
| /// |
| /// This function returns a `usize` ID which should be used to send a message |
| /// via the channel. Responses are then received using `poll_recv`. |
| fn register_msg_interest(&self) -> InterestId { |
| // TODO(cramertj) use `try_from` here and assert that the conversion from |
| // `usize` to `u32` hasn't overflowed. |
| InterestId(self.message_interests.lock().insert( |
| MessageInterest::WillPoll)) |
| } |
| |
| fn poll_recv_event( |
| &self, |
| lw: &LocalWaker, |
| ) -> Poll<Result<zx::MessageBuf, Error>> { |
| let is_closed = self.recv_all(lw)?; |
| |
| let mut lock = self.event_channel.lock(); |
| |
| if let Some(msg_buf) = lock.queue.pop_front() { |
| Poll::Ready(Ok(msg_buf)) |
| } else { |
| lock.listener = EventListener::Some(lw.clone().into_waker()); |
| if is_closed { |
| Poll::Ready(Err(Error::ClientRead(zx::Status::PEER_CLOSED))) |
| } else { |
| Poll::Pending |
| } |
| } |
| } |
| |
| fn poll_recv_msg_response( |
| &self, |
| txid: Txid, |
| lw: &LocalWaker, |
| ) -> Poll<Result<zx::MessageBuf, Error>> { |
| let is_closed = self.recv_all(lw)?; |
| |
| let mut message_interests = self.message_interests.lock(); |
| let interest_id = InterestId::from_txid(txid); |
| if message_interests.get(interest_id.as_raw_id()) |
| .expect("Polled unregistered interest") |
| .is_received() |
| { |
| // If, by happy accident, we just raced to getting the result, |
| // then yay! Return success. |
| let buf = message_interests.remove(interest_id.as_raw_id()).unwrap_received(); |
| Poll::Ready(Ok(buf)) |
| } else { |
| // Set the current waker to be notified when a response arrives. |
| *message_interests.get_mut(interest_id.as_raw_id()) |
| .expect("Polled unregistered interest") = |
| MessageInterest::Waiting(lw.clone().into_waker()); |
| |
| if is_closed { |
| Poll::Ready(Err(Error::ClientRead(zx::Status::PEER_CLOSED))) |
| } else { |
| Poll::Pending |
| } |
| } |
| } |
| |
| /// Poll for the receipt of any response message or an event. |
| /// |
| /// Returns whether or not the channel is closed. |
| fn recv_all( |
| &self, |
| lw: &LocalWaker, |
| ) -> Result<bool, Error> { |
| // TODO(cramertj) return errors if one has occured _ever_ in recv_all, not just if |
| // one happens on this call. |
| |
| loop { |
| let mut buf = zx::MessageBuf::new(); |
| match self.channel.recv_from(&mut buf, lw) { |
| Poll::Ready(Ok(())) => {} |
| Poll::Ready(Err(zx::Status::PEER_CLOSED)) => return Ok(true), |
| Poll::Ready(Err(e)) => return Err(Error::ClientRead(e)), |
| Poll::Pending => return Ok(false), |
| } |
| |
| let (header, _) = decode_transaction_header(buf.bytes()).map_err(|_| Error::InvalidHeader)?; |
| if header.tx_id == 0 { // received an event |
| let mut lock = self.event_channel.lock(); |
| lock.queue.push_back(buf); |
| if let EventListener::Some(ref waker) = lock.listener { waker.wake(); } |
| } else { // received a message response |
| let recvd_interest_id = InterestId::from_txid(Txid(header.tx_id)); |
| |
| // Look for a message interest with the given ID. |
| // If one is found, store the message so that it can be picked up later. |
| let mut message_interests = self.message_interests.lock(); |
| let raw_recvd_interest_id = recvd_interest_id.as_raw_id(); |
| if let Some(&MessageInterest::Discard) = |
| message_interests.get(raw_recvd_interest_id) |
| { |
| message_interests.remove(raw_recvd_interest_id); |
| } else if let Some(entry) = message_interests.get_mut(raw_recvd_interest_id) { |
| let old_entry = mem::replace(entry, MessageInterest::Received(buf)); |
| if let MessageInterest::Waiting(waker) = old_entry { |
| // Wake up the task to let them know a message has arrived. |
| waker.wake(); |
| } |
| } |
| } |
| } |
| } |
| |
| fn deregister_msg_interest(&self, InterestId(id): InterestId) { |
| let mut lock = self.message_interests.lock(); |
| if lock[id].is_received() { |
| lock.remove(id); |
| } else { |
| lock[id] = MessageInterest::Discard; |
| } |
| } |
| |
| // Wakes up an arbitrary task that has begun polling on the channel so that |
| // it will call recv_all and be registered as the new channel reader. |
| fn wake_any(&self) { |
| // Try to wake up message interests first, rather than the event listener. |
| // The event listener is a stream, and so could be between poll_nexts, |
| // blocked on a message interest completing before it begins polling again. |
| // Waking up only the event listener in these cases would cause a deadlock |
| // since the channel wouldn't be read from by the awoken event listener, |
| // so the message interest the event listener was blocked on would never |
| // complete. |
| // |
| // Message interests, however, should always be actively polled once |
| // they've begun being polled on a task. |
| { |
| let lock = self.message_interests.lock(); |
| for (_, message_interest) in lock.iter() { |
| if let MessageInterest::Waiting(waker) = message_interest { |
| waker.wake(); |
| return |
| } |
| } |
| } |
| { |
| let lock = self.event_channel.lock(); |
| if let EventListener::Some(waker) = &lock.listener { |
| waker.wake(); |
| return; |
| } |
| } |
| } |
| } |
| |
| pub mod sync { |
| //! Synchronous FIDL Client |
| use super::*; |
| |
| /// A synchronous client for making FIDL calls. |
| #[derive(Debug)] |
| pub struct Client { |
| // Underlying channel |
| channel: zx::Channel, |
| |
| // Reusable buffer for r/w |
| buf: zx::MessageBuf, |
| } |
| |
| // TODO: remove this and allow multiple overlapping queries on the same channel. |
| pub(super) const QUERY_TX_ID: u32 = 42; |
| |
| impl Client { |
| /// Create a new synchronous FIDL client. |
| pub fn new(channel: zx::Channel) -> Self { |
| Client { channel, buf: zx::MessageBuf::new() } |
| } |
| |
| /// Get the underlying channel out of the client. |
| pub fn into_channel(self) -> zx::Channel { |
| self.channel |
| } |
| |
| /// Send a new message. |
| pub fn send<E: Encodable>(&mut self, msg: &mut E, ordinal: u32) -> Result<(), Error> { |
| self.buf.clear(); |
| let (buf, handles) = self.buf.split_mut(); |
| let msg = &mut TransactionMessage { |
| header: TransactionHeader { |
| tx_id: 0, |
| flags: 0, |
| ordinal, |
| }, |
| body: msg, |
| }; |
| Encoder::encode(buf, handles, msg)?; |
| self.channel.write(buf, handles).map_err(Error::ClientWrite)?; |
| Ok(()) |
| } |
| |
| /// Send a new message expecting a response. |
| pub fn send_query<E: Encodable, D: Decodable>( |
| &mut self, |
| msg: &mut E, |
| ordinal: u32, |
| deadline: zx::Time, |
| ) -> Result<D, Error> { |
| // Write the message into the channel |
| self.buf.clear(); |
| let (buf, handles) = self.buf.split_mut(); |
| let msg = &mut TransactionMessage { |
| header: TransactionHeader { |
| tx_id: QUERY_TX_ID, |
| flags: 0, |
| ordinal, |
| }, |
| body: msg, |
| }; |
| Encoder::encode(buf, handles, msg)?; |
| self.channel.write(buf, handles).map_err(Error::ClientWrite)?; |
| |
| // Read the response |
| self.buf.clear(); |
| match self.channel.read(&mut self.buf) { |
| Ok(()) => {}, |
| Err(zx::Status::SHOULD_WAIT) => { |
| let signals = self.channel.wait_handle( |
| zx::Signals::CHANNEL_READABLE | zx::Signals::CHANNEL_PEER_CLOSED, |
| deadline, |
| ).map_err(Error::ClientRead)?; |
| if !signals.contains(zx::Signals::CHANNEL_READABLE) { |
| debug_assert!(signals.contains(zx::Signals::CHANNEL_PEER_CLOSED)); |
| return Err(Error::ClientRead(zx::Status::PEER_CLOSED)); |
| } |
| self.channel.read(&mut self.buf).map_err(Error::ClientRead)?; |
| }, |
| Err(e) => return Err(Error::ClientRead(e)), |
| } |
| let (buf, handles) = self.buf.split_mut(); |
| let (header, body_bytes) = decode_transaction_header(buf)?; |
| if header.tx_id != QUERY_TX_ID || header.ordinal != ordinal { |
| return Err(Error::UnexpectedSyncResponse); |
| } |
| let mut output = D::new_empty(); |
| Decoder::decode_into(body_bytes, handles, &mut output)?; |
| Ok(output) |
| } |
| } |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| use { |
| failure::{Error, ResultExt}, |
| futures::{FutureExt, StreamExt}, |
| fuchsia_async::TimeoutExt, |
| fuchsia_zircon::DurationNum, |
| std::{io, thread}, |
| }; |
| |
| const SEND_EXPECTED: &[u8] = &[ |
| 0, 0, 0, 0, 0, 0, 0, 0, // 32 bit tx_id followed by 32 bits of padding |
| 0, 0, 0, 0, // 32 bits for flags |
| SEND_ORDINAL_HIGH_BYTE, 0, 0, 0, // 32 bit ordinal |
| SEND_DATA, // 8 bit data |
| 0, 0, 0, 0, 0, 0, 0, // 7 bytes of padding after our 1 byte of data |
| ]; |
| const SEND_ORDINAL_HIGH_BYTE: u8 = 42; |
| const SEND_ORDINAL: u32 = 42; |
| const SEND_DATA: u8 = 55; |
| |
| #[test] |
| fn sync_client() -> Result<(), Error> { |
| let (client_end, server_end) = zx::Channel::create().context("chan create")?; |
| let mut client = sync::Client::new(client_end); |
| client.send(&mut SEND_DATA, SEND_ORDINAL).context("sending")?; |
| let mut received = zx::MessageBuf::new(); |
| server_end.read(&mut received).context("reading")?; |
| assert_eq!(SEND_EXPECTED, received.bytes()); |
| Ok(()) |
| } |
| |
| #[test] |
| fn sync_client_with_response() -> Result<(), Error> { |
| let (client_end, server_end) = zx::Channel::create().context("chan create")?; |
| let mut client = sync::Client::new(client_end); |
| thread::spawn(move || { |
| // Server |
| let mut received = zx::MessageBuf::new(); |
| server_end.wait_handle(zx::Signals::CHANNEL_READABLE, 5.seconds().after_now()) |
| .expect("failed to wait for channel readable"); |
| server_end.read(&mut received).expect("failed to read on server end"); |
| let (buf, handles) = received.split_mut(); |
| let (header, _body_bytes) = decode_transaction_header(buf).expect("server decode"); |
| assert_eq!(header.tx_id, sync::QUERY_TX_ID); |
| assert_eq!(header.ordinal, SEND_ORDINAL); |
| let response = &mut TransactionMessage { |
| header: TransactionHeader { |
| tx_id: header.tx_id, |
| flags: 0, |
| ordinal: header.ordinal, |
| }, |
| body: &mut SEND_DATA, |
| }; |
| Encoder::encode(buf, handles, response).expect("Encoding failure"); |
| server_end.write(buf, handles).expect("Server channel write failed"); |
| }); |
| let response_data = client.send_query::<u8, u8>( |
| &mut SEND_DATA, SEND_ORDINAL, 5.seconds().after_now(), |
| ).context("sending query")?; |
| assert_eq!(SEND_DATA, response_data); |
| Ok(()) |
| } |
| |
| #[test] |
| fn client() { |
| let mut executor = fasync::Executor::new().unwrap(); |
| let (client_end, server_end) = zx::Channel::create().unwrap(); |
| let client_end = fasync::Channel::from_channel(client_end).unwrap(); |
| let client = Client::new(client_end); |
| |
| let server = fasync::Channel::from_channel(server_end).unwrap(); |
| let receiver = async move { |
| let mut buffer = zx::MessageBuf::new(); |
| await!(server.recv_msg(&mut buffer)).expect("failed to recv msg"); |
| assert_eq!(SEND_EXPECTED, buffer.bytes()); |
| }; |
| |
| // add a timeout to receiver so if test is broken it doesn't take forever |
| let receiver = receiver.on_timeout( |
| 300.millis().after_now(), |
| || panic!("did not receive message in time!")); |
| |
| let sender = fasync::Timer::new(100.millis().after_now()).map(|()|{ |
| client.send(&mut SEND_DATA, SEND_ORDINAL).expect("failed to send msg"); |
| }); |
| |
| executor.run_singlethreaded(receiver.join(sender)); |
| } |
| |
| #[test] |
| fn client_with_response() { |
| const EXPECTED: &[u8] = &[ |
| 1, 0, 0, 0, 0, 0, 0, 0, // 32 bit tx_id followed by 32 bits of padding |
| 0, 0, 0, 0, // 32 bits for flags |
| 42, 0, 0, 0, // 32 bit ordinal |
| 55, // 8 bit data |
| 0, 0, 0, 0, 0, 0, 0, // 7 bytes of padding after our 1 byte of data |
| ]; |
| |
| let mut executor = fasync::Executor::new().unwrap(); |
| |
| let (client_end, server_end) = zx::Channel::create().unwrap(); |
| let client_end = fasync::Channel::from_channel(client_end).unwrap(); |
| let client = Client::new(client_end); |
| |
| let server = fasync::Channel::from_channel(server_end).unwrap(); |
| let mut buffer = zx::MessageBuf::new(); |
| let receiver = async move { |
| await!(server.recv_msg(&mut buffer)).expect("failed to recv msg"); |
| assert_eq!(EXPECTED, buffer.bytes()); |
| let id = 1; // internally, the first slot in a slab returns a `0`. We then add one |
| // since FIDL txids start with `1`. |
| |
| let response = &mut TransactionMessage { |
| header: TransactionHeader { |
| tx_id: id, |
| flags: 0, |
| ordinal: 42, |
| }, |
| body: &mut 55, |
| }; |
| |
| let (bytes, handles) = (&mut vec![], &mut vec![]); |
| Encoder::encode(bytes, handles, response).expect("Encoding failure"); |
| server.write(bytes, handles).expect("Server channel write failed"); |
| }; |
| |
| // add a timeout to receiver so if test is broken it doesn't take forever |
| let receiver = receiver.on_timeout( |
| 300.millis().after_now(), |
| || panic!("did not receiver message in time!" |
| )); |
| |
| let sender = client.send_query::<u8, u8>(&mut 55, 42) |
| .map_ok(|x|assert_eq!(x, 55)) |
| .map_err(|e| { |
| io::Error::new( |
| io::ErrorKind::Other, |
| &*format!("fidl error: {:?}", e)) |
| }); |
| |
| // add a timeout to receiver so if test is broken it doesn't take forever |
| let sender = sender.on_timeout( |
| 300.millis().after_now(), |
| || panic!("did not receive response in time!") |
| ); |
| |
| executor.run_singlethreaded(receiver.join(sender)); |
| } |
| |
| #[test] |
| #[should_panic] |
| fn event_cant_be_taken_twice() { |
| let _exec = fasync::Executor::new().unwrap(); |
| let (client_end, _) = zx::Channel::create().unwrap(); |
| let client_end = fasync::Channel::from_channel(client_end).unwrap(); |
| let client = Client::new(client_end); |
| let _foo = client.take_event_receiver(); |
| client.take_event_receiver(); |
| } |
| |
| #[test] |
| fn event_can_be_taken() { |
| let _exec = fasync::Executor::new().unwrap(); |
| let (client_end, _) = zx::Channel::create().unwrap(); |
| let client_end = fasync::Channel::from_channel(client_end).unwrap(); |
| let client = Client::new(client_end); |
| client.take_event_receiver(); |
| } |
| |
| #[test] |
| fn event_received() { |
| let mut executor = fasync::Executor::new().unwrap(); |
| |
| let (client_end, server_end) = zx::Channel::create().unwrap(); |
| let client_end = fasync::Channel::from_channel(client_end).unwrap(); |
| let client = Client::new(client_end); |
| |
| // Send the event from the server |
| let server = fasync::Channel::from_channel(server_end).unwrap(); |
| let event = &mut TransactionMessage { |
| header: TransactionHeader { |
| tx_id: 0, |
| flags: 0, |
| ordinal: 5, |
| }, |
| body: &mut 55i32, |
| }; |
| let (bytes, handles) = (&mut vec![], &mut vec![]); |
| Encoder::encode(bytes, handles, event).expect("Encoding failure"); |
| server.write(bytes, handles).expect("Server channel write failed"); |
| drop(server); |
| |
| let recv = client.take_event_receiver() |
| .into_future() |
| .then(|(x, stream)| { |
| let x = x.expect("should contain one element"); |
| let x = x.expect("fidl error"); |
| let x: i32 = decode_transaction_body(x).expect("failed to decode event"); |
| assert_eq!(x, 55); |
| stream.into_future() |
| }) |
| .map(|(x, _stream)| assert!(x.is_none(), "should have emptied")); |
| |
| // add a timeout to receiver so if test is broken it doesn't take forever |
| let recv = recv.on_timeout( |
| 300.millis().after_now(), |
| || panic!("did not receive event in time!") |
| ); |
| |
| executor.run_singlethreaded(recv); |
| } |
| } |
