Bug 1440538 - P2: Update audioipc to commit 933fb48. r=kinetik
authorDan Glastonbury <dan.glastonbury@gmail.com>
Tue, 13 Feb 2018 14:39:06 +1000
changeset 457905 fb7e2f2bea092f1ddc4e20ab2d9d64cf34fc27f8
parent 457904 34deadb4772f2f134feadb1d83855d2fba90694b
child 457906 6d9344794eb47976a0827984d76aefc2f899cdd7
push id8799
push usermtabara@mozilla.com
push dateThu, 01 Mar 2018 16:46:23 +0000
treeherdermozilla-beta@15334014dc67 [default view] [failures only]
perfherder[talos] [build metrics] [platform microbench] (compared to previous push)
reviewerskinetik
bugs1440538
milestone60.0a1
first release with
nightly linux32
nightly linux64
nightly mac
nightly win32
nightly win64
last release without
nightly linux32
nightly linux64
nightly mac
nightly win32
nightly win64
Bug 1440538 - P2: Update audioipc to commit 933fb48. r=kinetik MozReview-Commit-ID: 29VUZKxz3xR
media/audioipc/README_MOZILLA
media/audioipc/audioipc/Cargo.toml
media/audioipc/audioipc/src/async.rs
media/audioipc/audioipc/src/cmsg.rs
media/audioipc/audioipc/src/codec.rs
media/audioipc/audioipc/src/core.rs
media/audioipc/audioipc/src/errors.rs
media/audioipc/audioipc/src/fd_passing.rs
media/audioipc/audioipc/src/frame.rs
media/audioipc/audioipc/src/lib.rs
media/audioipc/audioipc/src/messages.rs
media/audioipc/audioipc/src/msg.rs
media/audioipc/audioipc/src/rpc/client/mod.rs
media/audioipc/audioipc/src/rpc/client/proxy.rs
media/audioipc/audioipc/src/rpc/driver.rs
media/audioipc/audioipc/src/rpc/server.rs
media/audioipc/audioipc/src/shm.rs
media/audioipc/client/Cargo.toml
media/audioipc/client/src/context.rs
media/audioipc/client/src/lib.rs
media/audioipc/client/src/send_recv.rs
media/audioipc/client/src/stream.rs
media/audioipc/gecko.patch
media/audioipc/server/Cargo.toml
media/audioipc/server/src/lib.rs
--- a/media/audioipc/README_MOZILLA
+++ b/media/audioipc/README_MOZILLA
@@ -1,8 +1,8 @@
 The source from this directory was copied from the audioipc-2
 git repository using the update.sh script.  The only changes
 made were those applied by update.sh and the addition of
 Makefile.in build files for the Mozilla build system.
 
 The audioipc-2 git repository is: https://github.com/djg/audioipc-2.git
 
-The git commit ID used was d7798606aa590ef402344b7a519a0053725a9805 (2018-01-27 09:07:03 +1000)
+The git commit ID used was 933fb48b252a10569ba8d598541577c6f2dc308f (2018-02-21 17:13:04 +1000)
--- a/media/audioipc/audioipc/Cargo.toml
+++ b/media/audioipc/audioipc/Cargo.toml
@@ -3,17 +3,17 @@ name = "audioipc"
 version = "0.2.1"
 authors = [
         "Matthew Gregan <kinetik@flim.org>",
         "Dan Glastonbury <dan.glastonbury@gmail.com>"
         ]
 description = "Remote Cubeb IPC"
 
 [dependencies]
-cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
+cubeb = "0.4"
 bincode = "0.8"
 bytes = "0.4"
 # rayon-core in Gecko uses futures 0.1.13
 futures = "=0.1.13"
 iovec = "0.1"
 libc = "0.2"
 log = "^0.3.6"
 memmap = "0.5.2"
--- a/media/audioipc/audioipc/src/async.rs
+++ b/media/audioipc/audioipc/src/async.rs
@@ -108,21 +108,21 @@ impl AsyncRecvMsg for UnixStream {
             Ok((n, cmsg_len, flags)) => {
                 unsafe {
                     buf.advance_mut(n);
                 }
                 unsafe {
                     cmsg.advance_mut(cmsg_len);
                 }
                 Ok((n, flags).into())
-            },
+            }
             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                 self.need_read();
                 Ok(Async::NotReady)
-            },
+            }
             Err(e) => Err(e),
         }
     }
 }
 
 impl AsyncSendMsg for UnixStream {
     fn send_msg_buf<B, C>(&mut self, buf: &mut B, cmsg: &C) -> Poll<usize, io::Error>
     where
@@ -134,41 +134,26 @@ impl AsyncSendMsg for UnixStream {
         }
         let r = {
             // The `IoVec` type can't have a zero-length size, so create a dummy
             // version from a 1-length slice which we'll overwrite with the
             // `bytes_vec` method.
             static DUMMY: &[u8] = &[0];
             let nom = <&IoVec>::from(DUMMY);
             let mut bufs = [
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
-                nom,
+                nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom, nom
             ];
             let n = buf.bytes_vec(&mut bufs);
             self.send_msg(&bufs[..n], cmsg.bytes())
         };
         match r {
             Ok(n) => {
                 buf.advance(n);
                 Ok(Async::Ready(n))
-            },
+            }
             Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                 self.need_write();
                 Ok(Async::NotReady)
-            },
+            }
             Err(e) => Err(e),
         }
     }
 }
--- a/media/audioipc/audioipc/src/cmsg.rs
+++ b/media/audioipc/audioipc/src/cmsg.rs
@@ -5,17 +5,17 @@
 
 use bytes::{BufMut, Bytes, BytesMut};
 use libc::{self, cmsghdr};
 use std::{convert, mem, ops, slice};
 use std::os::unix::io::RawFd;
 
 #[derive(Clone, Debug)]
 pub struct Fds {
-    fds: Bytes
+    fds: Bytes,
 }
 
 impl convert::AsRef<[RawFd]> for Fds {
     fn as_ref(&self) -> &[RawFd] {
         let n = self.fds.len() / mem::size_of::<RawFd>();
         unsafe { slice::from_raw_parts(self.fds.as_ptr() as *const _, n) }
     }
 }
@@ -25,23 +25,21 @@ impl ops::Deref for Fds {
 
     #[inline]
     fn deref(&self) -> &[RawFd] {
         self.as_ref()
     }
 }
 
 pub struct ControlMsgIter {
-    control: Bytes
+    control: Bytes,
 }
 
 pub fn iterator(c: Bytes) -> ControlMsgIter {
-    ControlMsgIter {
-        control: c
-    }
+    ControlMsgIter { control: c }
 }
 
 impl Iterator for ControlMsgIter {
     type Item = Fds;
 
     // This follows the logic in __cmsg_nxthdr from glibc
     // /usr/include/bits/socket.h
     fn next(&mut self) -> Option<Self::Item> {
@@ -68,60 +66,60 @@ impl Iterator for ControlMsgIter {
             } else {
                 control.slice_from(next_cmsghdr)
             };
 
             match (cmsg.cmsg_level, cmsg.cmsg_type) {
                 (libc::SOL_SOCKET, libc::SCM_RIGHTS) => {
                     trace!("Found SCM_RIGHTS...");
                     return Some(Fds {
-                        fds: control.slice(cmsghdr_len, cmsg_len as _)
+                        fds: control.slice(cmsghdr_len, cmsg_len as _),
                     });
-                },
+                }
                 (level, kind) => {
                     trace!("Skipping cmsg level, {}, type={}...", level, kind);
-                },
+                }
             }
         }
     }
 }
 
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub enum Error {
     /// Not enough space in storage to insert control mesage.
-    NoSpace
+    NoSpace,
 }
 
 #[must_use]
 pub struct ControlMsgBuilder {
-    result: Result<BytesMut, Error>
+    result: Result<BytesMut, Error>,
 }
 
 pub fn builder(buf: &mut BytesMut) -> ControlMsgBuilder {
     let buf = aligned(buf);
-    ControlMsgBuilder {
-        result: Ok(buf)
-    }
+    ControlMsgBuilder { result: Ok(buf) }
 }
 
 impl ControlMsgBuilder {
     fn msg(mut self, level: libc::c_int, kind: libc::c_int, msg: &[u8]) -> Self {
         self.result = self.result.and_then(align_buf).and_then(|mut cmsg| {
             let cmsg_len = len(msg.len());
             if cmsg.remaining_mut() < cmsg_len {
                 return Err(Error::NoSpace);
             }
 
             let cmsghdr = cmsghdr {
                 cmsg_len: cmsg_len as _,
                 cmsg_level: level,
-                cmsg_type: kind
+                cmsg_type: kind,
             };
 
-            let cmsghdr = unsafe { slice::from_raw_parts(&cmsghdr as *const _ as *const _, mem::size_of::<cmsghdr>()) };
+            let cmsghdr = unsafe {
+                slice::from_raw_parts(&cmsghdr as *const _ as *const _, mem::size_of::<cmsghdr>())
+            };
             cmsg.put_slice(cmsghdr);
             let mut cmsg = try!(align_buf(cmsg));
             cmsg.put_slice(msg);
 
             Ok(cmsg)
         });
 
         self
--- a/media/audioipc/audioipc/src/codec.rs
+++ b/media/audioipc/audioipc/src/codec.rs
@@ -30,48 +30,48 @@ pub trait Codec {
 
     /// A default method available to be called when there are no more bytes
     /// available to be read from the I/O.
     fn decode_eof(&mut self, buf: &mut BytesMut) -> io::Result<Self::Out> {
         match try!(self.decode(buf)) {
             Some(frame) => Ok(frame),
             None => Err(io::Error::new(
                 io::ErrorKind::Other,
-                "bytes remaining on stream"
-            ))
+                "bytes remaining on stream",
+            )),
         }
     }
 
     /// Encodes a frame intox the buffer provided.
     fn encode(&mut self, msg: Self::In, buf: &mut BytesMut) -> io::Result<()>;
 }
 
 /// Codec based upon bincode serialization
 ///
 /// Messages that have been serialized using bincode are prefixed with
 /// the length of the message to aid in deserialization, so that it's
 /// known if enough data has been received to decode a complete
 /// message.
 pub struct LengthDelimitedCodec<In, Out> {
     state: State,
     __in: PhantomData<In>,
-    __out: PhantomData<Out>
+    __out: PhantomData<Out>,
 }
 
 enum State {
     Length,
-    Data(u16)
+    Data(u16),
 }
 
 impl<In, Out> Default for LengthDelimitedCodec<In, Out> {
     fn default() -> Self {
         LengthDelimitedCodec {
             state: State::Length,
             __in: PhantomData,
-            __out: PhantomData
+            __out: PhantomData,
         }
     }
 }
 
 impl<In, Out> LengthDelimitedCodec<In, Out> {
     // Lengths are encoded as little endian u16
     fn decode_length(&mut self, buf: &mut BytesMut) -> io::Result<Option<u16>> {
         if buf.len() < 2 {
@@ -84,97 +84,97 @@ impl<In, Out> LengthDelimitedCodec<In, O
         // Consume the length field
         let _ = buf.split_to(2);
 
         Ok(Some(n))
     }
 
     fn decode_data(&mut self, buf: &mut BytesMut, n: u16) -> io::Result<Option<Out>>
     where
-        Out: DeserializeOwned + Debug
+        Out: DeserializeOwned + Debug,
     {
         // At this point, the buffer has already had the required capacity
         // reserved. All there is to do is read.
         let n = n as usize;
         if buf.len() < n {
             return Ok(None);
         }
 
         let buf = buf.split_to(n).freeze();
 
         trace!("Attempting to decode");
         let msg = try!(deserialize::<Out>(buf.as_ref()).map_err(|e| match *e {
             bincode::ErrorKind::IoError(e) => e,
-            _ => io::Error::new(io::ErrorKind::Other, *e)
+            _ => io::Error::new(io::ErrorKind::Other, *e),
         }));
 
         trace!("... Decoded {:?}", msg);
         Ok(Some(msg))
     }
 }
 
 impl<In, Out> Codec for LengthDelimitedCodec<In, Out>
 where
     In: Serialize + Debug,
-    Out: DeserializeOwned + Debug
+    Out: DeserializeOwned + Debug,
 {
     type In = In;
     type Out = Out;
 
     fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Out>> {
         let n = match self.state {
             State::Length => {
                 match try!(self.decode_length(buf)) {
                     Some(n) => {
                         self.state = State::Data(n);
 
                         // Ensure that the buffer has enough space to read the
                         // incoming payload
                         buf.reserve(n as usize);
 
                         n
-                    },
-                    None => return Ok(None)
+                    }
+                    None => return Ok(None),
                 }
-            },
-            State::Data(n) => n
+            }
+            State::Data(n) => n,
         };
 
         match try!(self.decode_data(buf, n)) {
             Some(data) => {
                 // Update the decode state
                 self.state = State::Length;
 
                 // Make sure the buffer has enough space to read the next head
                 buf.reserve(2);
 
                 Ok(Some(data))
-            },
-            None => Ok(None)
+            }
+            None => Ok(None),
         }
     }
 
     fn encode(&mut self, item: Self::In, buf: &mut BytesMut) -> io::Result<()> {
         trace!("Attempting to encode");
         let encoded_len = serialized_size(&item);
         if encoded_len > 8 * 1024 {
             return Err(io::Error::new(
                 io::ErrorKind::InvalidInput,
-                "encoded message too big"
+                "encoded message too big",
             ));
         }
 
         buf.reserve((encoded_len + 2) as usize);
 
         buf.put_u16::<LittleEndian>(encoded_len as u16);
 
         if let Err(e) =
             serialize_into::<_, Self::In, _>(&mut buf.writer(), &item, Bounded(encoded_len))
         {
             match *e {
                 bincode::ErrorKind::IoError(e) => return Err(e),
-                _ => return Err(io::Error::new(io::ErrorKind::Other, *e))
+                _ => return Err(io::Error::new(io::ErrorKind::Other, *e)),
             }
         }
 
         Ok(())
     }
 }
--- a/media/audioipc/audioipc/src/core.rs
+++ b/media/audioipc/audioipc/src/core.rs
@@ -11,37 +11,37 @@ scoped_thread_local! {
 }
 
 pub fn handle() -> Handle {
     HANDLE.with(|handle| handle.clone())
 }
 
 pub fn spawn<F>(f: F)
 where
-    F: Future<Item = (), Error = ()> + 'static
+    F: Future<Item = (), Error = ()> + 'static,
 {
     HANDLE.with(|handle| handle.spawn(f))
 }
 
 pub fn spawn_fn<F, R>(f: F)
 where
     F: FnOnce() -> R + 'static,
-    R: IntoFuture<Item = (), Error = ()> + 'static
+    R: IntoFuture<Item = (), Error = ()> + 'static,
 {
     HANDLE.with(|handle| handle.spawn_fn(f))
 }
 
 struct Inner {
     join: thread::JoinHandle<()>,
-    shutdown: oneshot::Sender<()>
+    shutdown: oneshot::Sender<()>,
 }
 
 pub struct CoreThread {
     inner: Option<Inner>,
-    remote: Remote
+    remote: Remote,
 }
 
 impl CoreThread {
     pub fn remote(&self) -> Remote {
         self.remote.clone()
     }
 }
 
@@ -60,17 +60,17 @@ impl fmt::Debug for CoreThread {
         // f.debug_tuple("CoreThread").field(&"...").finish()
         f.debug_tuple("CoreThread").field(&self.remote).finish()
     }
 }
 
 pub fn spawn_thread<S, F>(name: S, f: F) -> io::Result<CoreThread>
 where
     S: Into<String>,
-    F: FnOnce() -> io::Result<()> + Send + 'static
+    F: FnOnce() -> io::Result<()> + Send + 'static,
 {
     let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
     let (remote_tx, remote_rx) = mpsc::channel::<Remote>();
 
     let join = try!(thread::Builder::new().name(name.into()).spawn(move || {
         let mut core = Core::new().expect("Failed to create reactor::Core");
         let handle = core.handle();
         let remote = handle.remote().clone();
@@ -80,23 +80,21 @@ where
             f().and_then(|_| {
                 let _ = core.run(shutdown_rx);
                 Ok(())
             })
         }));
         trace!("thread shutdown...");
     }));
 
-    let remote = try!(remote_rx.recv().or_else(|_| {
-        Err(io::Error::new(
-            io::ErrorKind::Other,
-            "Failed to receive remote handle from spawned thread"
-        ))
-    }));
+    let remote = try!(remote_rx.recv().or_else(|_| Err(io::Error::new(
+        io::ErrorKind::Other,
+        "Failed to receive remote handle from spawned thread"
+    ))));
 
     Ok(CoreThread {
         inner: Some(Inner {
             join: join,
-            shutdown: shutdown_tx
+            shutdown: shutdown_tx,
         }),
-        remote: remote
+        remote: remote,
     })
 }
--- a/media/audioipc/audioipc/src/errors.rs
+++ b/media/audioipc/audioipc/src/errors.rs
@@ -1,17 +1,17 @@
 use bincode;
-use cubeb_core;
+use cubeb;
 use std;
 
 error_chain! {
     // Maybe replace with chain_err to improve the error info.
     foreign_links {
         Bincode(bincode::Error);
         Io(std::io::Error);
-        Cubeb(cubeb_core::Error);
+        Cubeb(cubeb::Error);
     }
 
     // Replace bail!(str) with explicit errors.
     errors {
         Disconnected
     }
 }
--- a/media/audioipc/audioipc/src/fd_passing.rs
+++ b/media/audioipc/audioipc/src/fd_passing.rs
@@ -15,25 +15,25 @@ use std::collections::VecDeque;
 use std::os::unix::io::RawFd;
 
 const INITIAL_CAPACITY: usize = 1024;
 const BACKPRESSURE_THRESHOLD: usize = 4 * INITIAL_CAPACITY;
 const FDS_CAPACITY: usize = 16;
 
 struct IncomingFds {
     cmsg: BytesMut,
-    recv_fds: Option<cmsg::ControlMsgIter>
+    recv_fds: Option<cmsg::ControlMsgIter>,
 }
 
 impl IncomingFds {
     pub fn new(c: usize) -> Self {
         let capacity = c * cmsg::space(mem::size_of::<[RawFd; 3]>());
         IncomingFds {
             cmsg: BytesMut::with_capacity(capacity),
-            recv_fds: None
+            recv_fds: None,
         }
     }
 
     pub fn take_fds(&mut self) -> Option<[RawFd; 3]> {
         loop {
             let fds = self.recv_fds
                 .as_mut()
                 .and_then(|recv_fds| recv_fds.next())
@@ -55,38 +55,38 @@ impl IncomingFds {
         self.cmsg.reserve(cmsg::space(mem::size_of::<[RawFd; 3]>()));
         &mut self.cmsg
     }
 }
 
 #[derive(Debug)]
 struct Frame {
     msgs: Bytes,
-    fds: Option<Bytes>
+    fds: Option<Bytes>,
 }
 
 /// A unified `Stream` and `Sink` interface over an I/O object, using
 /// the `Codec` trait to encode and decode the payload.
 pub struct FramedWithFds<A, C> {
     io: A,
     codec: C,
     // Stream
     read_buf: BytesMut,
     incoming_fds: IncomingFds,
     is_readable: bool,
     eof: bool,
     // Sink
     frames: VecDeque<Frame>,
     write_buf: BytesMut,
-    outgoing_fds: BytesMut
+    outgoing_fds: BytesMut,
 }
 
 impl<A, C> FramedWithFds<A, C>
 where
-    A: AsyncSendMsg
+    A: AsyncSendMsg,
 {
     // If there is a buffered frame, try to write it to `A`
     fn do_write(&mut self) -> Poll<(), io::Error> {
         debug!("do_write...");
         // Create a frame from any pending message in `write_buf`.
         if !self.write_buf.is_empty() {
             self.set_frame(None);
         }
@@ -97,20 +97,20 @@ where
 
         loop {
             let n = match self.frames.front() {
                 Some(frame) => {
                     trace!("sending msg {:?}, fds {:?}", frame.msgs, frame.fds);
                     let mut msgs = frame.msgs.clone().into_buf();
                     let mut fds = match frame.fds {
                         Some(ref fds) => fds.clone(),
-                        None => Bytes::new()
+                        None => Bytes::new(),
                     }.into_buf();
                     try_ready!(self.io.send_msg_buf(&mut msgs, &fds))
-                },
+                }
                 _ => {
                     // No pending frames.
                     return Ok(().into());
                 }
             };
 
             match self.frames.pop_front() {
                 Some(mut frame) => {
@@ -132,18 +132,18 @@ where
                         // re-queue the remaining message at the head
                         // of the queue. (Don't need to resend the fds
                         // since they've been sent with the first
                         // part.)
                         drop(frame.msgs.split_to(n));
                         self.frames.push_front(frame);
                         break;
                     }
-                },
-                _ => panic!()
+                }
+                _ => panic!(),
             }
         }
         debug!("process {} frames", processed);
 
         trace!("pending frames: {:?}", self.frames);
 
         Ok(().into())
     }
@@ -153,28 +153,25 @@ where
             assert!(fds.is_none());
             trace!("set_frame: No pending messages...");
             return;
         }
 
         let msgs = self.write_buf.take().freeze();
         trace!("set_frame: msgs={:?} fds={:?}", msgs, fds);
 
-        self.frames.push_back(Frame {
-            msgs,
-            fds
-        });
+        self.frames.push_back(Frame { msgs, fds });
     }
 }
 
 impl<A, C> Stream for FramedWithFds<A, C>
 where
     A: AsyncRecvMsg,
     C: Codec,
-    C::Out: AssocRawFd
+    C::Out: AssocRawFd,
 {
     type Item = C::Out;
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
         loop {
             // Repeatedly call `decode` or `decode_eof` as long as it is
             // "readable". Readable is defined as not having returned `None`. If
@@ -221,25 +218,22 @@ where
         }
     }
 }
 
 impl<A, C> Sink for FramedWithFds<A, C>
 where
     A: AsyncSendMsg,
     C: Codec,
-    C::In: AssocRawFd + fmt::Debug
+    C::In: AssocRawFd + fmt::Debug,
 {
     type SinkItem = C::In;
     type SinkError = io::Error;
 
-    fn start_send(
-        &mut self,
-        item: Self::SinkItem
-    ) -> StartSend<Self::SinkItem, Self::SinkError> {
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
         trace!("start_send: item={:?}", item);
 
         // If the buffer is already over BACKPRESSURE_THRESHOLD,
         // then attempt to flush it. If after flush it's *still*
         // over BACKPRESSURE_THRESHOLD, then reject the send.
         if self.write_buf.len() > BACKPRESSURE_THRESHOLD {
             try!(self.poll_complete());
             if self.write_buf.len() > BACKPRESSURE_THRESHOLD {
@@ -290,29 +284,29 @@ pub fn framed_with_fds<A, C>(io: A, code
         codec: codec,
         read_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
         incoming_fds: IncomingFds::new(FDS_CAPACITY),
         is_readable: false,
         eof: false,
         frames: VecDeque::new(),
         write_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
         outgoing_fds: BytesMut::with_capacity(
-            FDS_CAPACITY * cmsg::space(mem::size_of::<[RawFd; 3]>())
-        )
+            FDS_CAPACITY * cmsg::space(mem::size_of::<[RawFd; 3]>()),
+        ),
     }
 }
 
 fn write_zero() -> io::Error {
     io::Error::new(io::ErrorKind::WriteZero, "failed to write frame to io")
 }
 
 fn clone_into_array<A, T>(slice: &[T]) -> A
 where
     A: Sized + Default + AsMut<[T]>,
-    T: Clone
+    T: Clone,
 {
     let mut a = Default::default();
     <A as AsMut<[T]>>::as_mut(&mut a).clone_from_slice(slice);
     a
 }
 
 fn close_fds(fds: &[RawFd]) {
     for fd in fds {
--- a/media/audioipc/audioipc/src/frame.rs
+++ b/media/audioipc/audioipc/src/frame.rs
@@ -16,17 +16,17 @@ const BACKPRESSURE_THRESHOLD: usize = 4 
 /// the `Codec` trait to encode and decode the payload.
 pub struct Framed<A, C> {
     io: A,
     codec: C,
     read_buf: BytesMut,
     write_buf: BytesMut,
     frame: Option<<Bytes as IntoBuf>::Buf>,
     is_readable: bool,
-    eof: bool
+    eof: bool,
 }
 
 impl<A, C> Framed<A, C>
 where
     A: AsyncWrite,
 {
     // If there is a buffered frame, try to write it to `A`
     fn do_write(&mut self) -> Poll<(), io::Error> {
@@ -154,11 +154,11 @@ fn write_zero() -> io::Error {
 pub fn framed<A, C>(io: A, codec: C) -> Framed<A, C> {
     Framed {
         io: io,
         codec: codec,
         read_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
         write_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
         frame: None,
         is_readable: false,
-        eof: false
+        eof: false,
     }
 }
--- a/media/audioipc/audioipc/src/lib.rs
+++ b/media/audioipc/audioipc/src/lib.rs
@@ -1,35 +1,34 @@
 // Copyright © 2017 Mozilla Foundation
 //
 // This program is made available under an ISC-style license.  See the
 // accompanying file LICENSE for details
 #![allow(dead_code)] // TODO: Remove.
-
 #![recursion_limit = "1024"]
 #[macro_use]
 extern crate error_chain;
 
 #[macro_use]
 extern crate log;
 
 #[macro_use]
 extern crate serde_derive;
 
 extern crate bincode;
 extern crate bytes;
-extern crate cubeb_core;
+extern crate cubeb;
 #[macro_use]
 extern crate futures;
 extern crate iovec;
 extern crate libc;
 extern crate memmap;
-extern crate serde;
 #[macro_use]
 extern crate scoped_tls;
+extern crate serde;
 extern crate tokio_core;
 #[macro_use]
 extern crate tokio_io;
 extern crate tokio_uds;
 
 pub mod async;
 pub mod cmsg;
 pub mod codec;
@@ -38,41 +37,47 @@ pub mod fd_passing;
 pub mod frame;
 pub mod rpc;
 pub mod core;
 pub mod messages;
 mod msg;
 pub mod shm;
 
 use iovec::IoVec;
-
 #[cfg(target_os = "linux")]
 use libc::MSG_CMSG_CLOEXEC;
 pub use messages::{ClientMessage, ServerMessage};
-
 use std::env::temp_dir;
 use std::io;
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
 use std::path::PathBuf;
 #[cfg(not(target_os = "linux"))]
 const MSG_CMSG_CLOEXEC: libc::c_int = 0;
 
 // Extend sys::os::unix::net::UnixStream to support sending and receiving a single file desc.
 // We can extend UnixStream by using traits, eliminating the need to introduce a new wrapped
 // UnixStream type.
 pub trait RecvMsg {
-    fn recv_msg(&mut self, iov: &mut [&mut IoVec], cmsg: &mut [u8]) -> io::Result<(usize, usize, i32)>;
+    fn recv_msg(
+        &mut self,
+        iov: &mut [&mut IoVec],
+        cmsg: &mut [u8],
+    ) -> io::Result<(usize, usize, i32)>;
 }
 
 pub trait SendMsg {
     fn send_msg(&mut self, iov: &[&IoVec], cmsg: &[u8]) -> io::Result<usize>;
 }
 
 impl<T: AsRawFd> RecvMsg for T {
-    fn recv_msg(&mut self, iov: &mut [&mut IoVec], cmsg: &mut [u8]) -> io::Result<(usize, usize, i32)> {
+    fn recv_msg(
+        &mut self,
+        iov: &mut [&mut IoVec],
+        cmsg: &mut [u8],
+    ) -> io::Result<(usize, usize, i32)> {
         msg::recv_msg_with_flags(self.as_raw_fd(), iov, cmsg, MSG_CMSG_CLOEXEC)
     }
 }
 
 impl<T: AsRawFd> SendMsg for T {
     fn send_msg(&mut self, iov: &[&IoVec], cmsg: &[u8]) -> io::Result<usize> {
         msg::send_msg_with_flags(self.as_raw_fd(), iov, cmsg, 0)
     }
--- a/media/audioipc/audioipc/src/messages.rs
+++ b/media/audioipc/audioipc/src/messages.rs
@@ -1,48 +1,48 @@
 // Copyright © 2017 Mozilla Foundation
 //
 // This program is made available under an ISC-style license.  See the
 // accompanying file LICENSE for details
 
-use cubeb_core::{self, ffi};
+use cubeb::{self, ffi};
 use std::ffi::{CStr, CString};
-use std::os::raw::c_char;
+use std::os::raw::{c_char, c_int, c_uint};
 use std::os::unix::io::RawFd;
 use std::ptr;
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Device {
     pub output_name: Option<Vec<u8>>,
-    pub input_name: Option<Vec<u8>>
+    pub input_name: Option<Vec<u8>>,
 }
 
-impl<'a> From<cubeb_core::Device<'a>> for Device {
-    fn from(info: cubeb_core::Device) -> Self {
+impl<'a> From<&'a cubeb::DeviceRef> for Device {
+    fn from(info: &'a cubeb::DeviceRef) -> Self {
         Self {
             output_name: info.output_name_bytes().map(|s| s.to_vec()),
-            input_name: info.input_name_bytes().map(|s| s.to_vec())
+            input_name: info.input_name_bytes().map(|s| s.to_vec()),
         }
     }
 }
 
 impl From<ffi::cubeb_device> for Device {
     fn from(info: ffi::cubeb_device) -> Self {
         Self {
             output_name: dup_str(info.output_name),
-            input_name: dup_str(info.input_name)
+            input_name: dup_str(info.input_name),
         }
     }
 }
 
 impl From<Device> for ffi::cubeb_device {
     fn from(info: Device) -> Self {
         Self {
             output_name: opt_str(info.output_name),
-            input_name: opt_str(info.input_name)
+            input_name: opt_str(info.input_name),
         }
     }
 }
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct DeviceInfo {
     pub devid: usize,
     pub device_id: Option<Vec<u8>>,
@@ -57,21 +57,22 @@ pub struct DeviceInfo {
     pub format: ffi::cubeb_device_fmt,
     pub default_format: ffi::cubeb_device_fmt,
     pub max_channels: u32,
     pub default_rate: u32,
     pub max_rate: u32,
     pub min_rate: u32,
 
     pub latency_lo: u32,
-    pub latency_hi: u32
+    pub latency_hi: u32,
 }
 
-impl<'a> From<&'a ffi::cubeb_device_info> for DeviceInfo {
-    fn from(info: &'a ffi::cubeb_device_info) -> Self {
+impl<'a> From<&'a cubeb::DeviceInfoRef> for DeviceInfo {
+    fn from(info: &'a cubeb::DeviceInfoRef) -> Self {
+        let info = unsafe { &*info.as_ptr() };
         DeviceInfo {
             devid: info.devid as _,
             device_id: dup_str(info.device_id),
             friendly_name: dup_str(info.friendly_name),
             group_id: dup_str(info.group_id),
             vendor_name: dup_str(info.vendor_name),
 
             device_type: info.device_type,
@@ -81,17 +82,17 @@ impl<'a> From<&'a ffi::cubeb_device_info
             format: info.format,
             default_format: info.default_format,
             max_channels: info.max_channels,
             default_rate: info.default_rate,
             max_rate: info.max_rate,
             min_rate: info.min_rate,
 
             latency_lo: info.latency_lo,
-            latency_hi: info.latency_hi
+            latency_hi: info.latency_hi,
         }
     }
 }
 
 impl From<DeviceInfo> for ffi::cubeb_device_info {
     fn from(info: DeviceInfo) -> Self {
         ffi::cubeb_device_info {
             devid: info.devid as _,
@@ -107,92 +108,72 @@ impl From<DeviceInfo> for ffi::cubeb_dev
             format: info.format,
             default_format: info.default_format,
             max_channels: info.max_channels,
             default_rate: info.default_rate,
             max_rate: info.max_rate,
             min_rate: info.min_rate,
 
             latency_lo: info.latency_lo,
-            latency_hi: info.latency_hi
+            latency_hi: info.latency_hi,
         }
     }
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, Deserialize, Serialize)]
 pub struct StreamParams {
-    pub format: u32,
-    pub rate: u16,
-    pub channels: u8,
-    pub layout: i32,
-    pub prefs: i32
+    pub format: ffi::cubeb_sample_format,
+    pub rate: c_uint,
+    pub channels: c_uint,
+    pub layout: ffi::cubeb_channel_layout,
+    pub prefs: ffi::cubeb_stream_prefs,
 }
 
-impl<'a> From<&'a ffi::cubeb_stream_params> for StreamParams {
-    fn from(params: &'a ffi::cubeb_stream_params) -> Self {
-        assert!(params.channels <= u32::from(u8::max_value()));
-
-        StreamParams {
-            format: params.format,
-            rate: params.rate as u16,
-            channels: params.channels as u8,
-            layout: params.layout,
-            prefs: params.prefs
-        }
-    }
-}
-
-impl<'a> From<&'a StreamParams> for ffi::cubeb_stream_params {
-    fn from(params: &StreamParams) -> Self {
-        ffi::cubeb_stream_params {
-            format: params.format,
-            rate: u32::from(params.rate),
-            channels: u32::from(params.channels),
-            layout: params.layout,
-            prefs: params.prefs
-        }
+impl<'a> From<&'a cubeb::StreamParamsRef> for StreamParams {
+    fn from(x: &cubeb::StreamParamsRef) -> StreamParams {
+        unsafe { *(x.as_ptr() as *mut StreamParams) }
     }
 }
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct StreamInitParams {
     pub stream_name: Option<Vec<u8>>,
     pub input_device: usize,
     pub input_stream_params: Option<StreamParams>,
     pub output_device: usize,
     pub output_stream_params: Option<StreamParams>,
-    pub latency_frames: u32
+    pub latency_frames: u32,
 }
 
 fn dup_str(s: *const c_char) -> Option<Vec<u8>> {
     if s.is_null() {
         None
     } else {
         let vec: Vec<u8> = unsafe { CStr::from_ptr(s) }.to_bytes().to_vec();
         Some(vec)
     }
 }
 
-fn opt_str(v: Option<Vec<u8>>) -> *const c_char {
+fn opt_str(v: Option<Vec<u8>>) -> *mut c_char {
     match v {
         Some(v) => match CString::new(v) {
             Ok(s) => s.into_raw(),
             Err(_) => {
                 debug!("Failed to convert bytes to CString");
-                ptr::null()
+                ptr::null_mut()
             }
         },
-        None => ptr::null()
+        None => ptr::null_mut(),
     }
 }
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct StreamCreate {
     pub token: usize,
-    pub fds: [RawFd; 3]
+    pub fds: [RawFd; 3],
 }
 
 // Client -> Server messages.
 // TODO: Callbacks should be different messages types so
 // ServerConn::process_msg doesn't have a catch-all case.
 #[derive(Debug, Serialize, Deserialize)]
 pub enum ServerMessage {
     ClientConnect,
@@ -210,17 +191,17 @@ pub enum ServerMessage {
 
     StreamStart(usize),
     StreamStop(usize),
     StreamResetDefaultDevice(usize),
     StreamGetPosition(usize),
     StreamGetLatency(usize),
     StreamSetVolume(usize, f32),
     StreamSetPanning(usize, f32),
-    StreamGetCurrentDevice(usize)
+    StreamGetCurrentDevice(usize),
 }
 
 // Server -> Client messages.
 // TODO: Streams need id.
 #[derive(Debug, Serialize, Deserialize)]
 pub enum ClientMessage {
     ClientConnected,
     ClientDisconnected,
@@ -239,52 +220,84 @@ pub enum ClientMessage {
     StreamStopped,
     StreamDefaultDeviceReset,
     StreamPosition(u64),
     StreamLatency(u32),
     StreamVolumeSet,
     StreamPanningSet,
     StreamCurrentDevice(Device),
 
-    Error(ffi::cubeb_error_code)
+    Error(c_int),
 }
 
 #[derive(Debug, Deserialize, Serialize)]
 pub enum CallbackReq {
     Data(isize, usize),
-    State(ffi::cubeb_state)
+    State(ffi::cubeb_state),
 }
 
 #[derive(Debug, Deserialize, Serialize)]
 pub enum CallbackResp {
     Data(isize),
-    State
+    State,
 }
 
 pub trait AssocRawFd {
     fn fd(&self) -> Option<[RawFd; 3]> {
         None
     }
     fn take_fd<F>(&mut self, _: F)
     where
-        F: FnOnce() -> Option<[RawFd; 3]>
+        F: FnOnce() -> Option<[RawFd; 3]>,
     {
     }
 }
 
 impl AssocRawFd for ServerMessage {}
 impl AssocRawFd for ClientMessage {
     fn fd(&self) -> Option<[RawFd; 3]> {
         match *self {
             ClientMessage::StreamCreated(ref data) => Some(data.fds),
-            _ => None
+            _ => None,
         }
     }
 
     fn take_fd<F>(&mut self, f: F)
     where
-        F: FnOnce() -> Option<[RawFd; 3]>
+        F: FnOnce() -> Option<[RawFd; 3]>,
     {
         if let ClientMessage::StreamCreated(ref mut data) = *self {
             data.fds = f().unwrap();
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::StreamParams;
+    use cubeb::ffi;
+    use std::mem;
+
+    #[test]
+    fn stream_params_size_check() {
+        assert_eq!(
+            mem::size_of::<StreamParams>(),
+            mem::size_of::<ffi::cubeb_stream_params>()
+        )
+    }
+
+    #[test]
+    fn stream_params_from() {
+        let mut raw = ffi::cubeb_stream_params::default();
+        raw.format = ffi::CUBEB_SAMPLE_FLOAT32BE;
+        raw.rate = 96_000;
+        raw.channels = 32;
+        raw.layout = ffi::CUBEB_LAYOUT_3F1_LFE;
+        raw.prefs = ffi::CUBEB_STREAM_PREF_LOOPBACK;
+        let wrapped = ::cubeb::StreamParams::from(raw);
+        let params = StreamParams::from(wrapped.as_ref());
+        assert_eq!(params.format, raw.format);
+        assert_eq!(params.rate, raw.rate);
+        assert_eq!(params.channels, raw.channels);
+        assert_eq!(params.layout, raw.layout);
+        assert_eq!(params.prefs, raw.prefs);
+    }
+}
--- a/media/audioipc/audioipc/src/msg.rs
+++ b/media/audioipc/audioipc/src/msg.rs
@@ -11,27 +11,27 @@ fn cvt(r: libc::ssize_t) -> io::Result<u
         Ok(r as usize)
     }
 }
 
 // Convert return of -1 into error message, handling retry on EINTR
 fn cvt_r<F: FnMut() -> libc::ssize_t>(mut f: F) -> io::Result<usize> {
     loop {
         match cvt(f()) {
-            Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {},
-            other => return other
+            Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
+            other => return other,
         }
     }
 }
 
 pub fn recv_msg_with_flags(
     socket: RawFd,
     bufs: &mut [&mut IoVec],
     cmsg: &mut [u8],
-    flags: libc::c_int
+    flags: libc::c_int,
 ) -> io::Result<(usize, usize, libc::c_int)> {
     let slice = iovec::as_os_slice_mut(bufs);
     let len = cmp::min(<libc::c_int>::max_value() as usize, slice.len());
     let (control, controllen) = if cmsg.is_empty() {
         (ptr::null_mut(), 0)
     } else {
         (cmsg.as_ptr() as *mut _, cmsg.len())
     };
@@ -51,17 +51,17 @@ pub fn recv_msg_with_flags(
     let controllen = msghdr.msg_controllen as usize;
     Ok((n, controllen, msghdr.msg_flags))
 }
 
 pub fn send_msg_with_flags(
     socket: RawFd,
     bufs: &[&IoVec],
     cmsg: &[u8],
-    flags: libc::c_int
+    flags: libc::c_int,
 ) -> io::Result<usize> {
     let slice = iovec::as_os_slice(bufs);
     let len = cmp::min(<libc::c_int>::max_value() as usize, slice.len());
     let (control, controllen) = if cmsg.is_empty() {
         (ptr::null_mut(), 0)
     } else {
         (cmsg.as_ptr() as *mut _, cmsg.len())
     };
@@ -69,12 +69,10 @@ pub fn send_msg_with_flags(
     let mut msghdr: libc::msghdr = unsafe { mem::zeroed() };
     msghdr.msg_name = ptr::null_mut();
     msghdr.msg_namelen = 0;
     msghdr.msg_iov = slice.as_ptr() as *mut _;
     msghdr.msg_iovlen = len as _;
     msghdr.msg_control = control;
     msghdr.msg_controllen = controllen as _;
 
-    cvt_r(|| unsafe {
-        libc::sendmsg(socket, &msghdr as *const _, flags)
-    })
+    cvt_r(|| unsafe { libc::sendmsg(socket, &msghdr as *const _, flags) })
 }
--- a/media/audioipc/audioipc/src/rpc/client/mod.rs
+++ b/media/audioipc/audioipc/src/rpc/client/mod.rs
@@ -48,28 +48,28 @@ use std::io;
 use tokio_core::reactor::Handle;
 
 mod proxy;
 
 pub use self::proxy::{ClientProxy, Response};
 
 pub fn bind_client<C>(
     transport: C::Transport,
-    handle: &Handle
+    handle: &Handle,
 ) -> proxy::ClientProxy<C::Request, C::Response>
 where
-    C: Client
+    C: Client,
 {
     let (tx, rx) = proxy::channel();
 
     let fut = {
         let handler = ClientHandler::<C> {
             transport: transport,
             requests: rx,
-            in_flight: VecDeque::with_capacity(32)
+            in_flight: VecDeque::with_capacity(32),
         };
         Driver::new(handler)
     };
 
     // Spawn the RPC driver into task
     handle.spawn(Box::new(fut.map_err(|_| ())));
 
     tx
@@ -87,43 +87,43 @@ pub trait Client: 'static {
         + Stream<Item = Self::Response, Error = io::Error>
         + Sink<SinkItem = Self::Request, SinkError = io::Error>;
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 
 struct ClientHandler<C>
 where
-    C: Client
+    C: Client,
 {
     transport: C::Transport,
     requests: proxy::Receiver<C::Request, C::Response>,
-    in_flight: VecDeque<oneshot::Sender<C::Response>>
+    in_flight: VecDeque<oneshot::Sender<C::Response>>,
 }
 
 impl<C> Handler for ClientHandler<C>
 where
-    C: Client
+    C: Client,
 {
     type In = C::Response;
     type Out = C::Request;
     type Transport = C::Transport;
 
     fn transport(&mut self) -> &mut Self::Transport {
         &mut self.transport
     }
 
     fn consume(&mut self, response: Self::In) -> io::Result<()> {
         trace!("ClientHandler::consume");
         if let Some(complete) = self.in_flight.pop_front() {
             drop(complete.send(response));
         } else {
             return Err(io::Error::new(
                 io::ErrorKind::Other,
-                "request / response mismatch"
+                "request / response mismatch",
             ));
         }
 
         Ok(())
     }
 
     /// Produce a message
     fn produce(&mut self) -> Poll<Option<Self::Out>, io::Error> {
@@ -133,26 +133,26 @@ where
         match self.requests.poll() {
             Ok(Async::Ready(Some((request, complete)))) => {
                 trace!("  --> received request");
 
                 // Track complete handle
                 self.in_flight.push_back(complete);
 
                 Ok(Some(request).into())
-            },
+            }
             Ok(Async::Ready(None)) => {
                 trace!("  --> client dropped");
                 Ok(None.into())
-            },
+            }
             Ok(Async::NotReady) => {
                 trace!("  --> not ready");
                 Ok(Async::NotReady)
-            },
-            Err(_) => unreachable!()
+            }
+            Err(_) => unreachable!(),
         }
     }
 
     /// RPC currently in flight
     fn has_in_flight(&self) -> bool {
         !self.in_flight.is_empty()
     }
 }
--- a/media/audioipc/audioipc/src/rpc/client/proxy.rs
+++ b/media/audioipc/audioipc/src/rpc/client/proxy.rs
@@ -52,66 +52,62 @@ use std::io;
 /// client connection.
 pub type Request<R, Q> = (R, oneshot::Sender<Q>);
 
 /// Receive requests submitted to the client
 pub type Receiver<R, Q> = mpsc::UnboundedReceiver<Request<R, Q>>;
 
 /// Response future returned from a client
 pub struct Response<Q> {
-    inner: oneshot::Receiver<Q>
+    inner: oneshot::Receiver<Q>,
 }
 
 pub struct ClientProxy<R, Q> {
-    tx: mpsc::UnboundedSender<Request<R, Q>>
+    tx: mpsc::UnboundedSender<Request<R, Q>>,
 }
 
 impl<R, Q> Clone for ClientProxy<R, Q> {
     fn clone(&self) -> Self {
         ClientProxy {
-            tx: self.tx.clone()
+            tx: self.tx.clone(),
         }
     }
 }
 
 pub fn channel<R, Q>() -> (ClientProxy<R, Q>, Receiver<R, Q>) {
     // Create a channel to send the requests to client-side of rpc.
     let (tx, rx) = mpsc::unbounded();
 
     // Wrap the `tx` part in ClientProxy so the rpc call interface
     // can be implemented.
-    let client = ClientProxy {
-        tx
-    };
+    let client = ClientProxy { tx };
 
     (client, rx)
 }
 
 impl<R, Q> ClientProxy<R, Q> {
     pub fn call(&self, request: R) -> Response<Q> {
         // The response to returned from the rpc client task over a
         // oneshot channel.
         let (tx, rx) = oneshot::channel();
 
         // If send returns an Err, its because the other side has been dropped.
         // By ignoring it, we are just dropping the `tx`, which will mean the
         // rx will return Canceled when polled. In turn, that is translated
         // into a BrokenPipe, which conveys the proper error.
         let _ = self.tx.send((request, tx));
 
-        Response {
-            inner: rx
-        }
+        Response { inner: rx }
     }
 }
 
 impl<R, Q> fmt::Debug for ClientProxy<R, Q>
 where
     R: fmt::Debug,
-    Q: fmt::Debug
+    Q: fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(f, "ClientProxy {{ ... }}")
     }
 }
 
 impl<Q> Future for Response<Q> {
     type Item = Q;
@@ -127,14 +123,14 @@ impl<Q> Future for Response<Q> {
                 Err(e)
             }
         }
     }
 }
 
 impl<Q> fmt::Debug for Response<Q>
 where
-    Q: fmt::Debug
+    Q: fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         write!(f, "Response {{ ... }}")
     }
 }
--- a/media/audioipc/audioipc/src/rpc/driver.rs
+++ b/media/audioipc/audioipc/src/rpc/driver.rs
@@ -5,38 +5,38 @@
 
 use futures::{Async, AsyncSink, Future, Poll, Sink, Stream};
 use rpc::Handler;
 use std::fmt;
 use std::io;
 
 pub struct Driver<T>
 where
-    T: Handler
+    T: Handler,
 {
     // Glue
     handler: T,
 
     // True as long as the connection has more request frames to read.
     run: bool,
 
     // True when the transport is fully flushed
-    is_flushed: bool
+    is_flushed: bool,
 }
 
 impl<T> Driver<T>
 where
-    T: Handler
+    T: Handler,
 {
     /// Create a new rpc driver with the given service and transport.
     pub fn new(handler: T) -> Driver<T> {
         Driver {
             handler: handler,
             run: true,
-            is_flushed: true
+            is_flushed: true,
         }
     }
 
     /// Returns true if the driver has nothing left to do
     fn is_done(&self) -> bool {
         !self.run && self.is_flushed && !self.has_in_flight()
     }
 
@@ -60,17 +60,17 @@ where
         match req {
             Some(message) => {
                 trace!("received message");
 
                 if let Err(e) = self.handler.consume(message) {
                     // TODO: Should handler be infalliable?
                     panic!("unimplemented error handling: {:?}", e);
                 }
-            },
+            }
             None => {
                 trace!("received None");
                 // At this point, we just return. This works
                 // because poll with be called again and go
                 // through the receive-cycle again.
                 self.run = false;
             }
         }
@@ -81,24 +81,24 @@ where
     /// Send outgoing messages to the transport.
     fn send_outgoing(&mut self) -> io::Result<()> {
         trace!("send_responses");
         loop {
             match try!(self.handler.produce()) {
                 Async::Ready(Some(message)) => {
                     trace!("  --> got message");
                     try!(self.process_outgoing(message));
-                },
+                }
                 Async::Ready(None) => {
                     trace!("  --> got None");
                     // The service is done with the connection.
                     break;
-                },
+                }
                 // Nothing to dispatch
-                Async::NotReady => break
+                Async::NotReady => break,
             }
         }
 
         Ok(())
     }
 
     fn process_outgoing(&mut self, message: T::Out) -> io::Result<()> {
         trace!("process_outgoing");
@@ -116,17 +116,17 @@ where
 
     fn has_in_flight(&self) -> bool {
         self.handler.has_in_flight()
     }
 }
 
 impl<T> Future for Driver<T>
 where
-    T: Handler
+    T: Handler,
 {
     type Item = ();
     type Error = io::Error;
 
     fn poll(&mut self) -> Poll<(), Self::Error> {
         trace!("rpc::Driver::tick");
 
         // First read off data from the socket
@@ -148,23 +148,23 @@ where
 }
 
 fn assert_send<S: Sink>(s: &mut S, item: S::SinkItem) -> Result<(), S::SinkError> {
     match try!(s.start_send(item)) {
         AsyncSink::Ready => Ok(()),
         AsyncSink::NotReady(_) => panic!(
             "sink reported itself as ready after `poll_ready` but was \
              then unable to accept a message"
-        )
+        ),
     }
 }
 
 impl<T> fmt::Debug for Driver<T>
 where
-    T: Handler + fmt::Debug
+    T: Handler + fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         f.debug_struct("rpc::Handler")
             .field("handler", &self.handler)
             .field("run", &self.run)
             .field("is_flushed", &self.is_flushed)
             .finish()
     }
--- a/media/audioipc/audioipc/src/rpc/server.rs
+++ b/media/audioipc/audioipc/src/rpc/server.rs
@@ -44,23 +44,23 @@ use rpc::Handler;
 use rpc::driver::Driver;
 use std::collections::VecDeque;
 use std::io;
 use tokio_core::reactor::Handle;
 
 /// Bind an async I/O object `io` to the `server`.
 pub fn bind_server<S>(transport: S::Transport, server: S, handle: &Handle)
 where
-    S: Server
+    S: Server,
 {
     let fut = {
         let handler = ServerHandler {
             server: server,
             transport: transport,
-            in_flight: VecDeque::with_capacity(32)
+            in_flight: VecDeque::with_capacity(32),
         };
         Driver::new(handler)
     };
 
     // Spawn the RPC driver into task
     handle.spawn(Box::new(fut.map_err(|_| ())))
 }
 
@@ -79,34 +79,33 @@ pub trait Server: 'static {
     type Transport: 'static
         + Stream<Item = Self::Request, Error = io::Error>
         + Sink<SinkItem = Self::Response, SinkError = io::Error>;
 
     /// Process the request and return the response asynchronously.
     fn process(&mut self, req: Self::Request) -> Self::Future;
 }
 
-
 ////////////////////////////////////////////////////////////////////////////////
 
 struct ServerHandler<S>
 where
-    S: Server
+    S: Server,
 {
     // The service handling the connection
     server: S,
     // The transport responsible for sending/receving messages over the wire
     transport: S::Transport,
     // FIFO of "in flight" responses to requests.
-    in_flight: VecDeque<InFlight<S::Future>>
+    in_flight: VecDeque<InFlight<S::Future>>,
 }
 
 impl<S> Handler for ServerHandler<S>
 where
-    S: Server
+    S: Server,
 {
     type In = S::Request;
     type Out = S::Response;
     type Transport = S::Transport;
 
     /// Mutable reference to the transport
     fn transport(&mut self) -> &mut Self::Transport {
         &mut self.transport
@@ -128,51 +127,51 @@ where
 
         // Make progress on pending responses
         for pending in &mut self.in_flight {
             pending.poll();
         }
 
         // Is the head of the queue ready?
         match self.in_flight.front() {
-            Some(&InFlight::Done(_)) => {},
+            Some(&InFlight::Done(_)) => {}
             _ => {
                 trace!("  --> not ready");
                 return Ok(Async::NotReady);
             }
         }
 
         // Return the ready response
         match self.in_flight.pop_front() {
             Some(InFlight::Done(res)) => {
                 trace!("  --> received response");
                 Ok(Async::Ready(Some(res)))
-            },
-            _ => panic!()
+            }
+            _ => panic!(),
         }
     }
 
     /// RPC currently in flight
     fn has_in_flight(&self) -> bool {
         !self.in_flight.is_empty()
     }
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 
 enum InFlight<F: Future<Error = ()>> {
     Active(F),
-    Done(F::Item)
+    Done(F::Item),
 }
 
 impl<F: Future<Error = ()>> InFlight<F> {
     fn poll(&mut self) {
         let res = match *self {
             InFlight::Active(ref mut f) => match f.poll() {
                 Ok(Async::Ready(e)) => e,
                 Err(_) => unreachable!(),
-                Ok(Async::NotReady) => return
+                Ok(Async::NotReady) => return,
             },
-            _ => return
+            _ => return,
         };
         *self = InFlight::Done(res);
     }
 }
--- a/media/audioipc/audioipc/src/shm.rs
+++ b/media/audioipc/audioipc/src/shm.rs
@@ -1,35 +1,30 @@
 use errors::*;
 use memmap::{Mmap, MmapViewSync, Protection};
 use std::fs::{remove_file, File, OpenOptions};
 use std::path::Path;
 use std::sync::atomic;
 
 pub struct SharedMemReader {
-    mmap: Mmap
+    mmap: Mmap,
 }
 
 impl SharedMemReader {
     pub fn new(path: &Path, size: usize) -> Result<(SharedMemReader, File)> {
         let file = OpenOptions::new()
             .read(true)
             .write(true)
             .create_new(true)
             .open(path)?;
         let _ = remove_file(path);
         file.set_len(size as u64)?;
         let mmap = Mmap::open(&file, Protection::Read)?;
         assert_eq!(mmap.len(), size);
-        Ok((
-            SharedMemReader {
-                mmap
-            },
-            file
-        ))
+        Ok((SharedMemReader { mmap }, file))
     }
 
     pub fn read(&self, buf: &mut [u8]) -> Result<()> {
         if buf.is_empty() {
             return Ok(());
         }
         // TODO: Track how much is in the shm area.
         if buf.len() <= self.mmap.len() {
@@ -41,27 +36,25 @@ impl SharedMemReader {
             Ok(())
         } else {
             bail!("mmap size");
         }
     }
 }
 
 pub struct SharedMemSlice {
-    view: MmapViewSync
+    view: MmapViewSync,
 }
 
 impl SharedMemSlice {
     pub fn from(file: &File, size: usize) -> Result<SharedMemSlice> {
         let mmap = Mmap::open(file, Protection::Read)?;
         assert_eq!(mmap.len(), size);
         let view = mmap.into_view_sync();
-        Ok(SharedMemSlice {
-            view
-        })
+        Ok(SharedMemSlice { view })
     }
 
     pub fn get_slice(&self, size: usize) -> Result<&[u8]> {
         if size == 0 {
             return Ok(&[]);
         }
         // TODO: Track how much is in the shm area.
         if size <= self.view.len() {
@@ -74,41 +67,36 @@ impl SharedMemSlice {
     }
 
     /// Clones the view of the memory map.
     ///
     /// The underlying memory map is shared, and thus the caller must ensure that the memory
     /// underlying the view is not illegally aliased.
     pub unsafe fn clone_view(&self) -> Self {
         SharedMemSlice {
-            view: self.view.clone()
+            view: self.view.clone(),
         }
     }
 }
 
 pub struct SharedMemWriter {
-    mmap: Mmap
+    mmap: Mmap,
 }
 
 impl SharedMemWriter {
     pub fn new(path: &Path, size: usize) -> Result<(SharedMemWriter, File)> {
         let file = OpenOptions::new()
             .read(true)
             .write(true)
             .create_new(true)
             .open(path)?;
         let _ = remove_file(path);
         file.set_len(size as u64)?;
         let mmap = Mmap::open(&file, Protection::ReadWrite)?;
-        Ok((
-            SharedMemWriter {
-                mmap
-            },
-            file
-        ))
+        Ok((SharedMemWriter { mmap }, file))
     }
 
     pub fn write(&mut self, buf: &[u8]) -> Result<()> {
         if buf.is_empty() {
             return Ok(());
         }
         // TODO: Track how much is in the shm area.
         if buf.len() <= self.mmap.len() {
@@ -119,27 +107,25 @@ impl SharedMemWriter {
             Ok(())
         } else {
             bail!("mmap size");
         }
     }
 }
 
 pub struct SharedMemMutSlice {
-    view: MmapViewSync
+    view: MmapViewSync,
 }
 
 impl SharedMemMutSlice {
     pub fn from(file: &File, size: usize) -> Result<SharedMemMutSlice> {
         let mmap = Mmap::open(file, Protection::ReadWrite)?;
         assert_eq!(mmap.len(), size);
         let view = mmap.into_view_sync();
-        Ok(SharedMemMutSlice {
-            view
-        })
+        Ok(SharedMemMutSlice { view })
     }
 
     pub fn get_mut_slice(&mut self, size: usize) -> Result<&mut [u8]> {
         if size == 0 {
             return Ok(&mut []);
         }
         // TODO: Track how much is in the shm area.
         if size <= self.view.len() {
@@ -153,12 +139,12 @@ impl SharedMemMutSlice {
 
     /// Clones the view of the memory map.
     ///
     /// The underlying memory map is shared, and thus the caller must
     /// ensure that the memory underlying the view is not illegally
     /// aliased.
     pub unsafe fn clone_view(&self) -> Self {
         SharedMemMutSlice {
-            view: self.view.clone()
+            view: self.view.clone(),
         }
     }
 }
--- a/media/audioipc/client/Cargo.toml
+++ b/media/audioipc/client/Cargo.toml
@@ -1,21 +1,21 @@
 [package]
 name = "audioipc-client"
-version = "0.2.0"
+version = "0.3.0"
 authors = [
         "Matthew Gregan <kinetik@flim.org>",
         "Dan Glastonbury <dan.glastonbury@gmail.com>"
         ]
 description = "Cubeb Backend for talking to remote cubeb server."
 
 [dependencies]
 audioipc = { path="../audioipc" }
-cubeb-backend = { path = "../../cubeb-rs/cubeb-backend" }
-cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
+cubeb-backend = "0.4"
+foreign-types = "0.3"
 # rayon-core in Gecko uses futures 0.1.13
 futures = { version="=0.1.13", default-features=false, features=["use_std"] }
 # futures-cpupool 0.1.5 matches futures 0.1.13
 futures-cpupool = { version="=0.1.5", default-features=false }
 libc = "0.2"
 log = "^0.3.6"
 tokio-core = "0.1"
 tokio-uds = "0.1.7"
\ No newline at end of file
--- a/media/audioipc/client/src/context.rs
+++ b/media/audioipc/client/src/context.rs
@@ -4,56 +4,54 @@
 // accompanying file LICENSE for details
 
 use ClientStream;
 use assert_not_in_callback;
 use audioipc::{messages, ClientMessage, ServerMessage};
 use audioipc::{core, rpc};
 use audioipc::codec::LengthDelimitedCodec;
 use audioipc::fd_passing::{framed_with_fds, FramedWithFds};
-use cubeb_backend::{Context, Ops};
-use cubeb_core::{ffi, DeviceId, DeviceType, Error, Result, StreamParams};
-use cubeb_core::binding::Binding;
+use cubeb_backend::{ffi, ChannelLayout, Context, ContextOps, DeviceCollectionRef, DeviceId,
+                    DeviceType, Error, Ops, Result, Stream, StreamParams, StreamParamsRef};
 use futures::Future;
 use futures_cpupool::{self, CpuPool};
 use libc;
-use std::{fmt, io, mem};
+use std::{fmt, io, mem, ptr};
 use std::ffi::{CStr, CString};
 use std::os::raw::c_void;
 use std::os::unix::io::FromRawFd;
 use std::os::unix::net;
 use std::sync::mpsc;
 use stream;
 use tokio_core::reactor::{Handle, Remote};
 use tokio_uds::UnixStream;
 
 struct CubebClient;
 
 impl rpc::Client for CubebClient {
     type Request = ServerMessage;
     type Response = ClientMessage;
-    type Transport =
-        FramedWithFds<UnixStream, LengthDelimitedCodec<Self::Request, Self::Response>>;
+    type Transport = FramedWithFds<UnixStream, LengthDelimitedCodec<Self::Request, Self::Response>>;
 }
 
 macro_rules! t(
     ($e:expr) => (
         match $e {
             Ok(e) => e,
             Err(_) => return Err(Error::default())
         }
     ));
 
 pub const CLIENT_OPS: Ops = capi_new!(ClientContext, ClientStream);
 
 pub struct ClientContext {
     _ops: *const Ops,
     rpc: rpc::ClientProxy<ServerMessage, ClientMessage>,
     core: core::CoreThread,
-    cpu_pool: CpuPool
+    cpu_pool: CpuPool,
 }
 
 impl ClientContext {
     #[doc(hidden)]
     pub fn remote(&self) -> Remote {
         self.core.remote()
     }
 
@@ -74,192 +72,202 @@ fn open_server_stream() -> Result<net::U
         if let Some(fd) = super::G_SERVER_FD {
             return Ok(net::UnixStream::from_raw_fd(fd));
         }
 
         Err(Error::default())
     }
 }
 
-impl Context for ClientContext {
-    fn init(_context_name: Option<&CStr>) -> Result<*mut ffi::cubeb> {
+impl ContextOps for ClientContext {
+    fn init(_context_name: Option<&CStr>) -> Result<Context> {
         fn bind_and_send_client(
             stream: UnixStream,
             handle: &Handle,
-            tx_rpc: &mpsc::Sender<rpc::ClientProxy<ServerMessage, ClientMessage>>
+            tx_rpc: &mpsc::Sender<rpc::ClientProxy<ServerMessage, ClientMessage>>,
         ) -> Option<()> {
             let transport = framed_with_fds(stream, Default::default());
             let rpc = rpc::bind_client::<CubebClient>(transport, handle);
             // If send fails then the rx end has closed
             // which is unlikely here.
             let _ = tx_rpc.send(rpc);
             Some(())
         }
 
         assert_not_in_callback();
 
         let (tx_rpc, rx_rpc) = mpsc::channel();
 
         let core = t!(core::spawn_thread("AudioIPC Client RPC", move || {
             let handle = core::handle();
 
-            open_server_stream().ok()
+            open_server_stream()
+                .ok()
                 .and_then(|stream| UnixStream::from_stream(stream, &handle).ok())
                 .and_then(|stream| bind_and_send_client(stream, &handle, &tx_rpc))
-                .ok_or_else(|| io::Error::new(
-                    io::ErrorKind::Other,
-                    "Failed to open stream and create rpc."
-                ))
+                .ok_or_else(|| {
+                    io::Error::new(
+                        io::ErrorKind::Other,
+                        "Failed to open stream and create rpc.",
+                    )
+                })
         }));
 
         let rpc = t!(rx_rpc.recv());
 
         let cpupool = futures_cpupool::Builder::new()
             .name_prefix("AudioIPC")
             .create();
 
         let ctx = Box::new(ClientContext {
             _ops: &CLIENT_OPS as *const _,
             rpc: rpc,
             core: core,
-            cpu_pool: cpupool
+            cpu_pool: cpupool,
         });
-        Ok(Box::into_raw(ctx) as *mut _)
+        Ok(unsafe { Context::from_ptr(Box::into_raw(ctx) as *mut _) })
     }
 
-    fn backend_id(&self) -> &'static CStr {
+    fn backend_id(&mut self) -> &'static CStr {
         assert_not_in_callback();
         unsafe { CStr::from_ptr(b"remote\0".as_ptr() as *const _) }
     }
 
-    fn max_channel_count(&self) -> Result<u32> {
+    fn max_channel_count(&mut self) -> Result<u32> {
         assert_not_in_callback();
         send_recv!(self.rpc(), ContextGetMaxChannelCount => ContextMaxChannelCount())
     }
 
-    fn min_latency(&self, params: &StreamParams) -> Result<u32> {
+    fn min_latency(&mut self, params: StreamParams) -> Result<u32> {
         assert_not_in_callback();
-        let params = messages::StreamParams::from(unsafe { &*params.raw() });
+        let params = messages::StreamParams::from(params.as_ref());
         send_recv!(self.rpc(), ContextGetMinLatency(params) => ContextMinLatency())
     }
 
-    fn preferred_sample_rate(&self) -> Result<u32> {
+    fn preferred_sample_rate(&mut self) -> Result<u32> {
         assert_not_in_callback();
         send_recv!(self.rpc(), ContextGetPreferredSampleRate => ContextPreferredSampleRate())
     }
 
-    fn preferred_channel_layout(&self) -> Result<ffi::cubeb_channel_layout> {
+    fn preferred_channel_layout(&mut self) -> Result<ChannelLayout> {
         assert_not_in_callback();
         send_recv!(self.rpc(),
                    ContextGetPreferredChannelLayout => ContextPreferredChannelLayout())
+            .map(|l| {
+            ChannelLayout::from(l)
+        })
     }
 
-    fn enumerate_devices(&self, devtype: DeviceType) -> Result<ffi::cubeb_device_collection> {
+    fn enumerate_devices(
+        &mut self,
+        devtype: DeviceType,
+        collection: &DeviceCollectionRef,
+    ) -> Result<()> {
         assert_not_in_callback();
         let v: Vec<ffi::cubeb_device_info> = match send_recv!(self.rpc(),
                              ContextGetDeviceEnumeration(devtype.bits()) =>
                              ContextEnumeratedDevices())
         {
             Ok(mut v) => v.drain(..).map(|i| i.into()).collect(),
-            Err(e) => return Err(e)
+            Err(e) => return Err(e),
         };
-        let vs = v.into_boxed_slice();
-        let coll = ffi::cubeb_device_collection {
-            count: vs.len(),
-            device: vs.as_ptr()
-        };
+        let mut vs = v.into_boxed_slice();
+        let coll = unsafe { &mut *collection.as_ptr() };
+        coll.device = vs.as_mut_ptr();
+        coll.count = vs.len();
         // Giving away the memory owned by vs.  Don't free it!
         // Reclaimed in `device_collection_destroy`.
         mem::forget(vs);
-        Ok(coll)
+        Ok(())
     }
 
-    fn device_collection_destroy(&self, collection: *mut ffi::cubeb_device_collection) {
+    fn device_collection_destroy(&mut self, collection: &mut DeviceCollectionRef) -> Result<()> {
         assert_not_in_callback();
         unsafe {
-            let coll = &*collection;
+            let coll = &mut *collection.as_ptr();
             let mut devices = Vec::from_raw_parts(
                 coll.device as *mut ffi::cubeb_device_info,
                 coll.count,
-                coll.count
+                coll.count,
             );
             for dev in &mut devices {
                 if !dev.device_id.is_null() {
                     let _ = CString::from_raw(dev.device_id as *mut _);
                 }
                 if !dev.group_id.is_null() {
                     let _ = CString::from_raw(dev.group_id as *mut _);
                 }
                 if !dev.vendor_name.is_null() {
                     let _ = CString::from_raw(dev.vendor_name as *mut _);
                 }
                 if !dev.friendly_name.is_null() {
                     let _ = CString::from_raw(dev.friendly_name as *mut _);
                 }
             }
+            coll.device = ptr::null_mut();
+            coll.count = 0;
+            Ok(())
         }
     }
 
     fn stream_init(
-        &self,
+        &mut self,
         stream_name: Option<&CStr>,
         input_device: DeviceId,
-        input_stream_params: Option<&ffi::cubeb_stream_params>,
+        input_stream_params: Option<&StreamParamsRef>,
         output_device: DeviceId,
-        output_stream_params: Option<&ffi::cubeb_stream_params>,
+        output_stream_params: Option<&StreamParamsRef>,
         latency_frame: u32,
         // These params aren't sent to the server
         data_callback: ffi::cubeb_data_callback,
         state_callback: ffi::cubeb_state_callback,
-        user_ptr: *mut c_void
-    ) -> Result<*mut ffi::cubeb_stream> {
+        user_ptr: *mut c_void,
+    ) -> Result<Stream> {
         assert_not_in_callback();
 
-        fn opt_stream_params(
-            p: Option<&ffi::cubeb_stream_params>
-        ) -> Option<messages::StreamParams> {
+        fn opt_stream_params(p: Option<&StreamParamsRef>) -> Option<messages::StreamParams> {
             match p {
-                Some(raw) => Some(messages::StreamParams::from(raw)),
-                None => None
+                Some(p) => Some(messages::StreamParams::from(p)),
+                None => None,
             }
         }
 
         let stream_name = match stream_name {
             Some(s) => Some(s.to_bytes().to_vec()),
-            None => None
+            None => None,
         };
 
         let input_stream_params = opt_stream_params(input_stream_params);
         let output_stream_params = opt_stream_params(output_stream_params);
 
         let init_params = messages::StreamInitParams {
             stream_name: stream_name,
-            input_device: input_device.raw() as _,
+            input_device: input_device as usize,
             input_stream_params: input_stream_params,
-            output_device: output_device.raw() as _,
+            output_device: output_device as usize,
             output_stream_params: output_stream_params,
-            latency_frames: latency_frame
+            latency_frames: latency_frame,
         };
         stream::init(self, init_params, data_callback, state_callback, user_ptr)
     }
 
     fn register_device_collection_changed(
-        &self,
+        &mut self,
         _dev_type: DeviceType,
         _collection_changed_callback: ffi::cubeb_device_collection_changed_callback,
-        _user_ptr: *mut c_void
+        _user_ptr: *mut c_void,
     ) -> Result<()> {
         assert_not_in_callback();
         Ok(())
     }
 }
 
 impl Drop for ClientContext {
     fn drop(&mut self) {
-        info!("ClientContext drop...");
+        debug!("ClientContext drop...");
         let _ = send_recv!(self.rpc(), ClientDisconnect => ClientDisconnected);
         unsafe {
             if super::G_SERVER_FD.is_some() {
                 libc::close(super::G_SERVER_FD.take().unwrap());
             }
         }
     }
 }
--- a/media/audioipc/client/src/lib.rs
+++ b/media/audioipc/client/src/lib.rs
@@ -1,33 +1,32 @@
 // Copyright © 2017 Mozilla Foundation
 //
 // This program is made available under an ISC-style license.  See the
 // accompanying file LICENSE for details.
 
 extern crate audioipc;
 #[macro_use]
 extern crate cubeb_backend;
-extern crate cubeb_core;
+extern crate foreign_types;
 extern crate futures;
 extern crate futures_cpupool;
 extern crate libc;
 #[macro_use]
 extern crate log;
 extern crate tokio_core;
 extern crate tokio_uds;
 
 #[macro_use]
 mod send_recv;
 mod context;
 mod stream;
 
 use context::ClientContext;
-use cubeb_backend::capi;
-use cubeb_core::ffi;
+use cubeb_backend::{capi, ffi};
 use std::os::raw::{c_char, c_int};
 use std::os::unix::io::RawFd;
 use stream::ClientStream;
 
 thread_local!(static IN_CALLBACK: std::cell::RefCell<bool> = std::cell::RefCell::new(false));
 
 fn set_in_callback(in_callback: bool) {
     IN_CALLBACK.with(|b| {
@@ -44,17 +43,17 @@ fn assert_not_in_callback() {
 
 static mut G_SERVER_FD: Option<RawFd> = None;
 
 #[no_mangle]
 /// Entry point from C code.
 pub unsafe extern "C" fn audioipc_client_init(
     c: *mut *mut ffi::cubeb,
     context_name: *const c_char,
-    server_connection: c_int
+    server_connection: c_int,
 ) -> c_int {
     // TODO: Windows portability (for fd).
     // TODO: Better way to pass extra parameters to Context impl.
     if G_SERVER_FD.is_some() {
         panic!("audioipc client's server connection already initialized.");
     }
     if server_connection >= 0 {
         G_SERVER_FD = Some(server_connection);
--- a/media/audioipc/client/src/send_recv.rs
+++ b/media/audioipc/client/src/send_recv.rs
@@ -1,19 +1,19 @@
-use cubeb_core::Error;
-use cubeb_core::ffi;
+use cubeb_backend::Error;
+use std::os::raw::c_int;
 
 #[doc(hidden)]
 pub fn _err<E>(e: E) -> Error
 where
-    E: Into<Option<ffi::cubeb_error_code>>
+    E: Into<Option<c_int>>,
 {
     match e.into() {
         Some(e) => unsafe { Error::from_raw(e) },
-        None => Error::new()
+        None => Error::error(),
     }
 }
 
 #[macro_export]
 macro_rules! send_recv {
     ($rpc:expr, $smsg:ident => $rmsg:ident) => {{
         let resp = send_recv!(__send $rpc, $smsg);
         send_recv!(__recv resp, $rmsg)
--- a/media/audioipc/client/src/stream.rs
+++ b/media/audioipc/client/src/stream.rs
@@ -5,116 +5,135 @@
 
 use {assert_not_in_callback, set_in_callback};
 use ClientContext;
 use audioipc::codec::LengthDelimitedCodec;
 use audioipc::frame::{framed, Framed};
 use audioipc::messages::{self, CallbackReq, CallbackResp, ClientMessage, ServerMessage};
 use audioipc::rpc;
 use audioipc::shm::{SharedMemMutSlice, SharedMemSlice};
-use cubeb_backend::Stream;
-use cubeb_core::{ffi, Result};
+use cubeb_backend::{ffi, DeviceRef, Error, Result, Stream, StreamOps};
 use futures::Future;
 use futures_cpupool::{CpuFuture, CpuPool};
 use std::ffi::CString;
 use std::fs::File;
 use std::os::raw::c_void;
 use std::os::unix::io::FromRawFd;
 use std::os::unix::net;
 use std::ptr;
 use std::sync::mpsc;
 use tokio_uds::UnixStream;
 
 // TODO: Remove and let caller allocate based on cubeb backend requirements.
 const SHM_AREA_SIZE: usize = 2 * 1024 * 1024;
 
+pub struct Device(ffi::cubeb_device);
+
+impl Drop for Device {
+    fn drop(&mut self) {
+        unsafe {
+            if !self.0.input_name.is_null() {
+                let _ = CString::from_raw(self.0.input_name as *mut _);
+            }
+            if !self.0.output_name.is_null() {
+                let _ = CString::from_raw(self.0.output_name as *mut _);
+            }
+        }
+    }
+}
+
+#[derive(Debug)]
 pub struct ClientStream<'ctx> {
-    // This must be a reference to Context for cubeb, cubeb accesses stream methods via stream->context->ops
+    // This must be a reference to Context for cubeb, cubeb accesses
+    // stream methods via stream->context->ops
     context: &'ctx ClientContext,
-    token: usize
+    token: usize,
 }
 
 struct CallbackServer {
     input_shm: SharedMemSlice,
     output_shm: SharedMemMutSlice,
     data_cb: ffi::cubeb_data_callback,
     state_cb: ffi::cubeb_state_callback,
     user_ptr: usize,
-    cpu_pool: CpuPool
+    cpu_pool: CpuPool,
 }
 
 impl rpc::Server for CallbackServer {
     type Request = CallbackReq;
     type Response = CallbackResp;
     type Future = CpuFuture<Self::Response, ()>;
     type Transport = Framed<UnixStream, LengthDelimitedCodec<Self::Response, Self::Request>>;
 
     fn process(&mut self, req: Self::Request) -> Self::Future {
         match req {
             CallbackReq::Data(nframes, frame_size) => {
-                info!(
+                debug!(
                     "stream_thread: Data Callback: nframes={} frame_size={}",
-                    nframes,
-                    frame_size
+                    nframes, frame_size
                 );
 
                 // Clone values that need to be moved into the cpu pool thread.
                 let input_shm = unsafe { self.input_shm.clone_view() };
                 let mut output_shm = unsafe { self.output_shm.clone_view() };
                 let user_ptr = self.user_ptr;
-                let cb = self.data_cb;
+                let cb = self.data_cb.unwrap();
 
                 self.cpu_pool.spawn_fn(move || {
                     // TODO: This is proof-of-concept. Make it better.
                     let input_ptr: *const u8 = input_shm
                         .get_slice(nframes as usize * frame_size)
                         .unwrap()
                         .as_ptr();
                     let output_ptr: *mut u8 = output_shm
                         .get_mut_slice(nframes as usize * frame_size)
                         .unwrap()
                         .as_mut_ptr();
 
                     set_in_callback(true);
-                    let nframes = cb(
-                        ptr::null_mut(),
-                        user_ptr as *mut c_void,
-                        input_ptr as *const _,
-                        output_ptr as *mut _,
-                        nframes as _
-                    );
+                    let nframes = unsafe {
+                        cb(
+                            ptr::null_mut(),
+                            user_ptr as *mut c_void,
+                            input_ptr as *const _,
+                            output_ptr as *mut _,
+                            nframes as _,
+                        )
+                    };
                     set_in_callback(false);
 
                     Ok(CallbackResp::Data(nframes as isize))
                 })
-            },
+            }
             CallbackReq::State(state) => {
-                info!("stream_thread: State Callback: {:?}", state);
+                debug!("stream_thread: State Callback: {:?}", state);
                 let user_ptr = self.user_ptr;
-                let cb = self.state_cb;
+                let cb = self.state_cb.unwrap();
                 self.cpu_pool.spawn_fn(move || {
                     set_in_callback(true);
-                    cb(ptr::null_mut(), user_ptr as *mut _, state);
+                    unsafe {
+                        cb(ptr::null_mut(), user_ptr as *mut _, state);
+                    }
                     set_in_callback(false);
 
                     Ok(CallbackResp::State)
                 })
             }
         }
     }
 }
 
 impl<'ctx> ClientStream<'ctx> {
     fn init(
         ctx: &'ctx ClientContext,
         init_params: messages::StreamInitParams,
         data_callback: ffi::cubeb_data_callback,
         state_callback: ffi::cubeb_state_callback,
-        user_ptr: *mut c_void
-    ) -> Result<*mut ffi::cubeb_stream> {
+        user_ptr: *mut c_void,
+    ) -> Result<Stream> {
         assert_not_in_callback();
 
         let rpc = ctx.rpc();
         let data = try!(send_recv!(rpc, StreamInit(init_params) => StreamCreated()));
 
         trace!("token = {}, fds = {:?}", data.token, data.fds);
 
         let stm = data.fds[0];
@@ -133,124 +152,121 @@ impl<'ctx> ClientStream<'ctx> {
         let cpu_pool = ctx.cpu_pool();
 
         let server = CallbackServer {
             input_shm: input_shm,
             output_shm: output_shm,
             data_cb: data_callback,
             state_cb: state_callback,
             user_ptr: user_data,
-            cpu_pool: cpu_pool
+            cpu_pool: cpu_pool,
         };
 
         let (wait_tx, wait_rx) = mpsc::channel();
         ctx.remote().spawn(move |handle| {
             let stream = UnixStream::from_stream(stream, handle).unwrap();
             let transport = framed(stream, Default::default());
             rpc::bind_server(transport, server, handle);
             wait_tx.send(()).unwrap();
             Ok(())
         });
         wait_rx.recv().unwrap();
 
-        Ok(Box::into_raw(Box::new(ClientStream {
+        let stream = Box::into_raw(Box::new(ClientStream {
             context: ctx,
-            token: data.token
-        })) as _)
+            token: data.token,
+        }));
+        Ok(unsafe { Stream::from_ptr(stream as *mut _) })
     }
 }
 
 impl<'ctx> Drop for ClientStream<'ctx> {
     fn drop(&mut self) {
         trace!("ClientStream drop...");
         let rpc = self.context.rpc();
         let _ = send_recv!(rpc, StreamDestroy(self.token) => StreamDestroyed);
     }
 }
 
-impl<'ctx> Stream for ClientStream<'ctx> {
-    fn start(&self) -> Result<()> {
+impl<'ctx> StreamOps for ClientStream<'ctx> {
+    fn start(&mut self) -> Result<()> {
         assert_not_in_callback();
         let rpc = self.context.rpc();
         send_recv!(rpc, StreamStart(self.token) => StreamStarted)
     }
 
-    fn stop(&self) -> Result<()> {
+    fn stop(&mut self) -> Result<()> {
         assert_not_in_callback();
         let rpc = self.context.rpc();
         send_recv!(rpc, StreamStop(self.token) => StreamStopped)
     }
 
-    fn reset_default_device(&self) -> Result<()> {
+    fn reset_default_device(&mut self) -> Result<()> {
         assert_not_in_callback();
         let rpc = self.context.rpc();
         send_recv!(rpc, StreamResetDefaultDevice(self.token) => StreamDefaultDeviceReset)
     }
 
-    fn position(&self) -> Result<u64> {
+    fn position(&mut self) -> Result<u64> {
         assert_not_in_callback();
         let rpc = self.context.rpc();
         send_recv!(rpc, StreamGetPosition(self.token) => StreamPosition())
     }
 
-    fn latency(&self) -> Result<u32> {
+    fn latency(&mut self) -> Result<u32> {
         assert_not_in_callback();
         let rpc = self.context.rpc();
         send_recv!(rpc, StreamGetLatency(self.token) => StreamLatency())
     }
 
-    fn set_volume(&self, volume: f32) -> Result<()> {
+    fn set_volume(&mut self, volume: f32) -> Result<()> {
         assert_not_in_callback();
         let rpc = self.context.rpc();
         send_recv!(rpc, StreamSetVolume(self.token, volume) => StreamVolumeSet)
     }
 
-    fn set_panning(&self, panning: f32) -> Result<()> {
+    fn set_panning(&mut self, panning: f32) -> Result<()> {
         assert_not_in_callback();
         let rpc = self.context.rpc();
         send_recv!(rpc, StreamSetPanning(self.token, panning) => StreamPanningSet)
     }
 
-    fn current_device(&self) -> Result<*const ffi::cubeb_device> {
+    fn current_device(&mut self) -> Result<&DeviceRef> {
         assert_not_in_callback();
         let rpc = self.context.rpc();
         match send_recv!(rpc, StreamGetCurrentDevice(self.token) => StreamCurrentDevice()) {
-            Ok(d) => Ok(Box::into_raw(Box::new(d.into()))),
-            Err(e) => Err(e)
+            Ok(d) => Ok(unsafe { DeviceRef::from_ptr(Box::into_raw(Box::new(d.into()))) }),
+            Err(e) => Err(e),
         }
     }
 
-    fn device_destroy(&self, device: *const ffi::cubeb_device) -> Result<()> {
+    fn device_destroy(&mut self, device: &DeviceRef) -> Result<()> {
         assert_not_in_callback();
         // It's all unsafe...
-        if !device.is_null() {
+        if device.as_ptr().is_null() {
+            Err(Error::error())
+        } else {
             unsafe {
-                if !(*device).output_name.is_null() {
-                    let _ = CString::from_raw((*device).output_name as *mut _);
-                }
-                if !(*device).input_name.is_null() {
-                    let _ = CString::from_raw((*device).input_name as *mut _);
-                }
-                let _: Box<ffi::cubeb_device> = Box::from_raw(device as *mut _);
+                let _: Box<Device> = Box::from_raw(device.as_ptr() as *mut _);
             }
+            Ok(())
         }
-        Ok(())
     }
 
     // TODO: How do we call this back? On what thread?
     fn register_device_changed_callback(
-        &self,
-        _device_changed_callback: ffi::cubeb_device_changed_callback
+        &mut self,
+        _device_changed_callback: ffi::cubeb_device_changed_callback,
     ) -> Result<()> {
         assert_not_in_callback();
         Ok(())
     }
 }
 
 pub fn init(
     ctx: &ClientContext,
     init_params: messages::StreamInitParams,
     data_callback: ffi::cubeb_data_callback,
     state_callback: ffi::cubeb_state_callback,
-    user_ptr: *mut c_void
-) -> Result<*mut ffi::cubeb_stream> {
+    user_ptr: *mut c_void,
+) -> Result<Stream> {
     ClientStream::init(ctx, init_params, data_callback, state_callback, user_ptr)
 }
--- a/media/audioipc/gecko.patch
+++ b/media/audioipc/gecko.patch
@@ -7,54 +7,9 @@ Subject: gecko: Change paths to vendored
 diff --git a/media/audioipc/Cargo.toml b/media/audioipc/Cargo.toml
 index ede6064..d0a1979 100644
 --- a/media/audioipc/Cargo.toml
 +++ b/media/audioipc/Cargo.toml
 @@ -1,2 +1,2 @@
  [workspace]
 -members = ["audioipc", "client", "server", "ipctest"]
 +members = ["audioipc", "client", "server"]
-diff --git a/audioipc/Cargo.toml b/audioipc/Cargo.toml
-index 669c6ff..308cb5c 100644
---- a/media/audioipc/audioipc/Cargo.toml
-+++ b/media/audioipc/audioipc/Cargo.toml
-@@ -8,7 +8,7 @@ authors = [
- description = "Remote Cubeb IPC"
- 
- [dependencies]
--cubeb-core = { git = "https://github.com/djg/cubeb-rs", version="^0.1" }
-+cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
- bincode = "0.8"
- bytes = "0.4"
- # rayon-core in Gecko uses futures 0.1.13
-diff --git a/client/Cargo.toml b/client/Cargo.toml
-index c81b19a..9e3f8a5 100644
---- a/media/audioipc/client/Cargo.toml
-+++ b/media/audioipc/client/Cargo.toml
-@@ -9,8 +9,8 @@ description = "Cubeb Backend for talking to remote cubeb server."
- 
- [dependencies]
- audioipc = { path="../audioipc" }
--cubeb-backend = { git="https://github.com/djg/cubeb-rs", version="^0.2" }
--cubeb-core = { git="https://github.com/djg/cubeb-rs", version="^0.1" }
-+cubeb-backend = { path = "../../cubeb-rs/cubeb-backend" }
-+cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
- # rayon-core in Gecko uses futures 0.1.13
- futures = { version="=0.1.13", default-features=false, features=["use_std"] }
- # futures-cpupool 0.1.5 matches futures 0.1.13
-diff --git a/server/Cargo.toml b/server/Cargo.toml
-index 5b79b83..01463be 100644
---- a/media/audioipc/server/Cargo.toml
-+++ b/media/audioipc/server/Cargo.toml
-@@ -9,8 +9,8 @@ description = "Remote cubeb server"
- 
- [dependencies]
- audioipc = { path = "../audioipc" }
--cubeb-core = { git = "https://github.com/djg/cubeb-rs", version="^0.1" }
--cubeb = { git = "https://github.com/djg/cubeb-rs", version="^0.3.2" }
-+cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
-+cubeb = { path = "../../cubeb-rs/cubeb-api" }
- bytes = "0.4"
- lazycell = "^0.4"
- libc = "0.2"
--- 
-2.10.2
 
--- a/media/audioipc/server/Cargo.toml
+++ b/media/audioipc/server/Cargo.toml
@@ -4,18 +4,17 @@ version = "0.2.1"
 authors = [
         "Matthew Gregan <kinetik@flim.org>",
         "Dan Glastonbury <dan.glastonbury@gmail.com>"
         ]
 description = "Remote cubeb server"
 
 [dependencies]
 audioipc = { path = "../audioipc" }
-cubeb-core = { path = "../../cubeb-rs/cubeb-core" }
-cubeb = { path = "../../cubeb-rs/cubeb-api" }
+cubeb = "0.4"
 bytes = "0.4"
 lazycell = "^0.4"
 libc = "0.2"
 log = "^0.3.6"
 slab = "0.3.0"
 # rayon-core in Gecko uses futures 0.1.13
 futures = "=0.1.13"
 tokio-core = "0.1"
--- a/media/audioipc/server/src/lib.rs
+++ b/media/audioipc/server/src/lib.rs
@@ -2,34 +2,31 @@
 extern crate error_chain;
 
 #[macro_use]
 extern crate log;
 
 extern crate audioipc;
 extern crate bytes;
 extern crate cubeb;
-extern crate cubeb_core;
 extern crate futures;
 extern crate lazycell;
 extern crate libc;
 extern crate slab;
 extern crate tokio_core;
 extern crate tokio_uds;
 
 use audioipc::codec::LengthDelimitedCodec;
 use audioipc::core;
 use audioipc::fd_passing::{framed_with_fds, FramedWithFds};
 use audioipc::frame::{framed, Framed};
-use audioipc::messages::{CallbackReq, CallbackResp, ClientMessage, DeviceInfo, ServerMessage,
-                         StreamCreate, StreamInitParams, StreamParams};
+use audioipc::messages::{CallbackReq, CallbackResp, ClientMessage, Device, DeviceInfo,
+                         ServerMessage, StreamCreate, StreamInitParams, StreamParams};
 use audioipc::rpc;
 use audioipc::shm::{SharedMemReader, SharedMemWriter};
-use cubeb_core::binding::Binding;
-use cubeb_core::ffi;
 use futures::Future;
 use futures::future::{self, FutureResult};
 use futures::sync::oneshot;
 use std::{ptr, slice};
 use std::cell::RefCell;
 use std::convert::From;
 use std::error::Error;
 use std::os::raw::c_void;
@@ -39,35 +36,36 @@ use tokio_core::reactor::Remote;
 use tokio_uds::UnixStream;
 
 pub mod errors {
     error_chain! {
         links {
             AudioIPC(::audioipc::errors::Error, ::audioipc::errors::ErrorKind);
         }
         foreign_links {
-            Cubeb(::cubeb_core::Error);
+            Cubeb(::cubeb::Error);
             Io(::std::io::Error);
             Canceled(::futures::sync::oneshot::Canceled);
         }
     }
 }
 
 use errors::*;
 
-thread_local!(static CONTEXT_KEY: RefCell<Option<cubeb::Result<cubeb::Context>>> = RefCell::new(None));
+type ContextKey = RefCell<Option<cubeb::Result<cubeb::Context>>>;
+thread_local!(static CONTEXT_KEY:ContextKey = RefCell::new(None));
 
 fn with_local_context<T, F>(f: F) -> T
 where
-    F: FnOnce(&cubeb::Result<cubeb::Context>) -> T
+    F: FnOnce(&cubeb::Result<cubeb::Context>) -> T,
 {
     CONTEXT_KEY.with(|k| {
         let mut context = k.borrow_mut();
         if context.is_none() {
-            *context = Some(cubeb::Context::init("AudioIPC Server", None));
+            *context = Some(cubeb::init("AudioIPC Server"));
         }
         f(context.as_ref().unwrap())
     })
 }
 
 // TODO: Remove and let caller allocate based on cubeb backend requirements.
 const SHM_AREA_SIZE: usize = 2 * 1024 * 1024;
 
@@ -77,130 +75,56 @@ const STREAM_CONN_CHUNK_SIZE: usize = 64
 struct CallbackClient;
 
 impl rpc::Client for CallbackClient {
     type Request = CallbackReq;
     type Response = CallbackResp;
     type Transport = Framed<UnixStream, LengthDelimitedCodec<Self::Request, Self::Response>>;
 }
 
-// TODO: this should forward to the client.
-struct Callback {
-    /// Size of input frame in bytes
-    input_frame_size: u16,
-    /// Size of output frame in bytes
-    output_frame_size: u16,
-    input_shm: SharedMemWriter,
-    output_shm: SharedMemReader,
-    rpc: rpc::ClientProxy<CallbackReq, CallbackResp>
-}
-
-impl cubeb::StreamCallback for Callback {
-    type Frame = u8;
-
-    fn data_callback(&mut self, input: &[u8], output: &mut [u8]) -> isize {
-        trace!("Stream data callback: {} {}", input.len(), output.len());
-
-        // len is of input and output is frame len. Turn these into the real lengths.
-        let real_input = unsafe {
-            let size_bytes = input.len() * self.input_frame_size as usize;
-            slice::from_raw_parts(input.as_ptr(), size_bytes)
-        };
-        let real_output = unsafe {
-            let size_bytes = output.len() * self.output_frame_size as usize;
-            trace!("Resize output to {}", size_bytes);
-            slice::from_raw_parts_mut(output.as_mut_ptr(), size_bytes)
-        };
-
-        self.input_shm.write(real_input).unwrap();
-
-        let r = self.rpc
-            .call(CallbackReq::Data(
-                output.len() as isize,
-                self.output_frame_size as usize
-            ))
-            .wait();
-
-        match r {
-            Ok(CallbackResp::Data(cb_result)) => if cb_result >= 0 {
-                let len = cb_result as usize * self.output_frame_size as usize;
-                self.output_shm.read(&mut real_output[..len]).unwrap();
-                cb_result
-            } else {
-                cb_result
-            },
-            _ => {
-                debug!("Unexpected message {:?} during data_callback", r);
-                -1
-            }
-        }
-    }
-
-    fn state_callback(&mut self, state: cubeb::State) {
-        info!("Stream state callback: {:?}", state);
-        // TODO: Share this conversion with the same code in cubeb-rs?
-        let state = match state {
-            cubeb::State::Started => ffi::CUBEB_STATE_STARTED,
-            cubeb::State::Stopped => ffi::CUBEB_STATE_STOPPED,
-            cubeb::State::Drained => ffi::CUBEB_STATE_DRAINED,
-            cubeb::State::Error => ffi::CUBEB_STATE_ERROR
-        };
-
-        let r = self.rpc.call(CallbackReq::State(state)).wait();
-
-        match r {
-            Ok(CallbackResp::State) => {},
-            _ => {
-                debug!("Unexpected message {:?} during callback", r);
-            }
-        };
-    }
-}
-
-type StreamSlab = slab::Slab<cubeb::Stream<Callback>, usize>;
+type StreamSlab = slab::Slab<cubeb::Stream<u8>, usize>;
 
 pub struct CubebServer {
     cb_remote: Remote,
-    streams: StreamSlab
+    streams: StreamSlab,
 }
 
 impl rpc::Server for CubebServer {
     type Request = ServerMessage;
     type Response = ClientMessage;
     type Future = FutureResult<Self::Response, ()>;
-    type Transport =
-        FramedWithFds<UnixStream, LengthDelimitedCodec<Self::Response, Self::Request>>;
+    type Transport = FramedWithFds<UnixStream, LengthDelimitedCodec<Self::Response, Self::Request>>;
 
     fn process(&mut self, req: Self::Request) -> Self::Future {
         let resp = with_local_context(|context| match *context {
-            Err(_) => error(cubeb::Error::new()),
-            Ok(ref context) => self.process_msg(context, &req)
+            Err(_) => error(cubeb::Error::error()),
+            Ok(ref context) => self.process_msg(context, &req),
         });
         future::ok(resp)
     }
 }
 
 impl CubebServer {
     pub fn new(cb_remote: Remote) -> Self {
         CubebServer {
             cb_remote: cb_remote,
-            streams: StreamSlab::with_capacity(STREAM_CONN_CHUNK_SIZE)
+            streams: StreamSlab::with_capacity(STREAM_CONN_CHUNK_SIZE),
         }
     }
 
     // Process a request coming from the client.
     fn process_msg(&mut self, context: &cubeb::Context, msg: &ServerMessage) -> ClientMessage {
         let resp: ClientMessage = match *msg {
             ServerMessage::ClientConnect => panic!("already connected"),
 
             ServerMessage::ClientDisconnect => {
                 // TODO:
                 //self.connection.client_disconnect();
                 ClientMessage::ClientDisconnected
-            },
+            }
 
             ServerMessage::ContextGetBackendId => ClientMessage::ContextBackendId(),
 
             ServerMessage::ContextGetMaxChannelCount => context
                 .max_channel_count()
                 .map(ClientMessage::ContextMaxChannelCount)
                 .unwrap_or_else(error),
 
@@ -214,58 +138,58 @@ impl CubebServer {
                     .channels(u32::from(params.channels))
                     .layout(layout)
                     .take();
 
                 context
                     .min_latency(&params)
                     .map(ClientMessage::ContextMinLatency)
                     .unwrap_or_else(error)
-            },
+            }
 
             ServerMessage::ContextGetPreferredSampleRate => context
                 .preferred_sample_rate()
                 .map(ClientMessage::ContextPreferredSampleRate)
                 .unwrap_or_else(error),
 
             ServerMessage::ContextGetPreferredChannelLayout => context
                 .preferred_channel_layout()
                 .map(|l| ClientMessage::ContextPreferredChannelLayout(l as _))
                 .unwrap_or_else(error),
 
             ServerMessage::ContextGetDeviceEnumeration(device_type) => context
                 .enumerate_devices(cubeb::DeviceType::from_bits_truncate(device_type))
                 .map(|devices| {
-                    let v: Vec<DeviceInfo> = devices.iter().map(|i| i.raw().into()).collect();
+                    let v: Vec<DeviceInfo> = devices.iter().map(|i| i.as_ref().into()).collect();
                     ClientMessage::ContextEnumeratedDevices(v)
                 })
                 .unwrap_or_else(error),
 
             ServerMessage::StreamInit(ref params) => self.process_stream_init(context, params)
-                .unwrap_or_else(|_| error(cubeb::Error::new())),
+                .unwrap_or_else(|_| error(cubeb::Error::error())),
 
             ServerMessage::StreamDestroy(stm_tok) => {
                 self.streams.remove(stm_tok);
                 ClientMessage::StreamDestroyed
-            },
+            }
 
             ServerMessage::StreamStart(stm_tok) => {
                 let _ = self.streams[stm_tok].start();
                 ClientMessage::StreamStarted
-            },
+            }
 
             ServerMessage::StreamStop(stm_tok) => {
                 let _ = self.streams[stm_tok].stop();
                 ClientMessage::StreamStopped
-            },
+            }
 
             ServerMessage::StreamResetDefaultDevice(stm_tok) => {
                 let _ = self.streams[stm_tok].reset_default_device();
                 ClientMessage::StreamDefaultDeviceReset
-            },
+            }
 
             ServerMessage::StreamGetPosition(stm_tok) => self.streams[stm_tok]
                 .position()
                 .map(ClientMessage::StreamPosition)
                 .unwrap_or_else(error),
 
             ServerMessage::StreamGetLatency(stm_tok) => self.streams[stm_tok]
                 .latency()
@@ -279,86 +203,60 @@ impl CubebServer {
 
             ServerMessage::StreamSetPanning(stm_tok, panning) => self.streams[stm_tok]
                 .set_panning(panning)
                 .map(|_| ClientMessage::StreamPanningSet)
                 .unwrap_or_else(error),
 
             ServerMessage::StreamGetCurrentDevice(stm_tok) => self.streams[stm_tok]
                 .current_device()
-                .map(|device| ClientMessage::StreamCurrentDevice(device.into()))
-                .unwrap_or_else(error)
+                .map(|device| ClientMessage::StreamCurrentDevice(Device::from(device)))
+                .unwrap_or_else(error),
         };
 
         debug!("process_msg: req={:?}, resp={:?}", msg, resp);
 
         resp
     }
 
     // Stream init is special, so it's been separated from process_msg.
     fn process_stream_init(
         &mut self,
         context: &cubeb::Context,
-        params: &StreamInitParams
+        params: &StreamInitParams,
     ) -> Result<ClientMessage> {
-        fn opt_stream_params(params: Option<&StreamParams>) -> Option<cubeb::StreamParams> {
-            params.and_then(|p| {
-                let raw = ffi::cubeb_stream_params::from(p);
-                Some(unsafe { cubeb::StreamParams::from_raw(&raw as *const _) })
-            })
-        }
-
-        fn frame_size_in_bytes(params: Option<cubeb::StreamParams>) -> u16 {
+        fn frame_size_in_bytes(params: Option<&StreamParams>) -> u16 {
             params
                 .map(|p| {
-                    let sample_size = match p.format() {
+                    let format = p.format.into();
+                    let sample_size = match format {
                         cubeb::SampleFormat::S16LE
                         | cubeb::SampleFormat::S16BE
                         | cubeb::SampleFormat::S16NE => 2,
                         cubeb::SampleFormat::Float32LE
                         | cubeb::SampleFormat::Float32BE
-                        | cubeb::SampleFormat::Float32NE => 4
+                        | cubeb::SampleFormat::Float32NE => 4,
                     };
-                    let channel_count = p.channels() as u16;
+                    let channel_count = p.channels as u16;
                     sample_size * channel_count
                 })
                 .unwrap_or(0u16)
         }
 
         // TODO: Yuck!
-        let input_device =
-            unsafe { cubeb::DeviceId::from_raw(params.input_device as *const _) };
-        let output_device =
-            unsafe { cubeb::DeviceId::from_raw(params.output_device as *const _) };
+        let input_device = params.input_device as *const _;
+        let output_device = params.output_device as *const _;
         let latency = params.latency_frames;
-        let mut builder = cubeb::StreamInitOptionsBuilder::new();
-        builder
-            .input_device(input_device)
-            .output_device(output_device)
-            .latency(latency);
 
-        if let Some(ref stream_name) = params.stream_name {
-            builder.stream_name(stream_name);
-        }
-        let input_stream_params = opt_stream_params(params.input_stream_params.as_ref());
-        if let Some(ref isp) = input_stream_params {
-            builder.input_stream_param(isp);
-        }
-        let output_stream_params = opt_stream_params(params.output_stream_params.as_ref());
-        if let Some(ref osp) = output_stream_params {
-            builder.output_stream_param(osp);
-        }
-        let params = builder.take();
-
-        let input_frame_size = frame_size_in_bytes(input_stream_params);
-        let output_frame_size = frame_size_in_bytes(output_stream_params);
+        let input_frame_size = frame_size_in_bytes(params.input_stream_params.as_ref());
+        let output_frame_size = frame_size_in_bytes(params.output_stream_params.as_ref());
 
         let (stm1, stm2) = net::UnixStream::pair()?;
-        info!("Created callback pair: {:?}-{:?}", stm1, stm2);
-        let (input_shm, input_file) =
+        debug!("Created callback pair: {:?}-{:?}", stm1, stm2);
+        let (mut input_shm, input_file) =
             SharedMemWriter::new(&audioipc::get_shm_path("input"), SHM_AREA_SIZE)?;
         let (output_shm, output_file) =
             SharedMemReader::new(&audioipc::get_shm_path("output"), SHM_AREA_SIZE)?;
 
         // This code is currently running on the Client/Server RPC
         // handling thread.  We need to move the registration of the
         // bind_client to the callback RPC handling thread.  This is
         // done by spawning a future on cb_remote.
@@ -372,69 +270,128 @@ impl CubebServer {
             assert_ne!(id, handle.id());
             let stream = UnixStream::from_stream(stm2, handle).unwrap();
             let transport = framed(stream, Default::default());
             let rpc = rpc::bind_client::<CallbackClient>(transport, handle);
             drop(tx.send(rpc));
             Ok(())
         });
 
-        let rpc: rpc::ClientProxy<CallbackReq, CallbackResp> = match rx.wait() {
+        let rpc_data: rpc::ClientProxy<CallbackReq, CallbackResp> = match rx.wait() {
             Ok(rpc) => rpc,
-            Err(_) => bail!("Failed to create callback rpc.")
+            Err(_) => bail!("Failed to create callback rpc."),
         };
+        let rpc_state = rpc_data.clone();
+
+        let mut builder = cubeb::StreamBuilder::new();
+
+        if let Some(ref stream_name) = params.stream_name {
+            builder.name(stream_name.clone());
+        }
+
+        if let Some(ref isp) = params.input_stream_params {
+            let input_stream_params =
+                unsafe { cubeb::StreamParamsRef::from_ptr(isp as *const StreamParams as *mut _) };
+            builder.input(input_device, input_stream_params);
+        }
+
+        if let Some(ref osp) = params.output_stream_params {
+            let output_stream_params =
+                unsafe { cubeb::StreamParamsRef::from_ptr(osp as *const StreamParams as *mut _) };
+            builder.output(output_device, output_stream_params);
+        }
+
+        builder
+            .latency(latency)
+            .data_callback(move |input, output| {
+                trace!("Stream data callback: {} {}", input.len(), output.len());
+
+                // len is of input and output is frame len. Turn these into the real lengths.
+                let real_input = unsafe {
+                    let nbytes = input.len() * input_frame_size as usize;
+                    slice::from_raw_parts(input.as_ptr(), nbytes)
+                };
+
+                input_shm.write(real_input).unwrap();
 
-        context
-            .stream_init(
-                &params,
-                Callback {
-                    input_frame_size: input_frame_size,
-                    output_frame_size: output_frame_size,
-                    input_shm: input_shm,
-                    output_shm: output_shm,
-                    rpc: rpc
+                let r = rpc_data
+                    .call(CallbackReq::Data(
+                        output.len() as isize,
+                        output_frame_size as usize,
+                    ))
+                    .wait();
+
+                match r {
+                    Ok(CallbackResp::Data(frames)) => {
+                        if frames >= 0 {
+                            let nbytes = frames as usize * output_frame_size as usize;
+                            let real_output = unsafe {
+                                trace!("Resize output to {}", nbytes);
+                                slice::from_raw_parts_mut(output.as_mut_ptr(), nbytes)
+                            };
+                            output_shm.read(&mut real_output[..nbytes]).unwrap();
+                        }
+                        frames
+                    }
+                    _ => {
+                        debug!("Unexpected message {:?} during data_callback", r);
+                        -1
+                    }
                 }
-            )
+            })
+            .state_callback(move |state| {
+                trace!("Stream state callback: {:?}", state);
+                let r = rpc_state.call(CallbackReq::State(state.into())).wait();
+                match r {
+                    Ok(CallbackResp::State) => {}
+                    _ => {
+                        debug!("Unexpected message {:?} during callback", r);
+                    }
+                }
+            });
+
+        builder
+            .init(context)
             .and_then(|stream| {
                 if !self.streams.has_available() {
                     trace!(
                         "server connection ran out of stream slots. reserving {} more.",
                         STREAM_CONN_CHUNK_SIZE
                     );
                     self.streams.reserve_exact(STREAM_CONN_CHUNK_SIZE);
                 }
 
                 let stm_tok = match self.streams.vacant_entry() {
                     Some(entry) => {
                         debug!("Registering stream {:?}", entry.index(),);
 
                         entry.insert(stream).index()
-                    },
+                    }
                     None => {
                         // TODO: Turn into error
                         panic!("Failed to insert stream into slab. No entries")
                     }
                 };
 
                 Ok(ClientMessage::StreamCreated(StreamCreate {
                     token: stm_tok,
                     fds: [
                         stm1.into_raw_fd(),
                         input_file.into_raw_fd(),
                         output_file.into_raw_fd(),
-                    ]
+                    ],
                 }))
             })
             .map_err(|e| e.into())
     }
 }
 
 struct ServerWrapper {
     core_thread: core::CoreThread,
-    callback_thread: core::CoreThread
+    callback_thread: core::CoreThread,
 }
 
 fn run() -> Result<ServerWrapper> {
     trace!("Starting up cubeb audio server event loop thread...");
 
     let callback_thread = try!(
         core::spawn_thread("AudioIPC Callback RPC", || {
             trace!("Starting up cubeb audio callback event loop thread...");
@@ -455,25 +412,25 @@ fn run() -> Result<ServerWrapper> {
                 e.description()
             );
             Err(e)
         })
     );
 
     Ok(ServerWrapper {
         core_thread: core_thread,
-        callback_thread: callback_thread
+        callback_thread: callback_thread,
     })
 }
 
 #[no_mangle]
 pub extern "C" fn audioipc_server_start() -> *mut c_void {
     match run() {
         Ok(server) => Box::into_raw(Box::new(server)) as *mut _,
-        Err(_) => ptr::null_mut() as *mut _
+        Err(_) => ptr::null_mut() as *mut _,
     }
 }
 
 #[no_mangle]
 pub extern "C" fn audioipc_server_new_client(p: *mut c_void) -> libc::c_int {
     let (wait_tx, wait_rx) = oneshot::channel();
     let wrapper: &ServerWrapper = unsafe { &*(p as *mut _) };