Adding upstream version 86.0.1.upstream/86.0.1upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

38 files changed, 6057 insertions, 0 deletions

diff --git a/media/audioipc/Cargo.toml b/media/audioipc/Cargo.toml
new file mode 100644
index 0000000000..64d3e2bdb1
--- /dev/null
+++ b/media/audioipc/Cargo.toml
@@ -0,0 +1,3 @@
+[workspace]
+members = ["audioipc", "client", "server"]
+
diff --git a/media/audioipc/README.md b/media/audioipc/README.md
new file mode 100644
index 0000000000..ecbf259ec6
--- /dev/null
+++ b/media/audioipc/README.md
@@ -0,0 +1 @@
+# Cubeb Audio Remoting Prototype
diff --git a/media/audioipc/README_MOZILLA b/media/audioipc/README_MOZILLA
new file mode 100644
index 0000000000..df76f22059
--- /dev/null
+++ b/media/audioipc/README_MOZILLA
@@ -0,0 +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 db32323c92b2b4f5ca2e702dcf987fedc1e71768 (2020-10-21 09:02:32 +1300)
diff --git a/media/audioipc/audioipc/Cargo.toml b/media/audioipc/audioipc/Cargo.toml
new file mode 100644
index 0000000000..95441e345f
--- /dev/null
+++ b/media/audioipc/audioipc/Cargo.toml
@@ -0,0 +1,42 @@
+[package]
+name = "audioipc"
+version = "0.2.5"
+authors = [
+        "Matthew Gregan <kinetik@flim.org>",
+        "Dan Glastonbury <dan.glastonbury@gmail.com>"
+        ]
+description = "Remote Cubeb IPC"
+edition = "2018"
+
+[dependencies]
+bincode = "1"
+bytes = "0.4"
+cubeb = "0.8"
+futures = "0.1.29"
+log = "0.4"
+memmap = "0.7"
+serde = "1"
+serde_derive = "1"
+tokio = "0.1"
+tokio-io = "0.1"
+audio_thread_priority = "0.23.4"
+
+[target.'cfg(unix)'.dependencies]
+iovec = "0.1"
+libc = "0.2"
+mio = "0.6.19"
+mio-uds = "0.6.7"
+tokio-reactor = "0.1"
+
+[target.'cfg(windows)'.dependencies]
+mio = "0.6.19"
+miow = "0.3.3"
+mio-named-pipes = { git = "https://github.com/kinetiknz/mio-named-pipes", rev = "21c26326f5f45f415c49eac4ba5bc41a2f961321" }
+winapi = { version = "0.3.6", features = ["combaseapi", "objbase"] }
+
+[dependencies.error-chain]
+version = "0.11.0"
+default-features = false
+
+[build-dependencies]
+cc = "1.0"
diff --git a/media/audioipc/audioipc/src/async_msg.rs b/media/audioipc/audioipc/src/async_msg.rs
new file mode 100644
index 0000000000..fe682a71ed
--- /dev/null
+++ b/media/audioipc/audioipc/src/async_msg.rs
@@ -0,0 +1,168 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+//! Various async helpers modelled after futures-rs and tokio-io.
+
+#[cfg(unix)]
+use crate::msg::{RecvMsg, SendMsg};
+use bytes::{Buf, BufMut};
+#[cfg(unix)]
+use futures::Async;
+use futures::Poll;
+#[cfg(unix)]
+use iovec::IoVec;
+#[cfg(unix)]
+use mio::Ready;
+use std::io;
+use tokio_io::{AsyncRead, AsyncWrite};
+
+pub trait AsyncRecvMsg: AsyncRead {
+    /// Pull some bytes from this source into the specified `Buf`, returning
+    /// how many bytes were read.
+    ///
+    /// The `buf` provided will have bytes read into it and the internal cursor
+    /// will be advanced if any bytes were read. Note that this method typically
+    /// will not reallocate the buffer provided.
+    fn recv_msg_buf<B>(&mut self, buf: &mut B, cmsg: &mut B) -> Poll<(usize, i32), io::Error>
+    where
+        Self: Sized,
+        B: BufMut;
+}
+
+/// A trait for writable objects which operated in an async fashion.
+///
+/// This trait inherits from `std::io::Write` and indicates that an I/O object is
+/// **nonblocking**, meaning that it will return an error instead of blocking
+/// when bytes cannot currently be written, but hasn't closed. Specifically
+/// this means that the `write` function for types that implement this trait
+/// can have a few return values:
+///
+/// * `Ok(n)` means that `n` bytes of data was immediately written .
+/// * `Err(e) if e.kind() == ErrorKind::WouldBlock` means that no data was
+///   written from the buffer provided. The I/O object is not currently
+///   writable but may become writable in the future.
+/// * `Err(e)` for other errors are standard I/O errors coming from the
+///   underlying object.
+pub trait AsyncSendMsg: AsyncWrite {
+    /// Write a `Buf` into this value, returning how many bytes were written.
+    ///
+    /// Note that this method will advance the `buf` provided automatically by
+    /// the number of bytes written.
+    fn send_msg_buf<B, C>(&mut self, buf: &mut B, cmsg: &C) -> Poll<usize, io::Error>
+    where
+        Self: Sized,
+        B: Buf,
+        C: Buf;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+#[cfg(unix)]
+impl AsyncRecvMsg for super::AsyncMessageStream {
+    fn recv_msg_buf<B>(&mut self, buf: &mut B, cmsg: &mut B) -> Poll<(usize, i32), io::Error>
+    where
+        B: BufMut,
+    {
+        if let Async::NotReady =
+            <super::AsyncMessageStream>::poll_read_ready(self, Ready::readable())?
+        {
+            return Ok(Async::NotReady);
+        }
+        let r = unsafe {
+            // The `IoVec` type can't have a 0-length size, so we create a bunch
+            // of dummy versions on the stack with 1 length which we'll quickly
+            // overwrite.
+            let b1: &mut [u8] = &mut [0];
+            let b2: &mut [u8] = &mut [0];
+            let b3: &mut [u8] = &mut [0];
+            let b4: &mut [u8] = &mut [0];
+            let b5: &mut [u8] = &mut [0];
+            let b6: &mut [u8] = &mut [0];
+            let b7: &mut [u8] = &mut [0];
+            let b8: &mut [u8] = &mut [0];
+            let b9: &mut [u8] = &mut [0];
+            let b10: &mut [u8] = &mut [0];
+            let b11: &mut [u8] = &mut [0];
+            let b12: &mut [u8] = &mut [0];
+            let b13: &mut [u8] = &mut [0];
+            let b14: &mut [u8] = &mut [0];
+            let b15: &mut [u8] = &mut [0];
+            let b16: &mut [u8] = &mut [0];
+            let mut bufs: [&mut IoVec; 16] = [
+                b1.into(),
+                b2.into(),
+                b3.into(),
+                b4.into(),
+                b5.into(),
+                b6.into(),
+                b7.into(),
+                b8.into(),
+                b9.into(),
+                b10.into(),
+                b11.into(),
+                b12.into(),
+                b13.into(),
+                b14.into(),
+                b15.into(),
+                b16.into(),
+            ];
+            let n = buf.bytes_vec_mut(&mut bufs);
+            self.recv_msg(&mut bufs[..n], cmsg.bytes_mut())
+        };
+
+        match r {
+            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.clear_read_ready(mio::Ready::readable())?;
+                Ok(Async::NotReady)
+            }
+            Err(e) => Err(e),
+        }
+    }
+}
+
+#[cfg(unix)]
+impl AsyncSendMsg for super::AsyncMessageStream {
+    fn send_msg_buf<B, C>(&mut self, buf: &mut B, cmsg: &C) -> Poll<usize, io::Error>
+    where
+        B: Buf,
+        C: Buf,
+    {
+        if let Async::NotReady = <super::AsyncMessageStream>::poll_write_ready(self)? {
+            return Ok(Async::NotReady);
+        }
+        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,
+            ];
+            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.clear_write_ready()?;
+                Ok(Async::NotReady)
+            }
+            Err(e) => Err(e),
+        }
+    }
+}
diff --git a/media/audioipc/audioipc/src/cmsg.rs b/media/audioipc/audioipc/src/cmsg.rs
new file mode 100644
index 0000000000..19adb2a4fb
--- /dev/null
+++ b/media/audioipc/audioipc/src/cmsg.rs
@@ -0,0 +1,178 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use bytes::{BufMut, Bytes, BytesMut};
+use libc::{self, cmsghdr};
+use std::convert::TryInto;
+use std::os::unix::io::RawFd;
+use std::{convert, mem, ops, slice};
+
+#[derive(Clone, Debug)]
+pub struct Fds {
+    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) }
+    }
+}
+
+impl ops::Deref for Fds {
+    type Target = [RawFd];
+
+    #[inline]
+    fn deref(&self) -> &[RawFd] {
+        self.as_ref()
+    }
+}
+
+pub struct ControlMsgIter {
+    control: Bytes,
+}
+
+pub fn iterator(c: Bytes) -> ControlMsgIter {
+    ControlMsgIter { control: c }
+}
+
+impl Iterator for ControlMsgIter {
+    type Item = Fds;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        loop {
+            let control = self.control.clone();
+            let cmsghdr_len = len(0);
+
+            if control.len() < cmsghdr_len {
+                // No more entries---not enough data in `control` for a
+                // complete message.
+                return None;
+            }
+
+            let cmsg: &cmsghdr = unsafe { &*(control.as_ptr() as *const _) };
+            // The offset to the next cmsghdr in control.  This must be
+            // aligned to a boundary that matches the type used to
+            // represent the length of the message.
+            let cmsg_len = cmsg.cmsg_len as usize;
+            let cmsg_space = space(cmsg_len - cmsghdr_len);
+            self.control = if cmsg_space > control.len() {
+                // No more entries---not enough data in `control` for a
+                // complete message.
+                Bytes::new()
+            } else {
+                control.slice_from(cmsg_space)
+            };
+
+            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 _),
+                    });
+                }
+                (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,
+}
+
+#[must_use]
+pub struct ControlMsgBuilder {
+    result: Result<BytesMut, Error>,
+}
+
+pub fn builder(buf: &mut BytesMut) -> ControlMsgBuilder {
+    let buf = aligned(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(|mut cmsg| {
+            let cmsg_space = space(msg.len());
+            if cmsg.remaining_mut() < cmsg_space {
+                return Err(Error::NoSpace);
+            }
+
+            // Some definitions of cmsghdr contain padding.  Rather
+            // than try to keep an up-to-date #cfg list to handle
+            // that, just use a pre-zeroed struct to fill out any
+            // fields we don't care about.
+            let zeroed = unsafe { mem::zeroed() };
+            #[allow(clippy::needless_update)]
+            let cmsghdr = cmsghdr {
+                cmsg_len: len(msg.len()).try_into().unwrap(),
+                cmsg_level: level,
+                cmsg_type: kind,
+                ..zeroed
+            };
+
+            unsafe {
+                let cmsghdr_ptr = cmsg.bytes_mut().as_mut_ptr();
+                std::ptr::copy_nonoverlapping(
+                    &cmsghdr as *const _ as *const _,
+                    cmsghdr_ptr,
+                    mem::size_of::<cmsghdr>(),
+                );
+                let cmsg_data_ptr = libc::CMSG_DATA(cmsghdr_ptr as _);
+                std::ptr::copy_nonoverlapping(msg.as_ptr(), cmsg_data_ptr, msg.len());
+                cmsg.advance_mut(cmsg_space);
+            }
+
+            Ok(cmsg)
+        });
+
+        self
+    }
+
+    pub fn rights(self, fds: &[RawFd]) -> Self {
+        self.msg(libc::SOL_SOCKET, libc::SCM_RIGHTS, fds.as_bytes())
+    }
+
+    pub fn finish(self) -> Result<Bytes, Error> {
+        self.result.map(|mut cmsg| cmsg.take().freeze())
+    }
+}
+
+trait AsBytes {
+    fn as_bytes(&self) -> &[u8];
+}
+
+impl<'a, T: Sized> AsBytes for &'a [T] {
+    fn as_bytes(&self) -> &[u8] {
+        // TODO: This should account for the alignment of T
+        let byte_count = self.len() * mem::size_of::<T>();
+        unsafe { slice::from_raw_parts(self.as_ptr() as *const _, byte_count) }
+    }
+}
+
+fn aligned(buf: &BytesMut) -> BytesMut {
+    let mut aligned_buf = buf.clone();
+    aligned_buf.reserve(buf.remaining_mut());
+    let cmsghdr_align = mem::align_of::<cmsghdr>();
+    let n = unsafe { aligned_buf.bytes_mut().as_ptr() } as usize & (cmsghdr_align - 1);
+    if n != 0 {
+        unsafe { aligned_buf.advance_mut(n) };
+        drop(aligned_buf.take());
+    }
+    aligned_buf
+}
+
+fn len(len: usize) -> usize {
+    unsafe { libc::CMSG_LEN(len.try_into().unwrap()) as usize }
+}
+
+pub fn space(len: usize) -> usize {
+    unsafe { libc::CMSG_SPACE(len.try_into().unwrap()) as usize }
+}
diff --git a/media/audioipc/audioipc/src/codec.rs b/media/audioipc/audioipc/src/codec.rs
new file mode 100644
index 0000000000..2a6d15de0e
--- /dev/null
+++ b/media/audioipc/audioipc/src/codec.rs
@@ -0,0 +1,185 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+//! `Encoder`s and `Decoder`s from items to/from `BytesMut` buffers.
+
+use bincode::{self, deserialize, serialized_size};
+use bytes::{BufMut, ByteOrder, BytesMut, LittleEndian};
+use serde::de::DeserializeOwned;
+use serde::ser::Serialize;
+use std::fmt::Debug;
+use std::io;
+use std::marker::PhantomData;
+
+////////////////////////////////////////////////////////////////////////////////
+// Split buffer into size delimited frames - This appears more complicated than
+// might be necessary due to handling the possibility of messages being split
+// across reads.
+
+pub trait Codec {
+    /// The type of items to be encoded into byte buffer
+    type In;
+
+    /// The type of items to be returned by decoding from byte buffer
+    type Out;
+
+    /// Attempts to decode a frame from the provided buffer of bytes.
+    fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Self::Out>>;
+
+    /// 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 self.decode(buf)? {
+            Some(frame) => Ok(frame),
+            None => Err(io::Error::new(
+                io::ErrorKind::Other,
+                "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>,
+}
+
+enum State {
+    Length,
+    Data(usize),
+}
+
+const MAX_MESSAGE_LEN: u64 = 1024 * 1024;
+const MESSAGE_LENGTH_SIZE: usize = std::mem::size_of::<u32>();
+// TODO: static assert that MAX_MESSAGE_LEN can be encoded into MESSAGE_LENGTH_SIZE.
+
+impl<In, Out> Default for LengthDelimitedCodec<In, Out> {
+    fn default() -> Self {
+        LengthDelimitedCodec {
+            state: State::Length,
+            __in: PhantomData,
+            __out: PhantomData,
+        }
+    }
+}
+
+impl<In, Out> LengthDelimitedCodec<In, Out> {
+    // Lengths are encoded as little endian u32
+    fn decode_length(&mut self, buf: &mut BytesMut) -> io::Result<Option<usize>> {
+        if buf.len() < MESSAGE_LENGTH_SIZE {
+            // Not enough data
+            return Ok(None);
+        }
+
+        let n = LittleEndian::read_u32(buf.as_ref());
+
+        // Consume the length field
+        let _ = buf.split_to(MESSAGE_LENGTH_SIZE);
+
+        Ok(Some(n as usize))
+    }
+
+    fn decode_data(&mut self, buf: &mut BytesMut, n: usize) -> io::Result<Option<Out>>
+    where
+        Out: DeserializeOwned + Debug,
+    {
+        // At this point, the buffer has already had the required capacity
+        // reserved. All there is to do is read.
+        if buf.len() < n {
+            return Ok(None);
+        }
+
+        let buf = buf.split_to(n).freeze();
+
+        trace!("Attempting to decode");
+        let msg = deserialize::<Out>(buf.as_ref()).map_err(|e| match *e {
+            bincode::ErrorKind::Io(e) => 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,
+{
+    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 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);
+
+                        n
+                    }
+                    None => return Ok(None),
+                }
+            }
+            State::Data(n) => n,
+        };
+
+        match 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(MESSAGE_LENGTH_SIZE);
+
+                Ok(Some(data))
+            }
+            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).unwrap();
+        if encoded_len > MAX_MESSAGE_LEN {
+            trace!("oversized message {}", encoded_len);
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "encoded message too big",
+            ));
+        }
+
+        buf.reserve((encoded_len as usize) + MESSAGE_LENGTH_SIZE);
+
+        buf.put_u32_le(encoded_len as u32);
+
+        if let Err(e) = bincode::config()
+            .limit(encoded_len)
+            .serialize_into::<_, Self::In>(&mut buf.writer(), &item)
+        {
+            match *e {
+                bincode::ErrorKind::Io(e) => return Err(e),
+                _ => return Err(io::Error::new(io::ErrorKind::Other, *e)),
+            }
+        }
+
+        Ok(())
+    }
+}
diff --git a/media/audioipc/audioipc/src/core.rs b/media/audioipc/audioipc/src/core.rs
new file mode 100644
index 0000000000..0edb02b6e0
--- /dev/null
+++ b/media/audioipc/audioipc/src/core.rs
@@ -0,0 +1,85 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+
+// Ease accessing reactor::Core handles.
+
+use futures::sync::oneshot;
+use std::sync::mpsc;
+use std::{fmt, io, thread};
+use tokio::runtime::current_thread;
+
+struct Inner {
+    join: thread::JoinHandle<()>,
+    shutdown: oneshot::Sender<()>,
+}
+
+pub struct CoreThread {
+    inner: Option<Inner>,
+    handle: current_thread::Handle,
+}
+
+impl CoreThread {
+    pub fn handle(&self) -> current_thread::Handle {
+        self.handle.clone()
+    }
+}
+
+impl Drop for CoreThread {
+    fn drop(&mut self) {
+        debug!("Shutting down {:?}", self);
+        if let Some(inner) = self.inner.take() {
+            let _ = inner.shutdown.send(());
+            drop(inner.join.join());
+        }
+    }
+}
+
+impl fmt::Debug for CoreThread {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // f.debug_tuple("CoreThread").field(&"...").finish()
+        f.debug_tuple("CoreThread").field(&self.handle).finish()
+    }
+}
+
+pub fn spawn_thread<S, F, D>(name: S, f: F, d: D) -> io::Result<CoreThread>
+where
+    S: Into<String>,
+    F: FnOnce() -> io::Result<()> + Send + 'static,
+    D: FnOnce() -> () + Send + 'static,
+{
+    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+    let (handle_tx, handle_rx) = mpsc::channel::<current_thread::Handle>();
+
+    let join = thread::Builder::new().name(name.into()).spawn(move || {
+        let mut rt =
+            current_thread::Runtime::new().expect("Failed to create current_thread::Runtime");
+        let handle = rt.handle();
+        drop(handle_tx.send(handle.clone()));
+
+        rt.spawn(futures::future::lazy(|| {
+            let _ = f();
+            Ok(())
+        }));
+
+        let _ = rt.block_on(shutdown_rx);
+        trace!("thread shutdown...");
+        d();
+    })?;
+
+    let handle = handle_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,
+            shutdown: shutdown_tx,
+        }),
+        handle,
+    })
+}
diff --git a/media/audioipc/audioipc/src/errors.rs b/media/audioipc/audioipc/src/errors.rs
new file mode 100644
index 0000000000..5cc37503f9
--- /dev/null
+++ b/media/audioipc/audioipc/src/errors.rs
@@ -0,0 +1,22 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+
+use bincode;
+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::Error);
+    }
+
+    // Replace bail!(str) with explicit errors.
+    errors {
+        Disconnected
+    }
+}
diff --git a/media/audioipc/audioipc/src/fd_passing.rs b/media/audioipc/audioipc/src/fd_passing.rs
new file mode 100644
index 0000000000..411dcfd803
--- /dev/null
+++ b/media/audioipc/audioipc/src/fd_passing.rs
@@ -0,0 +1,363 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::async_msg::{AsyncRecvMsg, AsyncSendMsg};
+use crate::cmsg;
+use crate::codec::Codec;
+use crate::messages::AssocRawPlatformHandle;
+use bytes::{Bytes, BytesMut, IntoBuf};
+use futures::{task, AsyncSink, Poll, Sink, StartSend, Stream};
+use std::collections::VecDeque;
+use std::os::unix::io::RawFd;
+use std::{fmt, io, mem};
+
+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>,
+}
+
+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,
+        }
+    }
+
+    pub fn take_fds(&mut self) -> Option<[RawFd; 3]> {
+        loop {
+            let fds = self
+                .recv_fds
+                .as_mut()
+                .and_then(|recv_fds| recv_fds.next())
+                .and_then(|fds| Some(clone_into_array(&fds)));
+
+            if fds.is_some() {
+                return fds;
+            }
+
+            if self.cmsg.is_empty() {
+                return None;
+            }
+
+            self.recv_fds = Some(cmsg::iterator(self.cmsg.take().freeze()));
+        }
+    }
+
+    pub fn cmsg(&mut self) -> &mut BytesMut {
+        self.cmsg.reserve(cmsg::space(mem::size_of::<[RawFd; 3]>()));
+        &mut self.cmsg
+    }
+}
+
+#[derive(Debug)]
+struct Frame {
+    msgs: 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 FramedWithPlatformHandles<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,
+}
+
+impl<A, C> FramedWithPlatformHandles<A, C>
+where
+    A: AsyncSendMsg,
+{
+    // If there is a buffered frame, try to write it to `A`
+    fn do_write(&mut self) -> Poll<(), io::Error> {
+        trace!("do_write...");
+        // Create a frame from any pending message in `write_buf`.
+        if !self.write_buf.is_empty() {
+            self.set_frame(None);
+        }
+
+        trace!("pending frames: {:?}", self.frames);
+
+        let mut processed = 0;
+
+        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 fds = match frame.fds {
+                        Some(ref fds) => fds.clone(),
+                        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) => {
+                    processed += 1;
+
+                    // Close any fds that have been sent. The fds are
+                    // encoded in cmsg format inside frame.fds. Use
+                    // the cmsg iterator to access msg and extract
+                    // RawFds.
+                    frame.fds.take().and_then(|cmsg| {
+                        for fds in cmsg::iterator(cmsg) {
+                            close_fds(&*fds)
+                        }
+                        Some(())
+                    });
+
+                    if n != frame.msgs.len() {
+                        // If only part of the message was sent then
+                        // 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!(),
+            }
+        }
+        trace!("process {} frames", processed);
+        trace!("pending frames: {:?}", self.frames);
+
+        Ok(().into())
+    }
+
+    fn set_frame(&mut self, fds: Option<Bytes>) {
+        if self.write_buf.is_empty() {
+            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 });
+    }
+}
+
+impl<A, C> Stream for FramedWithPlatformHandles<A, C>
+where
+    A: AsyncRecvMsg,
+    C: Codec,
+    C::Out: AssocRawPlatformHandle,
+{
+    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
+            // the upstream has returned EOF, and the decoder is no longer
+            // readable, it can be assumed that the decoder will never become
+            // readable again, at which point the stream is terminated.
+            if self.is_readable {
+                if self.eof {
+                    let mut item = self.codec.decode_eof(&mut self.read_buf)?;
+                    item.take_platform_handles(|| self.incoming_fds.take_fds());
+                    return Ok(Some(item).into());
+                }
+
+                trace!("attempting to decode a frame");
+
+                if let Some(mut item) = self.codec.decode(&mut self.read_buf)? {
+                    trace!("frame decoded from buffer");
+                    item.take_platform_handles(|| self.incoming_fds.take_fds());
+                    return Ok(Some(item).into());
+                }
+
+                self.is_readable = false;
+            }
+
+            assert!(!self.eof);
+
+            // Otherwise, try to read more data and try again. Make sure we've
+            // got room for at least one byte to read to ensure that we don't
+            // get a spurious 0 that looks like EOF
+            let (n, _) = try_ready!(self
+                .io
+                .recv_msg_buf(&mut self.read_buf, self.incoming_fds.cmsg()));
+
+            if n == 0 {
+                self.eof = true;
+            }
+
+            self.is_readable = true;
+        }
+    }
+}
+
+impl<A, C> Sink for FramedWithPlatformHandles<A, C>
+where
+    A: AsyncSendMsg,
+    C: Codec,
+    C::In: AssocRawPlatformHandle + fmt::Debug,
+{
+    type SinkItem = C::In;
+    type SinkError = io::Error;
+
+    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 {
+            self.poll_complete()?;
+            if self.write_buf.len() > BACKPRESSURE_THRESHOLD {
+                return Ok(AsyncSink::NotReady(item));
+            }
+        }
+
+        let fds = item.platform_handles();
+        self.codec.encode(item, &mut self.write_buf)?;
+        let fds = fds.and_then(|fds| {
+            cmsg::builder(&mut self.outgoing_fds)
+                .rights(&fds.0[..])
+                .finish()
+                .ok()
+        });
+
+        trace!("item fds: {:?}", fds);
+
+        if fds.is_some() {
+            // Enforce splitting sends on messages that contain file
+            // descriptors.
+            self.set_frame(fds);
+        }
+
+        Ok(AsyncSink::Ready)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        trace!("flushing framed transport");
+
+        try_ready!(self.do_write());
+
+        try_nb!(self.io.flush());
+
+        trace!("framed transport flushed");
+        Ok(().into())
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        if task::is_in_task() {
+            try_ready!(self.poll_complete());
+        }
+        self.io.shutdown()
+    }
+}
+
+pub fn framed_with_platformhandles<A, C>(io: A, codec: C) -> FramedWithPlatformHandles<A, C> {
+    FramedWithPlatformHandles {
+        io,
+        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]>()),
+        ),
+    }
+}
+
+fn clone_into_array<A, T>(slice: &[T]) -> A
+where
+    A: Sized + Default + AsMut<[T]>,
+    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 {
+        unsafe {
+            super::close_platformhandle(*fd);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use bytes::BufMut;
+    use libc;
+    use std;
+
+    extern "C" {
+        fn cmsghdr_bytes(size: *mut libc::size_t) -> *const u8;
+    }
+
+    fn cmsg_bytes() -> &'static [u8] {
+        let mut size = 0;
+        unsafe {
+            let ptr = cmsghdr_bytes(&mut size);
+            std::slice::from_raw_parts(ptr, size)
+        }
+    }
+
+    #[test]
+    fn single_cmsg() {
+        let mut incoming = super::IncomingFds::new(16);
+
+        incoming.cmsg().put_slice(cmsg_bytes());
+        assert!(incoming.take_fds().is_some());
+        assert!(incoming.take_fds().is_none());
+    }
+
+    #[test]
+    fn multiple_cmsg_1() {
+        let mut incoming = super::IncomingFds::new(16);
+
+        incoming.cmsg().put_slice(cmsg_bytes());
+        assert!(incoming.take_fds().is_some());
+        incoming.cmsg().put_slice(cmsg_bytes());
+        assert!(incoming.take_fds().is_some());
+        assert!(incoming.take_fds().is_none());
+    }
+
+    #[test]
+    fn multiple_cmsg_2() {
+        let mut incoming = super::IncomingFds::new(16);
+        println!("cmsg_bytes() {}", cmsg_bytes().len());
+
+        incoming.cmsg().put_slice(cmsg_bytes());
+        incoming.cmsg().put_slice(cmsg_bytes());
+        assert!(incoming.take_fds().is_some());
+        incoming.cmsg().put_slice(cmsg_bytes());
+        assert!(incoming.take_fds().is_some());
+        assert!(incoming.take_fds().is_some());
+        assert!(incoming.take_fds().is_none());
+    }
+}
diff --git a/media/audioipc/audioipc/src/frame.rs b/media/audioipc/audioipc/src/frame.rs
new file mode 100644
index 0000000000..fa5b0bd4bd
--- /dev/null
+++ b/media/audioipc/audioipc/src/frame.rs
@@ -0,0 +1,165 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::codec::Codec;
+use bytes::{Buf, Bytes, BytesMut, IntoBuf};
+use futures::{task, AsyncSink, Poll, Sink, StartSend, Stream};
+use std::io;
+use tokio_io::{AsyncRead, AsyncWrite};
+
+const INITIAL_CAPACITY: usize = 1024;
+const BACKPRESSURE_THRESHOLD: usize = 4 * INITIAL_CAPACITY;
+
+/// A unified `Stream` and `Sink` interface over an I/O object, using
+/// 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,
+}
+
+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> {
+        loop {
+            if self.frame.is_none() {
+                self.set_frame();
+            }
+
+            if self.frame.is_none() {
+                return Ok(().into());
+            }
+
+            let done = {
+                let frame = self.frame.as_mut().unwrap();
+                try_ready!(self.io.write_buf(frame));
+                !frame.has_remaining()
+            };
+
+            if done {
+                self.frame = None;
+            }
+        }
+    }
+
+    fn set_frame(&mut self) {
+        if self.write_buf.is_empty() {
+            return;
+        }
+
+        debug_assert!(self.frame.is_none());
+
+        self.frame = Some(self.write_buf.take().freeze().into_buf());
+    }
+}
+
+impl<A, C> Stream for Framed<A, C>
+where
+    A: AsyncRead,
+    C: Codec,
+{
+    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
+            // the upstream has returned EOF, and the decoder is no longer
+            // readable, it can be assumed that the decoder will never become
+            // readable again, at which point the stream is terminated.
+            if self.is_readable {
+                if self.eof {
+                    let frame = self.codec.decode_eof(&mut self.read_buf)?;
+                    return Ok(Some(frame).into());
+                }
+
+                trace!("attempting to decode a frame");
+
+                if let Some(frame) = self.codec.decode(&mut self.read_buf)? {
+                    trace!("frame decoded from buffer");
+                    return Ok(Some(frame).into());
+                }
+
+                self.is_readable = false;
+            }
+
+            assert!(!self.eof);
+
+            // XXX(kinetik): work around tokio_named_pipes assuming at least 1kB available.
+            self.read_buf.reserve(INITIAL_CAPACITY);
+
+            // Otherwise, try to read more data and try again. Make sure we've
+            // got room for at least one byte to read to ensure that we don't
+            // get a spurious 0 that looks like EOF
+            if try_ready!(self.io.read_buf(&mut self.read_buf)) == 0 {
+                self.eof = true;
+            }
+
+            self.is_readable = true;
+        }
+    }
+}
+
+impl<A, C> Sink for Framed<A, C>
+where
+    A: AsyncWrite,
+    C: Codec,
+{
+    type SinkItem = C::In;
+    type SinkError = io::Error;
+
+    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
+        // 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 {
+            self.poll_complete()?;
+            if self.write_buf.len() > BACKPRESSURE_THRESHOLD {
+                return Ok(AsyncSink::NotReady(item));
+            }
+        }
+
+        self.codec.encode(item, &mut self.write_buf)?;
+        Ok(AsyncSink::Ready)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        trace!("flushing framed transport");
+
+        try_ready!(self.do_write());
+
+        try_nb!(self.io.flush());
+
+        trace!("framed transport flushed");
+        Ok(().into())
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        if task::is_in_task() {
+            try_ready!(self.poll_complete());
+        }
+        self.io.shutdown()
+    }
+}
+
+pub fn framed<A, C>(io: A, codec: C) -> Framed<A, C> {
+    Framed {
+        io,
+        codec,
+        read_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
+        write_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
+        frame: None,
+        is_readable: false,
+        eof: false,
+    }
+}
diff --git a/media/audioipc/audioipc/src/handle_passing.rs b/media/audioipc/audioipc/src/handle_passing.rs
new file mode 100644
index 0000000000..347613e5f5
--- /dev/null
+++ b/media/audioipc/audioipc/src/handle_passing.rs
@@ -0,0 +1,278 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::codec::Codec;
+use crate::messages::AssocRawPlatformHandle;
+use bytes::{Bytes, BytesMut, IntoBuf};
+use futures::{task, AsyncSink, Poll, Sink, StartSend, Stream};
+use std::collections::VecDeque;
+use std::{fmt, io};
+use tokio_io::{AsyncRead, AsyncWrite};
+
+const INITIAL_CAPACITY: usize = 1024;
+const BACKPRESSURE_THRESHOLD: usize = 4 * INITIAL_CAPACITY;
+
+#[derive(Debug)]
+struct Frame {
+    msgs: Bytes,
+}
+
+/// A unified `Stream` and `Sink` interface over an I/O object, using
+/// the `Codec` trait to encode and decode the payload.
+pub struct FramedWithPlatformHandles<A, C> {
+    io: A,
+    codec: C,
+    // Stream
+    read_buf: BytesMut,
+    is_readable: bool,
+    eof: bool,
+    // Sink
+    frames: VecDeque<Frame>,
+    write_buf: BytesMut,
+}
+
+impl<A, C> FramedWithPlatformHandles<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> {
+        trace!("do_write...");
+        // Create a frame from any pending message in `write_buf`.
+        if !self.write_buf.is_empty() {
+            self.set_frame();
+        }
+
+        trace!("pending frames: {:?}", self.frames);
+
+        let mut processed = 0;
+
+        loop {
+            let n = match self.frames.front() {
+                Some(frame) => {
+                    trace!("sending msg {:?}", frame.msgs);
+                    let mut msgs = frame.msgs.clone().into_buf();
+                    try_ready!(self.io.write_buf(&mut msgs))
+                }
+                _ => {
+                    // No pending frames.
+                    return Ok(().into());
+                }
+            };
+
+            match self.frames.pop_front() {
+                Some(mut frame) => {
+                    processed += 1;
+
+                    if n != frame.msgs.len() {
+                        // If only part of the message was sent then
+                        // re-queue the remaining message at the head
+                        // of the queue. (Don't need to resend the
+                        // handles since they've been sent with the
+                        // first part.)
+                        drop(frame.msgs.split_to(n));
+                        self.frames.push_front(frame);
+                        break;
+                    }
+                }
+                _ => panic!(),
+            }
+        }
+        trace!("process {} frames", processed);
+        trace!("pending frames: {:?}", self.frames);
+
+        Ok(().into())
+    }
+
+    fn set_frame(&mut self) {
+        if self.write_buf.is_empty() {
+            trace!("set_frame: No pending messages...");
+            return;
+        }
+
+        let msgs = self.write_buf.take().freeze();
+        trace!("set_frame: msgs={:?}", msgs);
+
+        self.frames.push_back(Frame { msgs });
+    }
+}
+
+impl<A, C> Stream for FramedWithPlatformHandles<A, C>
+where
+    A: AsyncRead,
+    C: Codec,
+    C::Out: AssocRawPlatformHandle,
+{
+    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
+            // the upstream has returned EOF, and the decoder is no longer
+            // readable, it can be assumed that the decoder will never become
+            // readable again, at which point the stream is terminated.
+            if self.is_readable {
+                if self.eof {
+                    let item = self.codec.decode_eof(&mut self.read_buf)?;
+                    return Ok(Some(item).into());
+                }
+
+                trace!("attempting to decode a frame");
+
+                if let Some(item) = self.codec.decode(&mut self.read_buf)? {
+                    trace!("frame decoded from buffer");
+                    return Ok(Some(item).into());
+                }
+
+                self.is_readable = false;
+            }
+
+            assert!(!self.eof);
+
+            // XXX(kinetik): work around tokio_named_pipes assuming at least 1kB available.
+            self.read_buf.reserve(INITIAL_CAPACITY);
+
+            // Otherwise, try to read more data and try again. Make sure we've
+            // got room for at least one byte to read to ensure that we don't
+            // get a spurious 0 that looks like EOF
+            let n = try_ready!(self.io.read_buf(&mut self.read_buf));
+
+            if n == 0 {
+                self.eof = true;
+            }
+
+            self.is_readable = true;
+        }
+    }
+}
+
+impl<A, C> Sink for FramedWithPlatformHandles<A, C>
+where
+    A: AsyncWrite,
+    C: Codec,
+    C::In: AssocRawPlatformHandle + fmt::Debug,
+{
+    type SinkItem = C::In;
+    type SinkError = io::Error;
+
+    fn start_send(
+        &mut self,
+        mut 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 {
+            self.poll_complete()?;
+            if self.write_buf.len() > BACKPRESSURE_THRESHOLD {
+                return Ok(AsyncSink::NotReady(item));
+            }
+        }
+
+        let mut got_handles = false;
+        if let Some((handles, target_pid)) = item.platform_handles() {
+            got_handles = true;
+            let remote_handles = unsafe {
+                // Attempt to duplicate all 3 handles before checking
+                // result, since we rely on duplicate_platformhandle closing
+                // our source handles.
+                let r1 = duplicate_platformhandle(handles[0], target_pid);
+                let r2 = duplicate_platformhandle(handles[1], target_pid);
+                let r3 = duplicate_platformhandle(handles[2], target_pid);
+                [r1?, r2?, r3?]
+            };
+            trace!(
+                "item handles: {:?} remote_handles: {:?}",
+                handles,
+                remote_handles
+            );
+            item.take_platform_handles(|| Some(remote_handles));
+        }
+
+        self.codec.encode(item, &mut self.write_buf)?;
+
+        if got_handles {
+            // Enforce splitting sends on messages that contain file
+            // descriptors.
+            self.set_frame();
+        }
+
+        Ok(AsyncSink::Ready)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        trace!("flushing framed transport");
+
+        try_ready!(self.do_write());
+
+        try_nb!(self.io.flush());
+
+        trace!("framed transport flushed");
+        Ok(().into())
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        if task::is_in_task() {
+            try_ready!(self.poll_complete());
+        }
+        self.io.shutdown()
+    }
+}
+
+pub fn framed_with_platformhandles<A, C>(io: A, codec: C) -> FramedWithPlatformHandles<A, C> {
+    FramedWithPlatformHandles {
+        io,
+        codec,
+        read_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
+        is_readable: false,
+        eof: false,
+        frames: VecDeque::new(),
+        write_buf: BytesMut::with_capacity(INITIAL_CAPACITY),
+    }
+}
+
+use super::PlatformHandleType;
+use winapi::shared::minwindef::{DWORD, FALSE};
+use winapi::um::{handleapi, processthreadsapi, winnt};
+
+// source_handle is effectively taken ownership of (consumed) and
+// closed when duplicate_platformhandle is called.
+// TODO: Make this transfer more explicit via the type system.
+unsafe fn duplicate_platformhandle(
+    source_handle: PlatformHandleType,
+    target_pid: DWORD,
+) -> Result<PlatformHandleType, std::io::Error> {
+    let source = processthreadsapi::GetCurrentProcess();
+    let target = processthreadsapi::OpenProcess(winnt::PROCESS_DUP_HANDLE, FALSE, target_pid);
+    if !super::valid_handle(target) {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "invalid target process",
+        ));
+    }
+
+    let mut target_handle = std::ptr::null_mut();
+    let ok = handleapi::DuplicateHandle(
+        source,
+        source_handle,
+        target,
+        &mut target_handle,
+        0,
+        FALSE,
+        winnt::DUPLICATE_CLOSE_SOURCE | winnt::DUPLICATE_SAME_ACCESS,
+    );
+    handleapi::CloseHandle(target);
+    if ok == FALSE {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "DuplicateHandle failed",
+        ));
+    }
+    Ok(target_handle)
+}
diff --git a/media/audioipc/audioipc/src/lib.rs b/media/audioipc/audioipc/src/lib.rs
new file mode 100644
index 0000000000..1c056bc123
--- /dev/null
+++ b/media/audioipc/audioipc/src/lib.rs
@@ -0,0 +1,212 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+#![warn(unused_extern_crates)]
+#![recursion_limit = "1024"]
+#[macro_use]
+extern crate error_chain;
+#[macro_use]
+extern crate log;
+#[macro_use]
+extern crate serde_derive;
+#[macro_use]
+extern crate futures;
+#[macro_use]
+extern crate tokio_io;
+
+mod async_msg;
+#[cfg(unix)]
+mod cmsg;
+pub mod codec;
+pub mod core;
+#[allow(deprecated)]
+pub mod errors;
+#[cfg(unix)]
+pub mod fd_passing;
+#[cfg(unix)]
+pub use crate::fd_passing as platformhandle_passing;
+#[cfg(windows)]
+pub mod handle_passing;
+#[cfg(windows)]
+pub use handle_passing as platformhandle_passing;
+pub mod frame;
+pub mod messages;
+#[cfg(unix)]
+mod msg;
+pub mod rpc;
+pub mod shm;
+
+// TODO: Remove local fork when https://github.com/tokio-rs/tokio/pull/1294 is resolved.
+#[cfg(unix)]
+mod tokio_uds_stream;
+
+#[cfg(windows)]
+mod tokio_named_pipes;
+
+pub use crate::messages::{ClientMessage, ServerMessage};
+
+// TODO: Remove hardcoded size and allow allocation based on cubeb backend requirements.
+pub const SHM_AREA_SIZE: usize = 2 * 1024 * 1024;
+
+#[cfg(unix)]
+use std::os::unix::io::{FromRawFd, IntoRawFd};
+#[cfg(windows)]
+use std::os::windows::io::{FromRawHandle, IntoRawHandle};
+
+use std::cell::RefCell;
+
+// This must match the definition of
+// ipc::FileDescriptor::PlatformHandleType in Gecko.
+#[cfg(windows)]
+pub type PlatformHandleType = std::os::windows::raw::HANDLE;
+#[cfg(unix)]
+pub type PlatformHandleType = libc::c_int;
+
+// This stands in for RawFd/RawHandle.
+#[derive(Clone, Debug)]
+pub struct PlatformHandle(RefCell<Inner>);
+
+#[derive(Clone, Debug)]
+struct Inner {
+    handle: PlatformHandleType,
+    owned: bool,
+}
+
+unsafe impl Send for PlatformHandle {}
+
+pub const INVALID_HANDLE_VALUE: PlatformHandleType = -1isize as PlatformHandleType;
+
+// Custom serialization to treat HANDLEs as i64.  This is not valid in
+// general, but after sending the HANDLE value to a remote process we
+// use it to create a valid HANDLE via DuplicateHandle.
+// To avoid duplicating the serialization code, we're lazy and treat
+// file descriptors as i64 rather than i32.
+impl serde::Serialize for PlatformHandle {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let h = self.0.borrow();
+        serializer.serialize_i64(h.handle as i64)
+    }
+}
+
+struct PlatformHandleVisitor;
+impl<'de> serde::de::Visitor<'de> for PlatformHandleVisitor {
+    type Value = PlatformHandle;
+
+    fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str("an integer between -2^63 and 2^63")
+    }
+
+    fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E>
+    where
+        E: serde::de::Error,
+    {
+        let owned = cfg!(windows);
+        Ok(PlatformHandle::new(value as PlatformHandleType, owned))
+    }
+}
+
+impl<'de> serde::Deserialize<'de> for PlatformHandle {
+    fn deserialize<D>(deserializer: D) -> Result<PlatformHandle, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        deserializer.deserialize_i64(PlatformHandleVisitor)
+    }
+}
+
+#[cfg(unix)]
+fn valid_handle(handle: PlatformHandleType) -> bool {
+    handle >= 0
+}
+
+#[cfg(windows)]
+fn valid_handle(handle: PlatformHandleType) -> bool {
+    const NULL_HANDLE_VALUE: PlatformHandleType = 0isize as PlatformHandleType;
+    handle != INVALID_HANDLE_VALUE && handle != NULL_HANDLE_VALUE
+}
+
+impl PlatformHandle {
+    pub fn new(raw: PlatformHandleType, owned: bool) -> PlatformHandle {
+        assert!(valid_handle(raw));
+        let inner = Inner {
+            handle: raw,
+            owned: owned,
+        };
+        PlatformHandle(RefCell::new(inner))
+    }
+
+    #[cfg(windows)]
+    pub fn from<T: IntoRawHandle>(from: T) -> PlatformHandle {
+        PlatformHandle::new(from.into_raw_handle(), true)
+    }
+
+    #[cfg(unix)]
+    pub fn from<T: IntoRawFd>(from: T) -> PlatformHandle {
+        PlatformHandle::new(from.into_raw_fd(), true)
+    }
+
+    #[cfg(windows)]
+    pub unsafe fn into_file(&self) -> std::fs::File {
+        std::fs::File::from_raw_handle(self.into_raw())
+    }
+
+    #[cfg(unix)]
+    pub unsafe fn into_file(&self) -> std::fs::File {
+        std::fs::File::from_raw_fd(self.into_raw())
+    }
+
+    pub unsafe fn into_raw(&self) -> PlatformHandleType {
+        let mut h = self.0.borrow_mut();
+        assert!(h.owned);
+        h.owned = false;
+        h.handle
+    }
+}
+
+impl Drop for PlatformHandle {
+    fn drop(&mut self) {
+        let inner = self.0.borrow();
+        if inner.owned {
+            unsafe { close_platformhandle(inner.handle) }
+        }
+    }
+}
+
+#[cfg(unix)]
+unsafe fn close_platformhandle(handle: PlatformHandleType) {
+    libc::close(handle);
+}
+
+#[cfg(windows)]
+unsafe fn close_platformhandle(handle: PlatformHandleType) {
+    winapi::um::handleapi::CloseHandle(handle);
+}
+
+#[cfg(unix)]
+pub mod messagestream_unix;
+#[cfg(unix)]
+pub use crate::messagestream_unix::*;
+
+#[cfg(windows)]
+pub mod messagestream_win;
+#[cfg(windows)]
+pub use messagestream_win::*;
+
+#[cfg(windows)]
+pub fn server_platform_init() {
+    use winapi::shared::winerror;
+    use winapi::um::combaseapi;
+    use winapi::um::objbase;
+
+    unsafe {
+        let r = combaseapi::CoInitializeEx(std::ptr::null_mut(), objbase::COINIT_MULTITHREADED);
+        assert!(winerror::SUCCEEDED(r));
+    }
+}
+
+#[cfg(unix)]
+pub fn server_platform_init() {}
diff --git a/media/audioipc/audioipc/src/messages.rs b/media/audioipc/audioipc/src/messages.rs
new file mode 100644
index 0000000000..94843e629c
--- /dev/null
+++ b/media/audioipc/audioipc/src/messages.rs
@@ -0,0 +1,388 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::PlatformHandle;
+use crate::PlatformHandleType;
+#[cfg(target_os = "linux")]
+use audio_thread_priority::RtPriorityThreadInfo;
+use cubeb::{self, ffi};
+use std::ffi::{CStr, CString};
+use std::os::raw::{c_char, c_int, c_uint};
+use std::ptr;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Device {
+    pub output_name: Option<Vec<u8>>,
+    pub input_name: Option<Vec<u8>>,
+}
+
+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()),
+        }
+    }
+}
+
+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),
+        }
+    }
+}
+
+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),
+        }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct DeviceInfo {
+    pub devid: usize,
+    pub device_id: Option<Vec<u8>>,
+    pub friendly_name: Option<Vec<u8>>,
+    pub group_id: Option<Vec<u8>>,
+    pub vendor_name: Option<Vec<u8>>,
+
+    pub device_type: ffi::cubeb_device_type,
+    pub state: ffi::cubeb_device_state,
+    pub preferred: ffi::cubeb_device_pref,
+
+    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,
+}
+
+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,
+            state: info.state,
+            preferred: info.preferred,
+
+            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,
+        }
+    }
+}
+
+impl From<DeviceInfo> for ffi::cubeb_device_info {
+    fn from(info: DeviceInfo) -> Self {
+        ffi::cubeb_device_info {
+            devid: info.devid as _,
+            device_id: opt_str(info.device_id),
+            friendly_name: opt_str(info.friendly_name),
+            group_id: opt_str(info.group_id),
+            vendor_name: opt_str(info.vendor_name),
+
+            device_type: info.device_type,
+            state: info.state,
+            preferred: info.preferred,
+
+            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,
+        }
+    }
+}
+
+#[repr(C)]
+#[derive(Clone, Copy, Debug, Deserialize, Serialize)]
+pub struct StreamParams {
+    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 cubeb::StreamParamsRef> for StreamParams {
+    fn from(x: &cubeb::StreamParamsRef) -> StreamParams {
+        unsafe { *(x.as_ptr() as *mut StreamParams) }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct StreamCreateParams {
+    pub input_stream_params: Option<StreamParams>,
+    pub output_stream_params: Option<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,
+}
+
+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>>) -> *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_mut()
+            }
+        },
+        None => ptr::null_mut(),
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct StreamCreate {
+    pub token: usize,
+    pub platform_handles: [PlatformHandle; 3],
+    pub target_pid: u32,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RegisterDeviceCollectionChanged {
+    pub platform_handles: [PlatformHandle; 3],
+    pub target_pid: u32,
+}
+
+// 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(u32),
+    ClientDisconnect,
+
+    ContextGetBackendId,
+    ContextGetMaxChannelCount,
+    ContextGetMinLatency(StreamParams),
+    ContextGetPreferredSampleRate,
+    ContextGetDeviceEnumeration(ffi::cubeb_device_type),
+    ContextSetupDeviceCollectionCallback,
+    ContextRegisterDeviceCollectionChanged(ffi::cubeb_device_type, bool),
+
+    StreamCreate(StreamCreateParams),
+    StreamInit(usize, StreamInitParams),
+    StreamDestroy(usize),
+
+    StreamStart(usize),
+    StreamStop(usize),
+    StreamResetDefaultDevice(usize),
+    StreamGetPosition(usize),
+    StreamGetLatency(usize),
+    StreamGetInputLatency(usize),
+    StreamSetVolume(usize, f32),
+    StreamSetName(usize, CString),
+    StreamGetCurrentDevice(usize),
+    StreamRegisterDeviceChangeCallback(usize, bool),
+
+    #[cfg(target_os = "linux")]
+    PromoteThreadToRealTime([u8; std::mem::size_of::<RtPriorityThreadInfo>()]),
+}
+
+// Server -> Client messages.
+// TODO: Streams need id.
+#[derive(Debug, Serialize, Deserialize)]
+pub enum ClientMessage {
+    ClientConnected,
+    ClientDisconnected,
+
+    ContextBackendId(String),
+    ContextMaxChannelCount(u32),
+    ContextMinLatency(u32),
+    ContextPreferredSampleRate(u32),
+    ContextEnumeratedDevices(Vec<DeviceInfo>),
+    ContextSetupDeviceCollectionCallback(RegisterDeviceCollectionChanged),
+    ContextRegisteredDeviceCollectionChanged,
+
+    StreamCreated(StreamCreate),
+    StreamInitialized,
+    StreamDestroyed,
+
+    StreamStarted,
+    StreamStopped,
+    StreamDefaultDeviceReset,
+    StreamPosition(u64),
+    StreamLatency(u32),
+    StreamInputLatency(u32),
+    StreamVolumeSet,
+    StreamNameSet,
+    StreamCurrentDevice(Device),
+    StreamRegisterDeviceChangeCallback,
+
+    #[cfg(target_os = "linux")]
+    ThreadPromoted,
+
+    Error(c_int),
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum CallbackReq {
+    Data {
+        nframes: isize,
+        input_frame_size: usize,
+        output_frame_size: usize,
+    },
+    State(ffi::cubeb_state),
+    DeviceChange,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum CallbackResp {
+    Data(isize),
+    State,
+    DeviceChange,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum DeviceCollectionReq {
+    DeviceChange(ffi::cubeb_device_type),
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum DeviceCollectionResp {
+    DeviceChange,
+}
+
+pub trait AssocRawPlatformHandle {
+    fn platform_handles(&self) -> Option<([PlatformHandleType; 3], u32)> {
+        None
+    }
+
+    fn take_platform_handles<F>(&mut self, f: F)
+    where
+        F: FnOnce() -> Option<[PlatformHandleType; 3]>,
+    {
+        assert!(f().is_none());
+    }
+}
+
+impl AssocRawPlatformHandle for ServerMessage {}
+
+impl AssocRawPlatformHandle for ClientMessage {
+    fn platform_handles(&self) -> Option<([PlatformHandleType; 3], u32)> {
+        unsafe {
+            match *self {
+                ClientMessage::StreamCreated(ref data) => Some((
+                    [
+                        data.platform_handles[0].into_raw(),
+                        data.platform_handles[1].into_raw(),
+                        data.platform_handles[2].into_raw(),
+                    ],
+                    data.target_pid,
+                )),
+                ClientMessage::ContextSetupDeviceCollectionCallback(ref data) => Some((
+                    [
+                        data.platform_handles[0].into_raw(),
+                        data.platform_handles[1].into_raw(),
+                        data.platform_handles[2].into_raw(),
+                    ],
+                    data.target_pid,
+                )),
+                _ => None,
+            }
+        }
+    }
+
+    fn take_platform_handles<F>(&mut self, f: F)
+    where
+        F: FnOnce() -> Option<[PlatformHandleType; 3]>,
+    {
+        let owned = cfg!(unix);
+        match *self {
+            ClientMessage::StreamCreated(ref mut data) => {
+                let handles =
+                    f().expect("platform_handles must be available when processing StreamCreated");
+                data.platform_handles = [
+                    PlatformHandle::new(handles[0], owned),
+                    PlatformHandle::new(handles[1], owned),
+                    PlatformHandle::new(handles[2], owned),
+                ]
+            }
+            ClientMessage::ContextSetupDeviceCollectionCallback(ref mut data) => {
+                let handles = f().expect("platform_handles must be available when processing ContextSetupDeviceCollectionCallback");
+                data.platform_handles = [
+                    PlatformHandle::new(handles[0], owned),
+                    PlatformHandle::new(handles[1], owned),
+                    PlatformHandle::new(handles[2], owned),
+                ]
+            }
+            _ => {}
+        }
+    }
+}
+
+#[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);
+    }
+}
diff --git a/media/audioipc/audioipc/src/messagestream_unix.rs b/media/audioipc/audioipc/src/messagestream_unix.rs
new file mode 100644
index 0000000000..4ce1e96d7b
--- /dev/null
+++ b/media/audioipc/audioipc/src/messagestream_unix.rs
@@ -0,0 +1,105 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use super::tokio_uds_stream as tokio_uds;
+use futures::Poll;
+use mio::Ready;
+use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, RawFd};
+use std::os::unix::net;
+use tokio_io::{AsyncRead, AsyncWrite};
+
+#[derive(Debug)]
+pub struct MessageStream(net::UnixStream);
+pub struct AsyncMessageStream(tokio_uds::UnixStream);
+
+impl MessageStream {
+    fn new(stream: net::UnixStream) -> MessageStream {
+        MessageStream(stream)
+    }
+
+    pub fn anonymous_ipc_pair(
+    ) -> std::result::Result<(MessageStream, MessageStream), std::io::Error> {
+        let pair = net::UnixStream::pair()?;
+        Ok((MessageStream::new(pair.0), MessageStream::new(pair.1)))
+    }
+
+    pub unsafe fn from_raw_fd(raw: super::PlatformHandleType) -> MessageStream {
+        MessageStream::new(net::UnixStream::from_raw_fd(raw))
+    }
+
+    pub fn into_tokio_ipc(
+        self,
+        handle: &tokio::reactor::Handle,
+    ) -> std::result::Result<AsyncMessageStream, std::io::Error> {
+        Ok(AsyncMessageStream::new(tokio_uds::UnixStream::from_std(
+            self.0, handle,
+        )?))
+    }
+}
+
+impl IntoRawFd for MessageStream {
+    fn into_raw_fd(self) -> RawFd {
+        self.0.into_raw_fd()
+    }
+}
+
+impl AsyncMessageStream {
+    fn new(stream: tokio_uds::UnixStream) -> AsyncMessageStream {
+        AsyncMessageStream(stream)
+    }
+
+    pub fn poll_read_ready(&self, ready: Ready) -> Poll<Ready, std::io::Error> {
+        self.0.poll_read_ready(ready)
+    }
+
+    pub fn clear_read_ready(&self, ready: Ready) -> Result<(), std::io::Error> {
+        self.0.clear_read_ready(ready)
+    }
+
+    pub fn poll_write_ready(&self) -> Poll<Ready, std::io::Error> {
+        self.0.poll_write_ready()
+    }
+
+    pub fn clear_write_ready(&self) -> Result<(), std::io::Error> {
+        self.0.clear_write_ready()
+    }
+}
+
+impl std::io::Read for AsyncMessageStream {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        self.0.read(buf)
+    }
+}
+
+impl std::io::Write for AsyncMessageStream {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.0.write(buf)
+    }
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.0.flush()
+    }
+}
+
+impl AsyncRead for AsyncMessageStream {
+    fn read_buf<B: bytes::BufMut>(&mut self, buf: &mut B) -> futures::Poll<usize, std::io::Error> {
+        <&tokio_uds::UnixStream>::read_buf(&mut &self.0, buf)
+    }
+}
+
+impl AsyncWrite for AsyncMessageStream {
+    fn shutdown(&mut self) -> futures::Poll<(), std::io::Error> {
+        <&tokio_uds::UnixStream>::shutdown(&mut &self.0)
+    }
+
+    fn write_buf<B: bytes::Buf>(&mut self, buf: &mut B) -> futures::Poll<usize, std::io::Error> {
+        <&tokio_uds::UnixStream>::write_buf(&mut &self.0, buf)
+    }
+}
+
+impl AsRawFd for AsyncMessageStream {
+    fn as_raw_fd(&self) -> RawFd {
+        self.0.as_raw_fd()
+    }
+}
diff --git a/media/audioipc/audioipc/src/messagestream_win.rs b/media/audioipc/audioipc/src/messagestream_win.rs
new file mode 100644
index 0000000000..9f2a6f0bd8
--- /dev/null
+++ b/media/audioipc/audioipc/src/messagestream_win.rs
@@ -0,0 +1,111 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use super::tokio_named_pipes;
+use mio_named_pipes;
+use std::os::windows::fs::*;
+use std::os::windows::io::{AsRawHandle, FromRawHandle, IntoRawHandle, RawHandle};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use tokio_io::{AsyncRead, AsyncWrite};
+use winapi::um::winbase::FILE_FLAG_OVERLAPPED;
+
+#[derive(Debug)]
+pub struct MessageStream(miow::pipe::NamedPipe);
+pub struct AsyncMessageStream(tokio_named_pipes::NamedPipe);
+
+impl MessageStream {
+    fn new(stream: miow::pipe::NamedPipe) -> MessageStream {
+        MessageStream(stream)
+    }
+
+    pub fn anonymous_ipc_pair(
+    ) -> std::result::Result<(MessageStream, MessageStream), std::io::Error> {
+        let pipe_name = get_pipe_name();
+        let pipe_server = miow::pipe::NamedPipe::new(&pipe_name)?;
+        let pipe_client = {
+            let mut opts = std::fs::OpenOptions::new();
+            opts.read(true)
+                .write(true)
+                .custom_flags(FILE_FLAG_OVERLAPPED);
+            let file = opts.open(&pipe_name)?;
+            unsafe { miow::pipe::NamedPipe::from_raw_handle(file.into_raw_handle()) }
+        };
+        Ok((
+            MessageStream::new(pipe_server),
+            MessageStream::new(pipe_client),
+        ))
+    }
+
+    pub unsafe fn from_raw_fd(raw: super::PlatformHandleType) -> MessageStream {
+        MessageStream::new(miow::pipe::NamedPipe::from_raw_handle(raw))
+    }
+
+    pub fn into_tokio_ipc(
+        self,
+        handle: &tokio::reactor::Handle,
+    ) -> std::result::Result<AsyncMessageStream, std::io::Error> {
+        let pipe = unsafe { mio_named_pipes::NamedPipe::from_raw_handle(self.into_raw_handle()) };
+        Ok(AsyncMessageStream::new(
+            tokio_named_pipes::NamedPipe::from_pipe(pipe, handle)?,
+        ))
+    }
+}
+
+impl IntoRawHandle for MessageStream {
+    fn into_raw_handle(self) -> RawHandle {
+        self.0.into_raw_handle()
+    }
+}
+
+impl AsyncMessageStream {
+    fn new(stream: tokio_named_pipes::NamedPipe) -> AsyncMessageStream {
+        AsyncMessageStream(stream)
+    }
+}
+
+impl std::io::Read for AsyncMessageStream {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        self.0.read(buf)
+    }
+}
+
+impl std::io::Write for AsyncMessageStream {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.0.write(buf)
+    }
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.0.flush()
+    }
+}
+
+impl AsyncRead for AsyncMessageStream {
+    fn read_buf<B: bytes::BufMut>(&mut self, buf: &mut B) -> futures::Poll<usize, std::io::Error> {
+        <tokio_named_pipes::NamedPipe>::read_buf(&mut self.0, buf)
+    }
+}
+
+impl AsyncWrite for AsyncMessageStream {
+    fn shutdown(&mut self) -> futures::Poll<(), std::io::Error> {
+        <tokio_named_pipes::NamedPipe>::shutdown(&mut self.0)
+    }
+
+    fn write_buf<B: bytes::Buf>(&mut self, buf: &mut B) -> futures::Poll<usize, std::io::Error> {
+        <tokio_named_pipes::NamedPipe>::write_buf(&mut self.0, buf)
+    }
+}
+
+impl AsRawHandle for AsyncMessageStream {
+    fn as_raw_handle(&self) -> RawHandle {
+        self.0.as_raw_handle()
+    }
+}
+
+static PIPE_ID: AtomicUsize = AtomicUsize::new(0);
+
+fn get_pipe_name() -> String {
+    let pid = std::process::id();
+    let pipe_id = PIPE_ID.fetch_add(1, Ordering::SeqCst);
+    format!("\\\\.\\pipe\\cubeb-pipe-{}-{}", pid, pipe_id)
+}
diff --git a/media/audioipc/audioipc/src/msg.rs b/media/audioipc/audioipc/src/msg.rs
new file mode 100644
index 0000000000..d41cc58703
--- /dev/null
+++ b/media/audioipc/audioipc/src/msg.rs
@@ -0,0 +1,116 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+
+use iovec::unix;
+use iovec::IoVec;
+use libc;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::{cmp, io, mem, ptr};
+
+// 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)>;
+}
+
+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)> {
+        #[cfg(target_os = "linux")]
+        let flags = libc::MSG_CMSG_CLOEXEC;
+        #[cfg(not(target_os = "linux"))]
+        let flags = 0;
+        recv_msg_with_flags(self.as_raw_fd(), iov, cmsg, flags)
+    }
+}
+
+impl<T: AsRawFd> SendMsg for T {
+    fn send_msg(&mut self, iov: &[&IoVec], cmsg: &[u8]) -> io::Result<usize> {
+        send_msg_with_flags(self.as_raw_fd(), iov, cmsg, 0)
+    }
+}
+
+fn cvt(r: libc::ssize_t) -> io::Result<usize> {
+    if r == -1 {
+        Err(io::Error::last_os_error())
+    } else {
+        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,
+        }
+    }
+}
+
+pub fn recv_msg_with_flags(
+    socket: RawFd,
+    bufs: &mut [&mut IoVec],
+    cmsg: &mut [u8],
+    flags: libc::c_int,
+) -> io::Result<(usize, usize, libc::c_int)> {
+    let slice = unix::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())
+    };
+
+    let mut msghdr: libc::msghdr = unsafe { mem::zeroed() };
+    msghdr.msg_name = ptr::null_mut();
+    msghdr.msg_namelen = 0;
+    msghdr.msg_iov = slice.as_mut_ptr();
+    msghdr.msg_iovlen = len as _;
+    msghdr.msg_control = control;
+    msghdr.msg_controllen = controllen as _;
+
+    let n = cvt_r(|| unsafe { libc::recvmsg(socket, &mut msghdr as *mut _, 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,
+) -> io::Result<usize> {
+    let slice = unix::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())
+    };
+
+    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) })
+}
diff --git a/media/audioipc/audioipc/src/rpc/client/mod.rs b/media/audioipc/audioipc/src/rpc/client/mod.rs
new file mode 100644
index 0000000000..0fb8aed565
--- /dev/null
+++ b/media/audioipc/audioipc/src/rpc/client/mod.rs
@@ -0,0 +1,162 @@
+// This is a derived version of simple/pipeline/client.rs from
+// tokio_proto crate used under MIT license.
+//
+// Original version of client.rs:
+// https://github.com/tokio-rs/tokio-proto/commit/8fb8e482dcd55cf02ceee165f8e08eee799c96d3
+//
+// The following modifications were made:
+// * Simplify the code to implement RPC for pipeline requests that
+//   contain simple request/response messages:
+//   * Remove `Error` types,
+//   * Remove `bind_transport` fn & `BindTransport` type,
+//   * Remove all "Lift"ing functionality.
+//   * Remove `Service` trait since audioipc doesn't use `tokio_service`
+//     crate.
+//
+// Copyright (c) 2016 Tokio contributors
+//
+// Permission is hereby granted, free of charge, to any
+// person obtaining a copy of this software and associated
+// documentation files (the "Software"), to deal in the
+// Software without restriction, including without
+// limitation the rights to use, copy, modify, merge,
+// publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software
+// is furnished to do so, subject to the following
+// conditions:
+//
+// The above copyright notice and this permission notice
+// shall be included in all copies or substantial portions
+// of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use crate::rpc::driver::Driver;
+use crate::rpc::Handler;
+use futures::sync::oneshot;
+use futures::{Async, Future, Poll, Sink, Stream};
+use std::collections::VecDeque;
+use std::io;
+use tokio::runtime::current_thread;
+
+mod proxy;
+
+pub use self::proxy::{ClientProxy, Response};
+
+pub fn bind_client<C>(transport: C::Transport) -> proxy::ClientProxy<C::Request, C::Response>
+where
+    C: Client,
+{
+    let (tx, rx) = proxy::channel();
+
+    let fut = {
+        let handler = ClientHandler::<C> {
+            transport,
+            requests: rx,
+            in_flight: VecDeque::with_capacity(32),
+        };
+        Driver::new(handler)
+    };
+
+    // Spawn the RPC driver into task
+    current_thread::spawn(fut.map_err(|_| ()));
+
+    tx
+}
+
+pub trait Client: 'static {
+    /// Request
+    type Request: 'static;
+
+    /// Response
+    type Response: 'static;
+
+    /// The message transport, which works with async I/O objects of type `A`
+    type Transport: 'static
+        + Stream<Item = Self::Response, Error = io::Error>
+        + Sink<SinkItem = Self::Request, SinkError = io::Error>;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct ClientHandler<C>
+where
+    C: Client,
+{
+    transport: C::Transport,
+    requests: proxy::Receiver<C::Request, C::Response>,
+    in_flight: VecDeque<oneshot::Sender<C::Response>>,
+}
+
+impl<C> Handler for ClientHandler<C>
+where
+    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",
+            ));
+        }
+
+        Ok(())
+    }
+
+    /// Produce a message
+    fn produce(&mut self) -> Poll<Option<Self::Out>, io::Error> {
+        trace!("ClientHandler::produce");
+
+        // Try to get a new request
+        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!(),
+        }
+    }
+
+    /// RPC currently in flight
+    fn has_in_flight(&self) -> bool {
+        !self.in_flight.is_empty()
+    }
+}
+
+impl<C: Client> Drop for ClientHandler<C> {
+    fn drop(&mut self) {
+        let _ = self.transport.close();
+        self.in_flight.clear();
+    }
+}
diff --git a/media/audioipc/audioipc/src/rpc/client/proxy.rs b/media/audioipc/audioipc/src/rpc/client/proxy.rs
new file mode 100644
index 0000000000..bd44110d59
--- /dev/null
+++ b/media/audioipc/audioipc/src/rpc/client/proxy.rs
@@ -0,0 +1,136 @@
+// This is a derived version of client_proxy.rs from
+// tokio_proto crate used under MIT license.
+//
+// Original version of client_proxy.rs:
+// https://github.com/tokio-rs/tokio-proto/commit/8fb8e482dcd55cf02ceee165f8e08eee799c96d3
+//
+// The following modifications were made:
+// * Remove `Service` trait since audioipc doesn't use `tokio_service`
+//   crate.
+// * Remove `RefCell` from `ClientProxy` since cubeb is called from
+//   multiple threads. `mpsc::UnboundedSender` is thread safe.
+// * Simplify the interface to just request (`R`) and response (`Q`)
+//   removing error (`E`).
+// * Remove the `Envelope` type.
+// * Renamed `pair` to `channel` to represent that an `rpc::channel`
+//   is being created.
+//
+// Original License:
+//
+// Copyright (c) 2016 Tokio contributors
+//
+// Permission is hereby granted, free of charge, to any
+// person obtaining a copy of this software and associated
+// documentation files (the "Software"), to deal in the
+// Software without restriction, including without
+// limitation the rights to use, copy, modify, merge,
+// publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software
+// is furnished to do so, subject to the following
+// conditions:
+//
+// The above copyright notice and this permission notice
+// shall be included in all copies or substantial portions
+// of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use futures::sync::{mpsc, oneshot};
+use futures::{Async, Future, Poll};
+use std::fmt;
+use std::io;
+
+/// Message used to dispatch requests to the task managing the
+/// 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>,
+}
+
+pub struct ClientProxy<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(),
+        }
+    }
+}
+
+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 };
+
+    (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.unbounded_send((request, tx));
+
+        Response { inner: rx }
+    }
+}
+
+impl<R, Q> fmt::Debug for ClientProxy<R, Q>
+where
+    R: 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;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<Q, io::Error> {
+        match self.inner.poll() {
+            Ok(Async::Ready(res)) => Ok(Async::Ready(res)),
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            // Convert oneshot::Canceled into io::Error
+            Err(_) => {
+                let e = io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe");
+                Err(e)
+            }
+        }
+    }
+}
+
+impl<Q> fmt::Debug for Response<Q>
+where
+    Q: fmt::Debug,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Response {{ ... }}")
+    }
+}
diff --git a/media/audioipc/audioipc/src/rpc/driver.rs b/media/audioipc/audioipc/src/rpc/driver.rs
new file mode 100644
index 0000000000..91463deaa7
--- /dev/null
+++ b/media/audioipc/audioipc/src/rpc/driver.rs
@@ -0,0 +1,173 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::rpc::Handler;
+use futures::{Async, AsyncSink, Future, Poll, Sink, Stream};
+use std::fmt;
+use std::io;
+
+pub struct Driver<T>
+where
+    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,
+}
+
+impl<T> Driver<T>
+where
+    T: Handler,
+{
+    /// Create a new rpc driver with the given service and transport.
+    pub fn new(handler: T) -> Driver<T> {
+        Driver {
+            handler,
+            run: 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()
+    }
+
+    /// Process incoming messages off the transport.
+    fn receive_incoming(&mut self) -> io::Result<()> {
+        while self.run {
+            if let Async::Ready(req) = self.handler.transport().poll()? {
+                self.process_incoming(req)?;
+            } else {
+                break;
+            }
+        }
+        Ok(())
+    }
+
+    /// Process an incoming message
+    fn process_incoming(&mut self, req: Option<T::In>) -> io::Result<()> {
+        trace!("process_incoming");
+        // At this point, the service & transport are ready to process the
+        // request, no matter what it is.
+        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;
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Send outgoing messages to the transport.
+    fn send_outgoing(&mut self) -> io::Result<()> {
+        trace!("send_responses");
+        loop {
+            match self.handler.produce()? {
+                Async::Ready(Some(message)) => {
+                    trace!("  --> got message");
+                    self.process_outgoing(message)?;
+                }
+                Async::Ready(None) => {
+                    trace!("  --> got None");
+                    // The service is done with the connection.
+                    self.run = false;
+                    break;
+                }
+                // Nothing to dispatch
+                Async::NotReady => break,
+            }
+        }
+
+        Ok(())
+    }
+
+    fn process_outgoing(&mut self, message: T::Out) -> io::Result<()> {
+        trace!("process_outgoing");
+        assert_send(&mut self.handler.transport(), message)?;
+
+        Ok(())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.is_flushed = self.handler.transport().poll_complete()?.is_ready();
+
+        // TODO:
+        Ok(())
+    }
+
+    fn has_in_flight(&self) -> bool {
+        self.handler.has_in_flight()
+    }
+}
+
+impl<T> Future for Driver<T>
+where
+    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
+        self.receive_incoming()?;
+
+        // Handle completed responses
+        self.send_outgoing()?;
+
+        // Try flushing buffered writes
+        self.flush()?;
+
+        if self.is_done() {
+            trace!("  --> is done.");
+            return Ok(().into());
+        }
+
+        // Tick again later
+        Ok(Async::NotReady)
+    }
+}
+
+fn assert_send<S: Sink>(s: &mut S, item: S::SinkItem) -> Result<(), S::SinkError> {
+    match 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,
+{
+    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()
+    }
+}
diff --git a/media/audioipc/audioipc/src/rpc/mod.rs b/media/audioipc/audioipc/src/rpc/mod.rs
new file mode 100644
index 0000000000..8d54fc6e00
--- /dev/null
+++ b/media/audioipc/audioipc/src/rpc/mod.rs
@@ -0,0 +1,36 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use futures::{Poll, Sink, Stream};
+use std::io;
+
+mod client;
+mod driver;
+mod server;
+
+pub use self::client::{bind_client, Client, ClientProxy, Response};
+pub use self::server::{bind_server, Server};
+
+pub trait Handler {
+    /// Message type read from transport
+    type In;
+    /// Message type written to transport
+    type Out;
+    type Transport: 'static
+        + Stream<Item = Self::In, Error = io::Error>
+        + Sink<SinkItem = Self::Out, SinkError = io::Error>;
+
+    /// Mutable reference to the transport
+    fn transport(&mut self) -> &mut Self::Transport;
+
+    /// Consume a request
+    fn consume(&mut self, message: Self::In) -> io::Result<()>;
+
+    /// Produce a response
+    fn produce(&mut self) -> Poll<Option<Self::Out>, io::Error>;
+
+    /// RPC currently in flight
+    fn has_in_flight(&self) -> bool;
+}
diff --git a/media/audioipc/audioipc/src/rpc/server.rs b/media/audioipc/audioipc/src/rpc/server.rs
new file mode 100644
index 0000000000..1cb037bd8a
--- /dev/null
+++ b/media/audioipc/audioipc/src/rpc/server.rs
@@ -0,0 +1,184 @@
+// This is a derived version of simple/pipeline/server.rs from
+// tokio_proto crate used under MIT license.
+//
+// Original version of server.rs:
+// https://github.com/tokio-rs/tokio-proto/commit/8fb8e482dcd55cf02ceee165f8e08eee799c96d3
+//
+// The following modifications were made:
+// * Simplify the code to implement RPC for pipeline requests that
+//   contain simple request/response messages:
+//   * Remove `Error` types,
+//   * Remove `bind_transport` fn & `BindTransport` type,
+//   * Remove all "Lift"ing functionality.
+//   * Remove `Service` trait since audioipc doesn't use `tokio_service`
+//     crate.
+//
+// Copyright (c) 2016 Tokio contributors
+//
+// Permission is hereby granted, free of charge, to any
+// person obtaining a copy of this software and associated
+// documentation files (the "Software"), to deal in the
+// Software without restriction, including without
+// limitation the rights to use, copy, modify, merge,
+// publish, distribute, sublicense, and/or sell copies of
+// the Software, and to permit persons to whom the Software
+// is furnished to do so, subject to the following
+// conditions:
+//
+// The above copyright notice and this permission notice
+// shall be included in all copies or substantial portions
+// of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
+// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
+// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
+// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
+// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
+// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+// DEALINGS IN THE SOFTWARE.
+
+use crate::rpc::driver::Driver;
+use crate::rpc::Handler;
+use futures::{Async, Future, Poll, Sink, Stream};
+use std::collections::VecDeque;
+use std::io;
+use tokio::runtime::current_thread;
+
+/// Bind an async I/O object `io` to the `server`.
+pub fn bind_server<S>(transport: S::Transport, server: S)
+where
+    S: Server,
+{
+    let fut = {
+        let handler = ServerHandler {
+            server,
+            transport,
+            in_flight: VecDeque::with_capacity(32),
+        };
+        Driver::new(handler)
+    };
+
+    // Spawn the RPC driver into task
+    current_thread::spawn(fut.map_err(|_| ()))
+}
+
+pub trait Server: 'static {
+    /// Request
+    type Request: 'static;
+
+    /// Response
+    type Response: 'static;
+
+    /// Future
+    type Future: Future<Item = Self::Response, Error = ()>;
+
+    /// The message transport, which works with async I/O objects of
+    /// type `A`.
+    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,
+{
+    // 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>>,
+}
+
+impl<S> Handler for ServerHandler<S>
+where
+    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
+    }
+
+    /// Consume a message
+    fn consume(&mut self, request: Self::In) -> io::Result<()> {
+        trace!("ServerHandler::consume");
+        let response = self.server.process(request);
+        self.in_flight.push_back(InFlight::Active(response));
+
+        // TODO: Should the error be handled differently?
+        Ok(())
+    }
+
+    /// Produce a message
+    fn produce(&mut self) -> Poll<Option<Self::Out>, io::Error> {
+        trace!("ServerHandler::produce");
+
+        // 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(_)) => {}
+            _ => {
+                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!(),
+        }
+    }
+
+    /// RPC currently in flight
+    fn has_in_flight(&self) -> bool {
+        !self.in_flight.is_empty()
+    }
+}
+
+impl<S: Server> Drop for ServerHandler<S> {
+    fn drop(&mut self) {
+        let _ = self.transport.close();
+        self.in_flight.clear();
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+enum InFlight<F: Future<Error = ()>> {
+    Active(F),
+    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,
+            },
+            _ => return,
+        };
+        *self = InFlight::Done(res);
+    }
+}
diff --git a/media/audioipc/audioipc/src/shm.rs b/media/audioipc/audioipc/src/shm.rs
new file mode 100644
index 0000000000..99012bfab4
--- /dev/null
+++ b/media/audioipc/audioipc/src/shm.rs
@@ -0,0 +1,266 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+
+use crate::errors::*;
+use memmap::{Mmap, MmapMut, MmapOptions};
+use std::cell::UnsafeCell;
+use std::convert::TryInto;
+use std::env::temp_dir;
+use std::fs::{remove_file, File, OpenOptions};
+use std::sync::{atomic, Arc};
+
+fn open_shm_file(id: &str) -> Result<File> {
+    #[cfg(target_os = "linux")]
+    {
+        let id_cstring = std::ffi::CString::new(id).unwrap();
+        unsafe {
+            let r = libc::syscall(libc::SYS_memfd_create, id_cstring.as_ptr(), 0);
+            if r >= 0 {
+                use std::os::unix::io::FromRawFd as _;
+                return Ok(File::from_raw_fd(r.try_into().unwrap()));
+            }
+        }
+
+        let mut path = std::path::PathBuf::from("/dev/shm");
+        path.push(id);
+
+        if let Ok(file) = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create_new(true)
+            .open(&path)
+        {
+            let _ = remove_file(&path);
+            return Ok(file);
+        }
+    }
+
+    let mut path = temp_dir();
+    path.push(id);
+
+    let file = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create_new(true)
+        .open(&path)?;
+
+    let _ = remove_file(&path);
+    Ok(file)
+}
+
+#[cfg(unix)]
+fn handle_enospc(s: &str) -> Result<()> {
+    let err = std::io::Error::last_os_error();
+    let errno = err.raw_os_error().unwrap_or(0);
+    assert_ne!(errno, 0);
+    debug!("allocate_file: {} failed errno={}", s, errno);
+    if errno == libc::ENOSPC {
+        return Err(err.into());
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+fn allocate_file(file: &File, size: usize) -> Result<()> {
+    use std::os::unix::io::AsRawFd;
+
+    // First, set the file size.  This may create a sparse file on
+    // many systems, which can fail with SIGBUS when accessed via a
+    // mapping and the lazy backing allocation fails due to low disk
+    // space.  To avoid this, try to force the entire file to be
+    // preallocated before mapping using OS-specific approaches below.
+
+    file.set_len(size.try_into().unwrap())?;
+
+    let fd = file.as_raw_fd();
+    let size: libc::off_t = size.try_into().unwrap();
+
+    // Try Linux-specific fallocate.
+    #[cfg(target_os = "linux")]
+    {
+        if unsafe { libc::fallocate(fd, 0, 0, size) } == 0 {
+            return Ok(());
+        }
+        handle_enospc("fallocate()")?;
+    }
+
+    // Try macOS-specific fcntl.
+    #[cfg(target_os = "macos")]
+    {
+        let params = libc::fstore_t {
+            fst_flags: libc::F_ALLOCATEALL,
+            fst_posmode: libc::F_PEOFPOSMODE,
+            fst_offset: 0,
+            fst_length: size,
+            fst_bytesalloc: 0,
+        };
+        if unsafe { libc::fcntl(fd, libc::F_PREALLOCATE, &params) } == 0 {
+            return Ok(());
+        }
+        handle_enospc("fcntl(F_PREALLOCATE)")?;
+    }
+
+    // Fall back to portable version, where available.
+    #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "dragonfly"))]
+    {
+        if unsafe { libc::posix_fallocate(fd, 0, size) } == 0 {
+            return Ok(());
+        }
+        handle_enospc("posix_fallocate()")?;
+    }
+
+    Ok(())
+}
+
+#[cfg(windows)]
+fn allocate_file(file: &File, size: usize) -> Result<()> {
+    // CreateFileMapping will ensure the entire file is allocated
+    // before it's mapped in, so we simply set the size here.
+    file.set_len(size.try_into().unwrap())?;
+    Ok(())
+}
+
+pub struct SharedMemReader {
+    mmap: Mmap,
+}
+
+impl SharedMemReader {
+    pub fn new(id: &str, size: usize) -> Result<(SharedMemReader, File)> {
+        let file = open_shm_file(id)?;
+        allocate_file(&file, size)?;
+        let mmap = unsafe { MmapOptions::new().map(&file)? };
+        assert_eq!(mmap.len(), size);
+        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() {
+            atomic::fence(atomic::Ordering::Acquire);
+            let len = buf.len();
+            buf.copy_from_slice(&self.mmap[..len]);
+            Ok(())
+        } else {
+            bail!("mmap size");
+        }
+    }
+}
+
+pub struct SharedMemSlice {
+    mmap: Arc<Mmap>,
+}
+
+impl SharedMemSlice {
+    pub fn from(file: &File, size: usize) -> Result<SharedMemSlice> {
+        let mmap = unsafe { MmapOptions::new().map(file)? };
+        assert_eq!(mmap.len(), size);
+        let mmap = Arc::new(mmap);
+        Ok(SharedMemSlice { mmap })
+    }
+
+    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.mmap.len() {
+            atomic::fence(atomic::Ordering::Acquire);
+            let buf = &self.mmap[..size];
+            Ok(buf)
+        } else {
+            bail!("mmap size");
+        }
+    }
+
+    /// Clones the memory map.
+    ///
+    /// The underlying memory map is shared, and thus the caller must
+    /// ensure that the memory is not illegally aliased.
+    pub unsafe fn unsafe_clone(&self) -> Self {
+        SharedMemSlice {
+            mmap: self.mmap.clone(),
+        }
+    }
+}
+
+unsafe impl Send for SharedMemSlice {}
+
+pub struct SharedMemWriter {
+    mmap: MmapMut,
+}
+
+impl SharedMemWriter {
+    pub fn new(id: &str, size: usize) -> Result<(SharedMemWriter, File)> {
+        let file = open_shm_file(id)?;
+        allocate_file(&file, size)?;
+        let mmap = unsafe { MmapOptions::new().map_mut(&file)? };
+        assert_eq!(mmap.len(), size);
+        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() {
+            self.mmap[..buf.len()].copy_from_slice(buf);
+            atomic::fence(atomic::Ordering::Release);
+            Ok(())
+        } else {
+            bail!("mmap size");
+        }
+    }
+}
+
+pub struct SharedMemMutSlice {
+    mmap: Arc<UnsafeCell<MmapMut>>,
+}
+
+impl SharedMemMutSlice {
+    pub fn from(file: &File, size: usize) -> Result<SharedMemMutSlice> {
+        let mmap = unsafe { MmapOptions::new().map_mut(file)? };
+        assert_eq!(mmap.len(), size);
+        let mmap = Arc::new(UnsafeCell::new(mmap));
+        Ok(SharedMemMutSlice { mmap })
+    }
+
+    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.inner().len() {
+            let buf = &mut self.inner_mut()[..size];
+            atomic::fence(atomic::Ordering::Release);
+            Ok(buf)
+        } else {
+            bail!("mmap size");
+        }
+    }
+
+    /// Clones the memory map.
+    ///
+    /// The underlying memory map is shared, and thus the caller must
+    /// ensure that the memory is not illegally aliased.
+    pub unsafe fn unsafe_clone(&self) -> Self {
+        SharedMemMutSlice {
+            mmap: self.mmap.clone(),
+        }
+    }
+
+    fn inner(&self) -> &MmapMut {
+        unsafe { &*self.mmap.get() }
+    }
+
+    fn inner_mut(&mut self) -> &mut MmapMut {
+        unsafe { &mut *self.mmap.get() }
+    }
+}
+
+unsafe impl Send for SharedMemMutSlice {}
diff --git a/media/audioipc/audioipc/src/tokio_named_pipes.rs b/media/audioipc/audioipc/src/tokio_named_pipes.rs
new file mode 100644
index 0000000000..d402337fc7
--- /dev/null
+++ b/media/audioipc/audioipc/src/tokio_named_pipes.rs
@@ -0,0 +1,155 @@
+// Copied from tokio-named-pipes/src/lib.rs revision 49ec1ba8bbc94ab6fc9636af2a00dfb3204080c8 (tokio-named-pipes 0.2.0)
+// This file is dual licensed under the MIT and Apache-2.0 per upstream: https://github.com/NikVolf/tokio-named-pipes/blob/stable/LICENSE-MIT and https://github.com/NikVolf/tokio-named-pipes/blob/stable/LICENSE-APACHE
+// - Implement AsyncWrite::shutdown
+
+#![cfg(windows)]
+
+use std::ffi::OsStr;
+use std::fmt;
+use std::io::{Read, Write};
+use std::os::windows::io::*;
+
+use bytes::{Buf, BufMut};
+use futures::{Async, Poll};
+use mio::Ready;
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::reactor::{Handle, PollEvented2};
+
+pub struct NamedPipe {
+    io: PollEvented2<mio_named_pipes::NamedPipe>,
+}
+
+impl NamedPipe {
+    pub fn new<P: AsRef<OsStr>>(p: P, handle: &Handle) -> std::io::Result<NamedPipe> {
+        NamedPipe::_new(p.as_ref(), handle)
+    }
+
+    fn _new(p: &OsStr, handle: &Handle) -> std::io::Result<NamedPipe> {
+        let inner = mio_named_pipes::NamedPipe::new(p)?;
+        NamedPipe::from_pipe(inner, handle)
+    }
+
+    pub fn from_pipe(
+        pipe: mio_named_pipes::NamedPipe,
+        handle: &Handle,
+    ) -> std::io::Result<NamedPipe> {
+        Ok(NamedPipe {
+            io: PollEvented2::new_with_handle(pipe, handle)?,
+        })
+    }
+
+    pub fn connect(&self) -> std::io::Result<()> {
+        self.io.get_ref().connect()
+    }
+
+    pub fn disconnect(&self) -> std::io::Result<()> {
+        self.io.get_ref().disconnect()
+    }
+
+    pub fn poll_read_ready_readable(&mut self) -> tokio::io::Result<Async<Ready>> {
+        self.io.poll_read_ready(Ready::readable())
+    }
+
+    pub fn poll_write_ready(&mut self) -> tokio::io::Result<Async<Ready>> {
+        self.io.poll_write_ready()
+    }
+
+    fn io_mut(&mut self) -> &mut PollEvented2<mio_named_pipes::NamedPipe> {
+        &mut self.io
+    }
+}
+
+impl Read for NamedPipe {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        self.io.read(buf)
+    }
+}
+
+impl Write for NamedPipe {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        self.io.write(buf)
+    }
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.io.flush()
+    }
+}
+
+impl<'a> Read for &'a NamedPipe {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        (&self.io).read(buf)
+    }
+}
+
+impl<'a> Write for &'a NamedPipe {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        (&self.io).write(buf)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        (&self.io).flush()
+    }
+}
+
+impl AsyncRead for NamedPipe {
+    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
+        false
+    }
+
+    fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, std::io::Error> {
+        if let Async::NotReady = self.io.poll_read_ready(Ready::readable())? {
+            return Ok(Async::NotReady);
+        }
+
+        let mut stack_buf = [0u8; 1024];
+        let bytes_read = self.io_mut().read(&mut stack_buf);
+        match bytes_read {
+            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                self.io_mut().clear_read_ready(Ready::readable())?;
+                return Ok(Async::NotReady);
+            }
+            Err(e) => Err(e),
+            Ok(bytes_read) => {
+                buf.put_slice(&stack_buf[0..bytes_read]);
+                Ok(Async::Ready(bytes_read))
+            }
+        }
+    }
+}
+
+impl AsyncWrite for NamedPipe {
+    fn shutdown(&mut self) -> Poll<(), std::io::Error> {
+        let _ = self.disconnect();
+        Ok(().into())
+    }
+
+    fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, std::io::Error> {
+        if let Async::NotReady = self.io.poll_write_ready()? {
+            return Ok(Async::NotReady);
+        }
+
+        let bytes_wrt = self.io_mut().write(buf.bytes());
+        match bytes_wrt {
+            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                self.io_mut().clear_write_ready()?;
+                return Ok(Async::NotReady);
+            }
+            Err(e) => Err(e),
+            Ok(bytes_wrt) => {
+                buf.advance(bytes_wrt);
+                Ok(Async::Ready(bytes_wrt))
+            }
+        }
+    }
+}
+
+impl fmt::Debug for NamedPipe {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        self.io.get_ref().fmt(f)
+    }
+}
+
+impl AsRawHandle for NamedPipe {
+    fn as_raw_handle(&self) -> RawHandle {
+        self.io.get_ref().as_raw_handle()
+    }
+}
diff --git a/media/audioipc/audioipc/src/tokio_uds_stream.rs b/media/audioipc/audioipc/src/tokio_uds_stream.rs
new file mode 100644
index 0000000000..db005d7287
--- /dev/null
+++ b/media/audioipc/audioipc/src/tokio_uds_stream.rs
@@ -0,0 +1,363 @@
+// Copied from tokio-uds/src/stream.rs revision 25e835c5b7e2cfeb9c22b1fd576844f6814a9477 (tokio-uds 0.2.5)
+// License MIT per upstream: https://github.com/tokio-rs/tokio/blob/master/tokio-uds/LICENSE
+// - Removed ucred for build simplicity
+// - Added clear_{read,write}_ready per: https://github.com/tokio-rs/tokio/pull/1294
+
+use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_reactor::{Handle, PollEvented};
+
+use bytes::{Buf, BufMut};
+use futures::{Async, Future, Poll};
+use iovec::{self, IoVec};
+use libc;
+use mio::Ready;
+use mio_uds;
+
+use std::fmt;
+use std::io::{self, Read, Write};
+use std::net::Shutdown;
+use std::os::unix::io::{AsRawFd, RawFd};
+use std::os::unix::net::{self, SocketAddr};
+use std::path::Path;
+
+/// A structure representing a connected Unix socket.
+///
+/// This socket can be connected directly with `UnixStream::connect` or accepted
+/// from a listener with `UnixListener::incoming`. Additionally, a pair of
+/// anonymous Unix sockets can be created with `UnixStream::pair`.
+pub struct UnixStream {
+    io: PollEvented<mio_uds::UnixStream>,
+}
+
+/// Future returned by `UnixStream::connect` which will resolve to a
+/// `UnixStream` when the stream is connected.
+#[derive(Debug)]
+pub struct ConnectFuture {
+    inner: State,
+}
+
+#[derive(Debug)]
+enum State {
+    Waiting(UnixStream),
+    Error(io::Error),
+    Empty,
+}
+
+impl UnixStream {
+    /// Connects to the socket named by `path`.
+    ///
+    /// This function will create a new Unix socket and connect to the path
+    /// specified, associating the returned stream with the default event loop's
+    /// handle.
+    pub fn connect<P>(path: P) -> ConnectFuture
+    where
+        P: AsRef<Path>,
+    {
+        let res = mio_uds::UnixStream::connect(path).map(UnixStream::new);
+
+        let inner = match res {
+            Ok(stream) => State::Waiting(stream),
+            Err(e) => State::Error(e),
+        };
+
+        ConnectFuture { inner }
+    }
+
+    /// Consumes a `UnixStream` in the standard library and returns a
+    /// nonblocking `UnixStream` from this crate.
+    ///
+    /// The returned stream will be associated with the given event loop
+    /// specified by `handle` and is ready to perform I/O.
+    pub fn from_std(stream: net::UnixStream, handle: &Handle) -> io::Result<UnixStream> {
+        let stream = mio_uds::UnixStream::from_stream(stream)?;
+        let io = PollEvented::new_with_handle(stream, handle)?;
+
+        Ok(UnixStream { io })
+    }
+
+    /// Creates an unnamed pair of connected sockets.
+    ///
+    /// This function will create a pair of interconnected Unix sockets for
+    /// communicating back and forth between one another. Each socket will
+    /// be associated with the default event loop's handle.
+    pub fn pair() -> io::Result<(UnixStream, UnixStream)> {
+        let (a, b) = mio_uds::UnixStream::pair()?;
+        let a = UnixStream::new(a);
+        let b = UnixStream::new(b);
+
+        Ok((a, b))
+    }
+
+    pub(crate) fn new(stream: mio_uds::UnixStream) -> UnixStream {
+        let io = PollEvented::new(stream);
+        UnixStream { io }
+    }
+
+    /// Test whether this socket is ready to be read or not.
+    pub fn poll_read_ready(&self, ready: Ready) -> Poll<Ready, io::Error> {
+        self.io.poll_read_ready(ready)
+    }
+
+    /// Clear read ready state.
+    pub fn clear_read_ready(&self, ready: mio::Ready) -> io::Result<()> {
+        self.io.clear_read_ready(ready)
+    }
+
+    /// Test whether this socket is ready to be written to or not.
+    pub fn poll_write_ready(&self) -> Poll<Ready, io::Error> {
+        self.io.poll_write_ready()
+    }
+
+    /// Clear write ready state.
+    pub fn clear_write_ready(&self) -> io::Result<()> {
+        self.io.clear_write_ready()
+    }
+
+    /// Returns the socket address of the local half of this connection.
+    pub fn local_addr(&self) -> io::Result<SocketAddr> {
+        self.io.get_ref().local_addr()
+    }
+
+    /// Returns the socket address of the remote half of this connection.
+    pub fn peer_addr(&self) -> io::Result<SocketAddr> {
+        self.io.get_ref().peer_addr()
+    }
+
+    /// Returns the value of the `SO_ERROR` option.
+    pub fn take_error(&self) -> io::Result<Option<io::Error>> {
+        self.io.get_ref().take_error()
+    }
+
+    /// Shuts down the read, write, or both halves of this connection.
+    ///
+    /// This function will cause all pending and future I/O calls on the
+    /// specified portions to immediately return with an appropriate value
+    /// (see the documentation of `Shutdown`).
+    pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
+        self.io.get_ref().shutdown(how)
+    }
+}
+
+impl Read for UnixStream {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.io.read(buf)
+    }
+}
+
+impl Write for UnixStream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.io.write(buf)
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        self.io.flush()
+    }
+}
+
+impl AsyncRead for UnixStream {
+    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
+        false
+    }
+
+    fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
+        <&UnixStream>::read_buf(&mut &*self, buf)
+    }
+}
+
+impl AsyncWrite for UnixStream {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        <&UnixStream>::shutdown(&mut &*self)
+    }
+
+    fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
+        <&UnixStream>::write_buf(&mut &*self, buf)
+    }
+}
+
+impl<'a> Read for &'a UnixStream {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        (&self.io).read(buf)
+    }
+}
+
+impl<'a> Write for &'a UnixStream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        (&self.io).write(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        (&self.io).flush()
+    }
+}
+
+impl<'a> AsyncRead for &'a UnixStream {
+    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [u8]) -> bool {
+        false
+    }
+
+    fn read_buf<B: BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
+        if let Async::NotReady = <UnixStream>::poll_read_ready(self, Ready::readable())? {
+            return Ok(Async::NotReady);
+        }
+        unsafe {
+            let r = read_ready(buf, self.as_raw_fd());
+            if r == -1 {
+                let e = io::Error::last_os_error();
+                if e.kind() == io::ErrorKind::WouldBlock {
+                    self.io.clear_read_ready(Ready::readable())?;
+                    Ok(Async::NotReady)
+                } else {
+                    Err(e)
+                }
+            } else {
+                let r = r as usize;
+                buf.advance_mut(r);
+                Ok(r.into())
+            }
+        }
+    }
+}
+
+impl<'a> AsyncWrite for &'a UnixStream {
+    fn shutdown(&mut self) -> Poll<(), io::Error> {
+        Ok(().into())
+    }
+
+    fn write_buf<B: Buf>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
+        if let Async::NotReady = <UnixStream>::poll_write_ready(self)? {
+            return Ok(Async::NotReady);
+        }
+        unsafe {
+            let r = write_ready(buf, self.as_raw_fd());
+            if r == -1 {
+                let e = io::Error::last_os_error();
+                if e.kind() == io::ErrorKind::WouldBlock {
+                    self.io.clear_write_ready()?;
+                    Ok(Async::NotReady)
+                } else {
+                    Err(e)
+                }
+            } else {
+                let r = r as usize;
+                buf.advance(r);
+                Ok(r.into())
+            }
+        }
+    }
+}
+
+impl fmt::Debug for UnixStream {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.io.get_ref().fmt(f)
+    }
+}
+
+impl AsRawFd for UnixStream {
+    fn as_raw_fd(&self) -> RawFd {
+        self.io.get_ref().as_raw_fd()
+    }
+}
+
+impl Future for ConnectFuture {
+    type Item = UnixStream;
+    type Error = io::Error;
+
+    fn poll(&mut self) -> Poll<UnixStream, io::Error> {
+        use std::mem;
+
+        match self.inner {
+            State::Waiting(ref mut stream) => {
+                if let Async::NotReady = stream.io.poll_write_ready()? {
+                    return Ok(Async::NotReady);
+                }
+
+                if let Some(e) = stream.io.get_ref().take_error()? {
+                    return Err(e);
+                }
+            }
+            State::Error(_) => {
+                let e = match mem::replace(&mut self.inner, State::Empty) {
+                    State::Error(e) => e,
+                    _ => unreachable!(),
+                };
+
+                return Err(e);
+            }
+            State::Empty => panic!("can't poll stream twice"),
+        }
+
+        match mem::replace(&mut self.inner, State::Empty) {
+            State::Waiting(stream) => Ok(Async::Ready(stream)),
+            _ => unreachable!(),
+        }
+    }
+}
+
+unsafe fn read_ready<B: BufMut>(buf: &mut B, raw_fd: RawFd) -> isize {
+    // The `IoVec` type can't have a 0-length size, so we create a bunch
+    // of dummy versions on the stack with 1 length which we'll quickly
+    // overwrite.
+    let b1: &mut [u8] = &mut [0];
+    let b2: &mut [u8] = &mut [0];
+    let b3: &mut [u8] = &mut [0];
+    let b4: &mut [u8] = &mut [0];
+    let b5: &mut [u8] = &mut [0];
+    let b6: &mut [u8] = &mut [0];
+    let b7: &mut [u8] = &mut [0];
+    let b8: &mut [u8] = &mut [0];
+    let b9: &mut [u8] = &mut [0];
+    let b10: &mut [u8] = &mut [0];
+    let b11: &mut [u8] = &mut [0];
+    let b12: &mut [u8] = &mut [0];
+    let b13: &mut [u8] = &mut [0];
+    let b14: &mut [u8] = &mut [0];
+    let b15: &mut [u8] = &mut [0];
+    let b16: &mut [u8] = &mut [0];
+    let mut bufs: [&mut IoVec; 16] = [
+        b1.into(),
+        b2.into(),
+        b3.into(),
+        b4.into(),
+        b5.into(),
+        b6.into(),
+        b7.into(),
+        b8.into(),
+        b9.into(),
+        b10.into(),
+        b11.into(),
+        b12.into(),
+        b13.into(),
+        b14.into(),
+        b15.into(),
+        b16.into(),
+    ];
+
+    let n = buf.bytes_vec_mut(&mut bufs);
+    read_ready_vecs(&mut bufs[..n], raw_fd)
+}
+
+unsafe fn read_ready_vecs(bufs: &mut [&mut IoVec], raw_fd: RawFd) -> isize {
+    let iovecs = iovec::unix::as_os_slice_mut(bufs);
+
+    libc::readv(raw_fd, iovecs.as_ptr(), iovecs.len() as i32)
+}
+
+unsafe fn write_ready<B: Buf>(buf: &mut B, raw_fd: RawFd) -> isize {
+    // 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 iovec = <&IoVec>::from(DUMMY);
+    let mut bufs = [
+        iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec, iovec,
+        iovec, iovec, iovec,
+    ];
+
+    let n = buf.bytes_vec(&mut bufs);
+    write_ready_vecs(&bufs[..n], raw_fd)
+}
+
+unsafe fn write_ready_vecs(bufs: &[&IoVec], raw_fd: RawFd) -> isize {
+    let iovecs = iovec::unix::as_os_slice(bufs);
+
+    libc::writev(raw_fd, iovecs.as_ptr(), iovecs.len() as i32)
+}
diff --git a/media/audioipc/client/Cargo.toml b/media/audioipc/client/Cargo.toml
new file mode 100644
index 0000000000..eff18b1a02
--- /dev/null
+++ b/media/audioipc/client/Cargo.toml
@@ -0,0 +1,18 @@
+[package]
+name = "audioipc-client"
+version = "0.4.0"
+authors = [
+        "Matthew Gregan <kinetik@flim.org>",
+        "Dan Glastonbury <dan.glastonbury@gmail.com>"
+        ]
+description = "Cubeb Backend for talking to remote cubeb server."
+edition = "2018"
+
+[dependencies]
+audio_thread_priority = "0.23.4"
+audioipc = { path="../audioipc" }
+cubeb-backend = "0.8"
+futures = { version="0.1.18", default-features=false, features=["use_std"] }
+futures-cpupool = { version="0.1.8", default-features=false }
+log = "0.4"
+tokio = "0.1"
diff --git a/media/audioipc/client/cbindgen.toml b/media/audioipc/client/cbindgen.toml
new file mode 100644
index 0000000000..9a7c204fe4
--- /dev/null
+++ b/media/audioipc/client/cbindgen.toml
@@ -0,0 +1,28 @@
+header = """/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */"""
+autogen_warning = """/* DO NOT MODIFY THIS MANUALLY! This file was generated using cbindgen. */"""
+include_version = true
+braces = "SameLine"
+line_length = 100
+tab_width = 2
+language = "C++"
+namespaces = ["mozilla", "audioipc"]
+
+[export]
+item_types = ["globals", "enums", "structs", "unions", "typedefs", "opaque", "functions", "constants"]
+
+[parse]
+parse_deps = true
+include = ['audioipc']
+
+[fn]
+args = "Vertical"
+rename_args = "GeckoCase"
+
+[struct]
+rename_fields = "GeckoCase"
+
+[defines]
+"windows" = "XP_WIN"
+"unix" = "XP_UNIX"
diff --git a/media/audioipc/client/src/context.rs b/media/audioipc/client/src/context.rs
new file mode 100644
index 0000000000..7441d0c0e0
--- /dev/null
+++ b/media/audioipc/client/src/context.rs
@@ -0,0 +1,451 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::stream;
+use crate::{assert_not_in_callback, run_in_callback};
+use crate::{ClientStream, AUDIOIPC_INIT_PARAMS};
+#[cfg(target_os = "linux")]
+use audio_thread_priority::get_current_thread_info;
+#[cfg(not(target_os = "linux"))]
+use audio_thread_priority::promote_current_thread_to_real_time;
+use audioipc::codec::LengthDelimitedCodec;
+use audioipc::frame::{framed, Framed};
+use audioipc::platformhandle_passing::{framed_with_platformhandles, FramedWithPlatformHandles};
+use audioipc::{core, rpc};
+use audioipc::{
+    messages, messages::DeviceCollectionReq, messages::DeviceCollectionResp, ClientMessage,
+    ServerMessage,
+};
+use cubeb_backend::{
+    ffi, Context, ContextOps, DeviceCollectionRef, DeviceId, DeviceType, Error, Ops, Result,
+    Stream, StreamParams, StreamParamsRef,
+};
+use futures::Future;
+use futures_cpupool::{CpuFuture, CpuPool};
+use std::ffi::{CStr, CString};
+use std::os::raw::c_void;
+use std::sync::mpsc;
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::{fmt, io, mem, ptr};
+use tokio::reactor;
+use tokio::runtime::current_thread;
+
+struct CubebClient;
+
+impl rpc::Client for CubebClient {
+    type Request = ServerMessage;
+    type Response = ClientMessage;
+    type Transport = FramedWithPlatformHandles<
+        audioipc::AsyncMessageStream,
+        LengthDelimitedCodec<Self::Request, Self::Response>,
+    >;
+}
+
+pub const CLIENT_OPS: Ops = capi_new!(ClientContext, ClientStream);
+
+// ClientContext's layout *must* match cubeb.c's `struct cubeb` for the
+// common fields.
+#[repr(C)]
+pub struct ClientContext {
+    _ops: *const Ops,
+    rpc: rpc::ClientProxy<ServerMessage, ClientMessage>,
+    core: core::CoreThread,
+    cpu_pool: CpuPool,
+    backend_id: CString,
+    device_collection_rpc: bool,
+    input_device_callback: Arc<Mutex<DeviceCollectionCallback>>,
+    output_device_callback: Arc<Mutex<DeviceCollectionCallback>>,
+}
+
+impl ClientContext {
+    #[doc(hidden)]
+    pub fn handle(&self) -> current_thread::Handle {
+        self.core.handle()
+    }
+
+    #[doc(hidden)]
+    pub fn rpc(&self) -> rpc::ClientProxy<ServerMessage, ClientMessage> {
+        self.rpc.clone()
+    }
+
+    #[doc(hidden)]
+    pub fn cpu_pool(&self) -> CpuPool {
+        self.cpu_pool.clone()
+    }
+}
+
+#[cfg(target_os = "linux")]
+fn promote_thread(rpc: &rpc::ClientProxy<ServerMessage, ClientMessage>) {
+    match get_current_thread_info() {
+        Ok(info) => {
+            let bytes = info.serialize();
+            // Don't wait for the response, this is on the callback thread, which must not block.
+            rpc.call(ServerMessage::PromoteThreadToRealTime(bytes));
+        }
+        Err(_) => {
+            warn!("Could not remotely promote thread to RT.");
+        }
+    }
+}
+
+#[cfg(not(target_os = "linux"))]
+fn promote_thread(_rpc: &rpc::ClientProxy<ServerMessage, ClientMessage>) {
+    match promote_current_thread_to_real_time(0, 48000) {
+        Ok(_) => {
+            info!("Audio thread promoted to real-time.");
+        }
+        Err(_) => {
+            warn!("Could not promote thread to real-time.");
+        }
+    }
+}
+
+fn register_thread(callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>) {
+    if let Some(func) = callback {
+        let thr = thread::current();
+        let name = CString::new(thr.name().unwrap()).unwrap();
+        func(name.as_ptr());
+    }
+}
+
+fn unregister_thread(callback: Option<extern "C" fn()>) {
+    if let Some(func) = callback {
+        func();
+    }
+}
+
+fn promote_and_register_thread(
+    rpc: &rpc::ClientProxy<ServerMessage, ClientMessage>,
+    callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>,
+) {
+    promote_thread(rpc);
+    register_thread(callback);
+}
+
+#[derive(Default)]
+struct DeviceCollectionCallback {
+    cb: ffi::cubeb_device_collection_changed_callback,
+    user_ptr: usize,
+}
+
+struct DeviceCollectionServer {
+    input_device_callback: Arc<Mutex<DeviceCollectionCallback>>,
+    output_device_callback: Arc<Mutex<DeviceCollectionCallback>>,
+    cpu_pool: CpuPool,
+}
+
+impl rpc::Server for DeviceCollectionServer {
+    type Request = DeviceCollectionReq;
+    type Response = DeviceCollectionResp;
+    type Future = CpuFuture<Self::Response, ()>;
+    type Transport =
+        Framed<audioipc::AsyncMessageStream, LengthDelimitedCodec<Self::Response, Self::Request>>;
+
+    fn process(&mut self, req: Self::Request) -> Self::Future {
+        match req {
+            DeviceCollectionReq::DeviceChange(device_type) => {
+                trace!(
+                    "ctx_thread: DeviceChange Callback: device_type={}",
+                    device_type
+                );
+
+                let devtype = cubeb_backend::DeviceType::from_bits_truncate(device_type);
+
+                let (input_cb, input_user_ptr) = {
+                    let dcb = self.input_device_callback.lock().unwrap();
+                    (dcb.cb, dcb.user_ptr)
+                };
+                let (output_cb, output_user_ptr) = {
+                    let dcb = self.output_device_callback.lock().unwrap();
+                    (dcb.cb, dcb.user_ptr)
+                };
+
+                self.cpu_pool.spawn_fn(move || {
+                    run_in_callback(|| {
+                        if devtype.contains(cubeb_backend::DeviceType::INPUT) {
+                            unsafe {
+                                input_cb.unwrap()(ptr::null_mut(), input_user_ptr as *mut c_void)
+                            }
+                        }
+                        if devtype.contains(cubeb_backend::DeviceType::OUTPUT) {
+                            unsafe {
+                                output_cb.unwrap()(ptr::null_mut(), output_user_ptr as *mut c_void)
+                            }
+                        }
+                    });
+
+                    Ok(DeviceCollectionResp::DeviceChange)
+                })
+            }
+        }
+    }
+}
+
+impl ContextOps for ClientContext {
+    fn init(_context_name: Option<&CStr>) -> Result<Context> {
+        fn bind_and_send_client(
+            stream: audioipc::AsyncMessageStream,
+            tx_rpc: &mpsc::Sender<rpc::ClientProxy<ServerMessage, ClientMessage>>,
+        ) -> io::Result<()> {
+            let transport = framed_with_platformhandles(stream, Default::default());
+            let rpc = rpc::bind_client::<CubebClient>(transport);
+            // If send fails then the rx end has closed
+            // which is unlikely here.
+            let _ = tx_rpc.send(rpc);
+            Ok(())
+        }
+
+        assert_not_in_callback();
+
+        let (tx_rpc, rx_rpc) = mpsc::channel();
+
+        let params = AUDIOIPC_INIT_PARAMS.with(|p| p.replace(None).unwrap());
+        let thread_create_callback = params.thread_create_callback;
+        let thread_destroy_callback = params.thread_destroy_callback;
+
+        let server_stream =
+            unsafe { audioipc::MessageStream::from_raw_fd(params.server_connection) };
+
+        let core = core::spawn_thread(
+            "AudioIPC Client RPC",
+            move || {
+                let handle = reactor::Handle::default();
+
+                register_thread(thread_create_callback);
+
+                server_stream
+                    .into_tokio_ipc(&handle)
+                    .and_then(|stream| bind_and_send_client(stream, &tx_rpc))
+            },
+            move || unregister_thread(thread_destroy_callback),
+        )
+        .map_err(|_| Error::default())?;
+
+        let rpc = rx_rpc.recv().map_err(|_| Error::default())?;
+        let rpc2 = rpc.clone();
+
+        // Don't let errors bubble from here.  Later calls against this context
+        // will return errors the caller expects to handle.
+        let _ = send_recv!(rpc, ClientConnect(std::process::id()) => ClientConnected);
+
+        let backend_id = send_recv!(rpc, ContextGetBackendId => ContextBackendId())
+            .unwrap_or_else(|_| "(remote error)".to_string());
+        let backend_id = CString::new(backend_id).expect("backend_id query failed");
+
+        let cpu_pool = futures_cpupool::Builder::new()
+            .name_prefix("AudioIPC")
+            .after_start(move || promote_and_register_thread(&rpc2, thread_create_callback))
+            .before_stop(move || unregister_thread(thread_destroy_callback))
+            .pool_size(params.pool_size)
+            .stack_size(params.stack_size)
+            .create();
+
+        let ctx = Box::new(ClientContext {
+            _ops: &CLIENT_OPS as *const _,
+            rpc,
+            core,
+            cpu_pool,
+            backend_id,
+            device_collection_rpc: false,
+            input_device_callback: Arc::new(Mutex::new(Default::default())),
+            output_device_callback: Arc::new(Mutex::new(Default::default())),
+        });
+        Ok(unsafe { Context::from_ptr(Box::into_raw(ctx) as *mut _) })
+    }
+
+    fn backend_id(&mut self) -> &CStr {
+        assert_not_in_callback();
+        self.backend_id.as_c_str()
+    }
+
+    fn max_channel_count(&mut self) -> Result<u32> {
+        assert_not_in_callback();
+        send_recv!(self.rpc(), ContextGetMaxChannelCount => ContextMaxChannelCount())
+    }
+
+    fn min_latency(&mut self, params: StreamParams) -> Result<u32> {
+        assert_not_in_callback();
+        let params = messages::StreamParams::from(params.as_ref());
+        send_recv!(self.rpc(), ContextGetMinLatency(params) => ContextMinLatency())
+    }
+
+    fn preferred_sample_rate(&mut self) -> Result<u32> {
+        assert_not_in_callback();
+        send_recv!(self.rpc(), ContextGetPreferredSampleRate => ContextPreferredSampleRate())
+    }
+
+    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),
+        };
+        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(())
+    }
+
+    fn device_collection_destroy(&mut self, collection: &mut DeviceCollectionRef) -> Result<()> {
+        assert_not_in_callback();
+        unsafe {
+            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,
+            );
+            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(
+        &mut self,
+        stream_name: Option<&CStr>,
+        input_device: DeviceId,
+        input_stream_params: Option<&StreamParamsRef>,
+        output_device: DeviceId,
+        output_stream_params: Option<&StreamParamsRef>,
+        latency_frames: 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<Stream> {
+        assert_not_in_callback();
+
+        fn opt_stream_params(p: Option<&StreamParamsRef>) -> Option<messages::StreamParams> {
+            match p {
+                Some(p) => Some(messages::StreamParams::from(p)),
+                None => None,
+            }
+        }
+
+        let stream_name = match stream_name {
+            Some(s) => Some(s.to_bytes_with_nul().to_vec()),
+            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,
+            input_device: input_device as usize,
+            input_stream_params,
+            output_device: output_device as usize,
+            output_stream_params,
+            latency_frames,
+        };
+        stream::init(self, init_params, data_callback, state_callback, user_ptr)
+    }
+
+    fn register_device_collection_changed(
+        &mut self,
+        devtype: DeviceType,
+        collection_changed_callback: ffi::cubeb_device_collection_changed_callback,
+        user_ptr: *mut c_void,
+    ) -> Result<()> {
+        assert_not_in_callback();
+
+        if !self.device_collection_rpc {
+            let fds = send_recv!(self.rpc(),
+                                 ContextSetupDeviceCollectionCallback =>
+                                 ContextSetupDeviceCollectionCallback())?;
+
+            let stream =
+                unsafe { audioipc::MessageStream::from_raw_fd(fds.platform_handles[0].into_raw()) };
+
+            // TODO: The lowest comms layer expects exactly 3 PlatformHandles, but we only
+            // need one here.  Drop the dummy handles the other side sent us to discard.
+            unsafe {
+                fds.platform_handles[1].into_file();
+                fds.platform_handles[2].into_file();
+            }
+
+            let server = DeviceCollectionServer {
+                input_device_callback: self.input_device_callback.clone(),
+                output_device_callback: self.output_device_callback.clone(),
+                cpu_pool: self.cpu_pool(),
+            };
+
+            let (wait_tx, wait_rx) = mpsc::channel();
+            self.handle()
+                .spawn(futures::future::lazy(move || {
+                    let handle = reactor::Handle::default();
+                    let stream = stream.into_tokio_ipc(&handle).unwrap();
+                    let transport = framed(stream, Default::default());
+                    rpc::bind_server(transport, server);
+                    wait_tx.send(()).unwrap();
+                    Ok(())
+                }))
+                .expect("Failed to spawn DeviceCollectionServer");
+            wait_rx.recv().unwrap();
+            self.device_collection_rpc = true;
+        }
+
+        if devtype.contains(cubeb_backend::DeviceType::INPUT) {
+            let mut cb = self.input_device_callback.lock().unwrap();
+            cb.cb = collection_changed_callback;
+            cb.user_ptr = user_ptr as usize;
+        }
+        if devtype.contains(cubeb_backend::DeviceType::OUTPUT) {
+            let mut cb = self.output_device_callback.lock().unwrap();
+            cb.cb = collection_changed_callback;
+            cb.user_ptr = user_ptr as usize;
+        }
+
+        let enable = collection_changed_callback.is_some();
+        send_recv!(self.rpc(),
+                   ContextRegisterDeviceCollectionChanged(devtype.bits(), enable) =>
+                   ContextRegisteredDeviceCollectionChanged)
+    }
+}
+
+impl Drop for ClientContext {
+    fn drop(&mut self) {
+        debug!("ClientContext dropped...");
+        let _ = send_recv!(self.rpc(), ClientDisconnect => ClientDisconnected);
+    }
+}
+
+impl fmt::Debug for ClientContext {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ClientContext")
+            .field("_ops", &self._ops)
+            .field("rpc", &self.rpc)
+            .field("core", &self.core)
+            .field("cpu_pool", &"...")
+            .finish()
+    }
+}
diff --git a/media/audioipc/client/src/lib.rs b/media/audioipc/client/src/lib.rs
new file mode 100644
index 0000000000..b9363ea2df
--- /dev/null
+++ b/media/audioipc/client/src/lib.rs
@@ -0,0 +1,84 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+#![warn(unused_extern_crates)]
+
+#[macro_use]
+extern crate cubeb_backend;
+#[macro_use]
+extern crate log;
+
+#[macro_use]
+mod send_recv;
+mod context;
+mod stream;
+
+use crate::context::ClientContext;
+use crate::stream::ClientStream;
+use audioipc::PlatformHandleType;
+use cubeb_backend::{capi, ffi};
+use std::os::raw::{c_char, c_int};
+
+thread_local!(static IN_CALLBACK: std::cell::RefCell<bool> = std::cell::RefCell::new(false));
+thread_local!(static AUDIOIPC_INIT_PARAMS: std::cell::RefCell<Option<AudioIpcInitParams>> = std::cell::RefCell::new(None));
+
+// This must match the definition of AudioIpcInitParams in
+// dom/media/CubebUtils.cpp in Gecko.
+#[repr(C)]
+#[derive(Clone, Copy, Debug)]
+pub struct AudioIpcInitParams {
+    // Fields only need to be public for ipctest.
+    pub server_connection: PlatformHandleType,
+    pub pool_size: usize,
+    pub stack_size: usize,
+    pub thread_create_callback: Option<extern "C" fn(*const ::std::os::raw::c_char)>,
+    pub thread_destroy_callback: Option<extern "C" fn()>,
+}
+
+unsafe impl Send for AudioIpcInitParams {}
+
+fn set_in_callback(in_callback: bool) {
+    IN_CALLBACK.with(|b| {
+        assert_eq!(*b.borrow(), !in_callback);
+        *b.borrow_mut() = in_callback;
+    });
+}
+
+fn run_in_callback<F, R>(f: F) -> R
+where
+    F: FnOnce() -> R,
+{
+    set_in_callback(true);
+
+    let r = f();
+
+    set_in_callback(false);
+
+    r
+}
+
+fn assert_not_in_callback() {
+    IN_CALLBACK.with(|b| {
+        assert_eq!(*b.borrow(), false);
+    });
+}
+
+#[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,
+    init_params: *const AudioIpcInitParams,
+) -> c_int {
+    if init_params.is_null() {
+        return cubeb_backend::ffi::CUBEB_ERROR;
+    }
+
+    let init_params = &*init_params;
+
+    AUDIOIPC_INIT_PARAMS.with(|p| {
+        *p.borrow_mut() = Some(*init_params);
+    });
+    capi::capi_init::<ClientContext>(c, context_name)
+}
diff --git a/media/audioipc/client/src/send_recv.rs b/media/audioipc/client/src/send_recv.rs
new file mode 100644
index 0000000000..482b4fa6e3
--- /dev/null
+++ b/media/audioipc/client/src/send_recv.rs
@@ -0,0 +1,73 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details.
+
+use cubeb_backend::Error;
+use std::os::raw::c_int;
+
+#[doc(hidden)]
+pub fn _err<E>(e: E) -> Error
+where
+    E: Into<Option<c_int>>,
+{
+    match e.into() {
+        Some(e) => unsafe { Error::from_raw(e) },
+        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)
+    }};
+    ($rpc:expr, $smsg:ident => $rmsg:ident()) => {{
+        let resp = send_recv!(__send $rpc, $smsg);
+        send_recv!(__recv resp, $rmsg __result)
+    }};
+    ($rpc:expr, $smsg:ident($($a:expr),*) => $rmsg:ident) => {{
+        let resp = send_recv!(__send $rpc, $smsg, $($a),*);
+        send_recv!(__recv resp, $rmsg)
+    }};
+    ($rpc:expr, $smsg:ident($($a:expr),*) => $rmsg:ident()) => {{
+        let resp = send_recv!(__send $rpc, $smsg, $($a),*);
+        send_recv!(__recv resp, $rmsg __result)
+    }};
+    //
+    (__send $rpc:expr, $smsg:ident) => ({
+        $rpc.call(ServerMessage::$smsg)
+    });
+    (__send $rpc:expr, $smsg:ident, $($a:expr),*) => ({
+        $rpc.call(ServerMessage::$smsg($($a),*))
+    });
+    (__recv $resp:expr, $rmsg:ident) => ({
+        match $resp.wait() {
+            Ok(ClientMessage::$rmsg) => Ok(()),
+            Ok(ClientMessage::Error(e)) => Err($crate::send_recv::_err(e)),
+            Ok(m) => {
+                debug!("received wrong message - got={:?}", m);
+                Err($crate::send_recv::_err(None))
+            },
+            Err(e) => {
+                debug!("received error from rpc - got={:?}", e);
+                Err($crate::send_recv::_err(None))
+            },
+        }
+    });
+    (__recv $resp:expr, $rmsg:ident __result) => ({
+        match $resp.wait() {
+            Ok(ClientMessage::$rmsg(v)) => Ok(v),
+            Ok(ClientMessage::Error(e)) => Err($crate::send_recv::_err(e)),
+            Ok(m) => {
+                debug!("received wrong message - got={:?}", m);
+                Err($crate::send_recv::_err(None))
+            },
+            Err(e) => {
+                debug!("received error - got={:?}", e);
+                Err($crate::send_recv::_err(None))
+            },
+        }
+    })
+}
diff --git a/media/audioipc/client/src/stream.rs b/media/audioipc/client/src/stream.rs
new file mode 100644
index 0000000000..00a335c720
--- /dev/null
+++ b/media/audioipc/client/src/stream.rs
@@ -0,0 +1,357 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+use crate::ClientContext;
+use crate::{assert_not_in_callback, run_in_callback};
+use audioipc::frame::{framed, Framed};
+use audioipc::messages::{self, CallbackReq, CallbackResp, ClientMessage, ServerMessage};
+use audioipc::rpc;
+use audioipc::shm::{SharedMemMutSlice, SharedMemSlice};
+use audioipc::{codec::LengthDelimitedCodec, messages::StreamCreateParams};
+use cubeb_backend::{ffi, DeviceRef, Error, Result, Stream, StreamOps};
+use futures::Future;
+use futures_cpupool::{CpuFuture, CpuPool};
+use std::ffi::{CStr, CString};
+use std::os::raw::c_void;
+use std::ptr;
+use std::sync::mpsc;
+use std::sync::{Arc, Mutex};
+use tokio::reactor;
+
+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 _);
+            }
+        }
+    }
+}
+
+// ClientStream's layout *must* match cubeb.c's `struct cubeb_stream` for the
+// common fields.
+#[repr(C)]
+#[derive(Debug)]
+pub struct ClientStream<'ctx> {
+    // This must be a reference to Context for cubeb, cubeb accesses
+    // stream methods via stream->context->ops
+    context: &'ctx ClientContext,
+    user_ptr: *mut c_void,
+    token: usize,
+    device_change_cb: Arc<Mutex<ffi::cubeb_device_changed_callback>>,
+}
+
+struct CallbackServer {
+    input_shm: Option<SharedMemSlice>,
+    output_shm: Option<SharedMemMutSlice>,
+    data_cb: ffi::cubeb_data_callback,
+    state_cb: ffi::cubeb_state_callback,
+    user_ptr: usize,
+    cpu_pool: CpuPool,
+    device_change_cb: Arc<Mutex<ffi::cubeb_device_changed_callback>>,
+}
+
+impl rpc::Server for CallbackServer {
+    type Request = CallbackReq;
+    type Response = CallbackResp;
+    type Future = CpuFuture<Self::Response, ()>;
+    type Transport =
+        Framed<audioipc::AsyncMessageStream, LengthDelimitedCodec<Self::Response, Self::Request>>;
+
+    fn process(&mut self, req: Self::Request) -> Self::Future {
+        match req {
+            CallbackReq::Data {
+                nframes,
+                input_frame_size,
+                output_frame_size,
+            } => {
+                trace!(
+                    "stream_thread: Data Callback: nframes={} input_fs={} output_fs={}",
+                    nframes,
+                    input_frame_size,
+                    output_frame_size,
+                );
+
+                // Clone values that need to be moved into the cpu pool thread.
+                let input_shm = match self.input_shm {
+                    Some(ref shm) => unsafe { Some(shm.unsafe_clone()) },
+                    None => None,
+                };
+                let mut output_shm = match self.output_shm {
+                    Some(ref shm) => unsafe { Some(shm.unsafe_clone()) },
+                    None => None,
+                };
+                let user_ptr = self.user_ptr;
+                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 = match input_shm {
+                        Some(shm) => shm
+                            .get_slice(nframes as usize * input_frame_size)
+                            .unwrap()
+                            .as_ptr(),
+                        None => ptr::null(),
+                    };
+                    let output_ptr: *mut u8 = match output_shm {
+                        Some(ref mut shm) => shm
+                            .get_mut_slice(nframes as usize * output_frame_size)
+                            .unwrap()
+                            .as_mut_ptr(),
+                        None => ptr::null_mut(),
+                    };
+
+                    run_in_callback(|| {
+                        let nframes = unsafe {
+                            cb(
+                                ptr::null_mut(),
+                                user_ptr as *mut c_void,
+                                input_ptr as *const _,
+                                output_ptr as *mut _,
+                                nframes as _,
+                            )
+                        };
+
+                        Ok(CallbackResp::Data(nframes as isize))
+                    })
+                })
+            }
+            CallbackReq::State(state) => {
+                trace!("stream_thread: State Callback: {:?}", state);
+                let user_ptr = self.user_ptr;
+                let cb = self.state_cb.unwrap();
+                self.cpu_pool.spawn_fn(move || {
+                    run_in_callback(|| unsafe {
+                        cb(ptr::null_mut(), user_ptr as *mut _, state);
+                    });
+
+                    Ok(CallbackResp::State)
+                })
+            }
+            CallbackReq::DeviceChange => {
+                let cb = self.device_change_cb.clone();
+                let user_ptr = self.user_ptr;
+                self.cpu_pool.spawn_fn(move || {
+                    run_in_callback(|| {
+                        let cb = cb.lock().unwrap();
+                        if let Some(cb) = *cb {
+                            unsafe {
+                                cb(user_ptr as *mut _);
+                            }
+                        } else {
+                            warn!("DeviceChange received with null callback");
+                        }
+                    });
+
+                    Ok(CallbackResp::DeviceChange)
+                })
+            }
+        }
+    }
+}
+
+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<Stream> {
+        assert_not_in_callback();
+
+        let rpc = ctx.rpc();
+        let create_params = StreamCreateParams {
+            input_stream_params: init_params.input_stream_params,
+            output_stream_params: init_params.output_stream_params,
+        };
+        let data = send_recv!(rpc, StreamCreate(create_params) => StreamCreated())?;
+
+        debug!(
+            "token = {}, handles = {:?}",
+            data.token, data.platform_handles
+        );
+
+        let has_input = init_params.input_stream_params.is_some();
+        let has_output = init_params.output_stream_params.is_some();
+
+        let stream =
+            unsafe { audioipc::MessageStream::from_raw_fd(data.platform_handles[0].into_raw()) };
+
+        let input_file = unsafe { data.platform_handles[1].into_file() };
+        let input_shm = if has_input {
+            match SharedMemSlice::from(&input_file, audioipc::SHM_AREA_SIZE) {
+                Ok(shm) => Some(shm),
+                Err(e) => {
+                    debug!("Client failed to set up input shmem: {}", e);
+                    return Err(Error::error());
+                }
+            }
+        } else {
+            None
+        };
+
+        let output_file = unsafe { data.platform_handles[2].into_file() };
+        let output_shm = if has_output {
+            match SharedMemMutSlice::from(&output_file, audioipc::SHM_AREA_SIZE) {
+                Ok(shm) => Some(shm),
+                Err(e) => {
+                    debug!("Client failed to set up output shmem: {}", e);
+                    return Err(Error::error());
+                }
+            }
+        } else {
+            None
+        };
+
+        let user_data = user_ptr as usize;
+
+        let cpu_pool = ctx.cpu_pool();
+
+        let null_cb: ffi::cubeb_device_changed_callback = None;
+        let device_change_cb = Arc::new(Mutex::new(null_cb));
+
+        let server = CallbackServer {
+            input_shm,
+            output_shm,
+            data_cb: data_callback,
+            state_cb: state_callback,
+            user_ptr: user_data,
+            cpu_pool,
+            device_change_cb: device_change_cb.clone(),
+        };
+
+        let (wait_tx, wait_rx) = mpsc::channel();
+        ctx.handle()
+            .spawn(futures::future::lazy(move || {
+                let handle = reactor::Handle::default();
+                let stream = stream.into_tokio_ipc(&handle).unwrap();
+                let transport = framed(stream, Default::default());
+                rpc::bind_server(transport, server);
+                wait_tx.send(()).unwrap();
+                Ok(())
+            }))
+            .expect("Failed to spawn CallbackServer");
+        wait_rx.recv().unwrap();
+
+        send_recv!(rpc, StreamInit(data.token, init_params) => StreamInitialized)?;
+
+        let stream = Box::into_raw(Box::new(ClientStream {
+            context: ctx,
+            user_ptr,
+            token: data.token,
+            device_change_cb,
+        }));
+        Ok(unsafe { Stream::from_ptr(stream as *mut _) })
+    }
+}
+
+impl<'ctx> Drop for ClientStream<'ctx> {
+    fn drop(&mut self) {
+        debug!("ClientStream dropped...");
+        let rpc = self.context.rpc();
+        let _ = send_recv!(rpc, StreamDestroy(self.token) => StreamDestroyed);
+    }
+}
+
+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(&mut self) -> Result<()> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamStop(self.token) => StreamStopped)
+    }
+
+    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(&mut self) -> Result<u64> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamGetPosition(self.token) => StreamPosition())
+    }
+
+    fn latency(&mut self) -> Result<u32> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamGetLatency(self.token) => StreamLatency())
+    }
+
+    fn input_latency(&mut self) -> Result<u32> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamGetInputLatency(self.token) => StreamInputLatency())
+    }
+
+    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_name(&mut self, name: &CStr) -> Result<()> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        send_recv!(rpc, StreamSetName(self.token, name.to_owned()) => StreamNameSet)
+    }
+
+    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(unsafe { DeviceRef::from_ptr(Box::into_raw(Box::new(d.into()))) }),
+            Err(e) => Err(e),
+        }
+    }
+
+    fn device_destroy(&mut self, device: &DeviceRef) -> Result<()> {
+        assert_not_in_callback();
+        // It's all unsafe...
+        if device.as_ptr().is_null() {
+            Err(Error::error())
+        } else {
+            unsafe {
+                let _: Box<Device> = Box::from_raw(device.as_ptr() as *mut _);
+            }
+            Ok(())
+        }
+    }
+
+    fn register_device_changed_callback(
+        &mut self,
+        device_changed_callback: ffi::cubeb_device_changed_callback,
+    ) -> Result<()> {
+        assert_not_in_callback();
+        let rpc = self.context.rpc();
+        let enable = device_changed_callback.is_some();
+        *self.device_change_cb.lock().unwrap() = device_changed_callback;
+        send_recv!(rpc, StreamRegisterDeviceChangeCallback(self.token, enable) => StreamRegisterDeviceChangeCallback)
+    }
+}
+
+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<Stream> {
+    let stm = ClientStream::init(ctx, init_params, data_callback, state_callback, user_ptr)?;
+    debug_assert_eq!(stm.user_ptr(), user_ptr);
+    Ok(stm)
+}
diff --git a/media/audioipc/gecko.patch b/media/audioipc/gecko.patch
new file mode 100644
index 0000000000..8a7d46532a
--- /dev/null
+++ b/media/audioipc/gecko.patch
@@ -0,0 +1,15 @@
+From 2c1d5189aedc7f90b603807a36a57254dc24471c Mon Sep 17 00:00:00 2001
+From: Dan Glastonbury <dglastonbury@mozilla.com>
+Date: Fri, 19 Jan 2018 11:56:58 +1000
+Subject: gecko: Change paths to vendored crates.
+
+
+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/media/audioipc/rustfmt.toml b/media/audioipc/rustfmt.toml
new file mode 100644
index 0000000000..9e4912fff7
--- /dev/null
+++ b/media/audioipc/rustfmt.toml
@@ -0,0 +1,10 @@
+ideal_width = 80
+match_block_trailing_comma = true
+max_width = 120
+newline_style = "Unix"
+normalize_comments = false
+struct_lit_multiline_style = "ForceMulti"
+where_trailing_comma = true
+reorder_imports = true
+reorder_imported_names = true
+trailing_comma = "Never"
\ No newline at end of file
diff --git a/media/audioipc/server/Cargo.toml b/media/audioipc/server/Cargo.toml
new file mode 100644
index 0000000000..6ae2877030
--- /dev/null
+++ b/media/audioipc/server/Cargo.toml
@@ -0,0 +1,23 @@
+[package]
+name = "audioipc-server"
+version = "0.2.3"
+authors = [
+        "Matthew Gregan <kinetik@flim.org>",
+        "Dan Glastonbury <dan.glastonbury@gmail.com>"
+        ]
+description = "Remote cubeb server"
+edition = "2018"
+
+[dependencies]
+audio_thread_priority = "0.23.4"
+audioipc = { path = "../audioipc" }
+cubeb-core = "0.8.0"
+futures = "0.1.29"
+once_cell = "1.2.0"
+log = "0.4"
+slab = "0.4"
+tokio = "0.1"
+
+[dependencies.error-chain]
+version = "0.11.0"
+default-features = false
diff --git a/media/audioipc/server/cbindgen.toml b/media/audioipc/server/cbindgen.toml
new file mode 100644
index 0000000000..9a7c204fe4
--- /dev/null
+++ b/media/audioipc/server/cbindgen.toml
@@ -0,0 +1,28 @@
+header = """/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/. */"""
+autogen_warning = """/* DO NOT MODIFY THIS MANUALLY! This file was generated using cbindgen. */"""
+include_version = true
+braces = "SameLine"
+line_length = 100
+tab_width = 2
+language = "C++"
+namespaces = ["mozilla", "audioipc"]
+
+[export]
+item_types = ["globals", "enums", "structs", "unions", "typedefs", "opaque", "functions", "constants"]
+
+[parse]
+parse_deps = true
+include = ['audioipc']
+
+[fn]
+args = "Vertical"
+rename_args = "GeckoCase"
+
+[struct]
+rename_fields = "GeckoCase"
+
+[defines]
+"windows" = "XP_WIN"
+"unix" = "XP_UNIX"
diff --git a/media/audioipc/server/src/lib.rs b/media/audioipc/server/src/lib.rs
new file mode 100644
index 0000000000..4d49c8fc08
--- /dev/null
+++ b/media/audioipc/server/src/lib.rs
@@ -0,0 +1,166 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+#![warn(unused_extern_crates)]
+
+#[macro_use]
+extern crate error_chain;
+#[macro_use]
+extern crate log;
+
+use audio_thread_priority::promote_current_thread_to_real_time;
+use audioipc::core;
+use audioipc::platformhandle_passing::framed_with_platformhandles;
+use audioipc::rpc;
+use audioipc::{MessageStream, PlatformHandle, PlatformHandleType};
+use futures::sync::oneshot;
+use futures::Future;
+use once_cell::sync::Lazy;
+use std::ffi::{CStr, CString};
+use std::os::raw::c_void;
+use std::ptr;
+use std::sync::Mutex;
+use tokio::reactor;
+
+mod server;
+
+struct CubebContextParams {
+    context_name: CString,
+    backend_name: Option<CString>,
+}
+
+static G_CUBEB_CONTEXT_PARAMS: Lazy<Mutex<CubebContextParams>> = Lazy::new(|| {
+    Mutex::new(CubebContextParams {
+        context_name: CString::new("AudioIPC Server").unwrap(),
+        backend_name: None,
+    })
+});
+
+#[allow(deprecated)]
+pub mod errors {
+    error_chain! {
+        links {
+            AudioIPC(::audioipc::errors::Error, ::audioipc::errors::ErrorKind);
+        }
+        foreign_links {
+            Cubeb(cubeb_core::Error);
+            Io(::std::io::Error);
+            Canceled(::futures::sync::oneshot::Canceled);
+        }
+    }
+}
+
+use crate::errors::*;
+
+struct ServerWrapper {
+    core_thread: core::CoreThread,
+    callback_thread: core::CoreThread,
+}
+
+fn run() -> Result<ServerWrapper> {
+    trace!("Starting up cubeb audio server event loop thread...");
+
+    let callback_thread = core::spawn_thread(
+        "AudioIPC Callback RPC",
+        || {
+            match promote_current_thread_to_real_time(0, 48000) {
+                Ok(_) => {}
+                Err(_) => {
+                    debug!("Failed to promote audio callback thread to real-time.");
+                }
+            }
+            trace!("Starting up cubeb audio callback event loop thread...");
+            Ok(())
+        },
+        || {},
+    )
+    .or_else(|e| {
+        debug!(
+            "Failed to start cubeb audio callback event loop thread: {:?}",
+            e
+        );
+        Err(e)
+    })?;
+
+    let core_thread = core::spawn_thread(
+        "AudioIPC Server RPC",
+        move || {
+            audioipc::server_platform_init();
+            Ok(())
+        },
+        || {},
+    )
+    .or_else(|e| {
+        debug!("Failed to cubeb audio core event loop thread: {:?}", e);
+        Err(e)
+    })?;
+
+    Ok(ServerWrapper {
+        core_thread,
+        callback_thread,
+    })
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn audioipc_server_start(
+    context_name: *const std::os::raw::c_char,
+    backend_name: *const std::os::raw::c_char,
+) -> *mut c_void {
+    let mut params = G_CUBEB_CONTEXT_PARAMS.lock().unwrap();
+    if !context_name.is_null() {
+        params.context_name = CStr::from_ptr(context_name).to_owned();
+    }
+    if !backend_name.is_null() {
+        let backend_string = CStr::from_ptr(backend_name).to_owned();
+        params.backend_name = Some(backend_string);
+    }
+    match run() {
+        Ok(server) => Box::into_raw(Box::new(server)) as *mut _,
+        Err(_) => ptr::null_mut() as *mut _,
+    }
+}
+
+#[no_mangle]
+pub extern "C" fn audioipc_server_new_client(p: *mut c_void) -> PlatformHandleType {
+    let (wait_tx, wait_rx) = oneshot::channel();
+    let wrapper: &ServerWrapper = unsafe { &*(p as *mut _) };
+
+    let core_handle = wrapper.callback_thread.handle();
+
+    // We create a connected pair of anonymous IPC endpoints. One side
+    // is registered with the reactor core, the other side is returned
+    // to the caller.
+    MessageStream::anonymous_ipc_pair()
+        .and_then(|(ipc_server, ipc_client)| {
+            // Spawn closure to run on same thread as reactor::Core
+            // via remote handle.
+            wrapper
+                .core_thread
+                .handle()
+                .spawn(futures::future::lazy(|| {
+                    trace!("Incoming connection");
+                    let handle = reactor::Handle::default();
+                    ipc_server.into_tokio_ipc(&handle)
+                    .and_then(|sock| {
+                        let transport = framed_with_platformhandles(sock, Default::default());
+                        rpc::bind_server(transport, server::CubebServer::new(core_handle));
+                        Ok(())
+                    }).map_err(|_| ())
+                    // Notify waiting thread that server has been registered.
+                    .and_then(|_| wait_tx.send(()))
+                }))
+                .expect("Failed to spawn CubebServer");
+            // Wait for notification that server has been registered
+            // with reactor::Core.
+            let _ = wait_rx.wait();
+            Ok(unsafe { PlatformHandle::from(ipc_client).into_raw() })
+        })
+        .unwrap_or(audioipc::INVALID_HANDLE_VALUE)
+}
+
+#[no_mangle]
+pub extern "C" fn audioipc_server_stop(p: *mut c_void) {
+    let wrapper = unsafe { Box::<ServerWrapper>::from_raw(p as *mut _) };
+    drop(wrapper);
+}
diff --git a/media/audioipc/server/src/server.rs b/media/audioipc/server/src/server.rs
new file mode 100644
index 0000000000..c2d4f31a02
--- /dev/null
+++ b/media/audioipc/server/src/server.rs
@@ -0,0 +1,862 @@
+// Copyright © 2017 Mozilla Foundation
+//
+// This program is made available under an ISC-style license.  See the
+// accompanying file LICENSE for details
+
+#[cfg(target_os = "linux")]
+use audio_thread_priority::{promote_thread_to_real_time, RtPriorityThreadInfo};
+use audioipc;
+use audioipc::codec::LengthDelimitedCodec;
+use audioipc::frame::{framed, Framed};
+use audioipc::messages::{
+    CallbackReq, CallbackResp, ClientMessage, Device, DeviceCollectionReq, DeviceCollectionResp,
+    DeviceInfo, RegisterDeviceCollectionChanged, ServerMessage, StreamCreate, StreamCreateParams,
+    StreamInitParams, StreamParams,
+};
+use audioipc::platformhandle_passing::FramedWithPlatformHandles;
+use audioipc::rpc;
+use audioipc::shm::{SharedMemReader, SharedMemWriter};
+use audioipc::{MessageStream, PlatformHandle};
+use cubeb_core as cubeb;
+use cubeb_core::ffi;
+use futures::future::{self, FutureResult};
+use futures::sync::oneshot;
+use futures::Future;
+use slab;
+use std::convert::From;
+use std::ffi::CStr;
+use std::mem::size_of;
+use std::os::raw::{c_long, c_void};
+use std::rc::Rc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::{cell::RefCell, sync::Mutex};
+use std::{panic, slice};
+use tokio::reactor;
+use tokio::runtime::current_thread;
+
+use crate::errors::*;
+
+fn error(error: cubeb::Error) -> ClientMessage {
+    ClientMessage::Error(error.raw_code())
+}
+
+struct CubebDeviceCollectionManager {
+    servers: Mutex<Vec<Rc<RefCell<CubebServerCallbacks>>>>,
+}
+
+impl CubebDeviceCollectionManager {
+    fn new() -> CubebDeviceCollectionManager {
+        CubebDeviceCollectionManager {
+            servers: Mutex::new(Vec::new()),
+        }
+    }
+
+    fn register(
+        &mut self,
+        context: &cubeb::Context,
+        server: &Rc<RefCell<CubebServerCallbacks>>,
+        devtype: cubeb::DeviceType,
+    ) -> cubeb::Result<()> {
+        let mut servers = self.servers.lock().unwrap();
+        if servers.is_empty() {
+            self.internal_register(context, true)?;
+        }
+        server.borrow_mut().devtype.insert(devtype);
+        if servers.iter().find(|s| Rc::ptr_eq(s, server)).is_none() {
+            servers.push(server.clone());
+        }
+        Ok(())
+    }
+
+    fn unregister(
+        &mut self,
+        context: &cubeb::Context,
+        server: &Rc<RefCell<CubebServerCallbacks>>,
+        devtype: cubeb::DeviceType,
+    ) -> cubeb::Result<()> {
+        let mut servers = self.servers.lock().unwrap();
+        server.borrow_mut().devtype.remove(devtype);
+        if server.borrow().devtype.is_empty() {
+            servers.retain(|s| !Rc::ptr_eq(&s, server));
+        }
+        if servers.is_empty() {
+            self.internal_register(context, false)?;
+        }
+        Ok(())
+    }
+
+    fn internal_register(&self, context: &cubeb::Context, enable: bool) -> cubeb::Result<()> {
+        let user_ptr = if enable {
+            self as *const CubebDeviceCollectionManager as *mut c_void
+        } else {
+            std::ptr::null_mut()
+        };
+        for &(dir, cb) in &[
+            (
+                cubeb::DeviceType::INPUT,
+                device_collection_changed_input_cb_c as _,
+            ),
+            (
+                cubeb::DeviceType::OUTPUT,
+                device_collection_changed_output_cb_c as _,
+            ),
+        ] {
+            unsafe {
+                context.register_device_collection_changed(
+                    dir,
+                    if enable { Some(cb) } else { None },
+                    user_ptr,
+                )?;
+            }
+        }
+        Ok(())
+    }
+
+    // Warning: this is called from an internal cubeb thread, so we must not mutate unprotected shared state.
+    unsafe fn device_collection_changed_callback(&self, device_type: ffi::cubeb_device_type) {
+        let servers = self.servers.lock().unwrap();
+        servers.iter().for_each(|server| {
+            if server
+                .borrow()
+                .devtype
+                .contains(cubeb::DeviceType::from_bits_truncate(device_type))
+            {
+                server
+                    .borrow_mut()
+                    .device_collection_changed_callback(device_type)
+            }
+        });
+    }
+}
+
+struct DevIdMap {
+    devices: Vec<usize>,
+}
+
+// A cubeb_devid is an opaque type which may be implemented with a stable
+// pointer in a cubeb backend.  cubeb_devids received remotely must be
+// validated before use, so DevIdMap provides a simple 1:1 mapping between a
+// cubeb_devid and an IPC-transportable value suitable for use as a unique
+// handle.
+impl DevIdMap {
+    fn new() -> DevIdMap {
+        let mut d = DevIdMap {
+            devices: Vec::with_capacity(32),
+        };
+        // A null cubeb_devid is used for selecting the default device.
+        // Pre-populate the mapping with 0 -> 0 to handle nulls.
+        d.devices.push(0);
+        d
+    }
+
+    // Given a cubeb_devid, return a unique stable value suitable for use
+    // over IPC.
+    fn to_handle(&mut self, devid: usize) -> usize {
+        if let Some(i) = self.devices.iter().position(|&d| d == devid) {
+            return i;
+        }
+        self.devices.push(devid);
+        self.devices.len() - 1
+    }
+
+    // Given a handle produced by `to_handle`, return the associated
+    // cubeb_devid.  Invalid handles result in a panic.
+    fn from_handle(&self, handle: usize) -> usize {
+        self.devices[handle]
+    }
+}
+
+struct CubebContextState {
+    context: cubeb::Result<cubeb::Context>,
+    manager: CubebDeviceCollectionManager,
+}
+
+thread_local!(static CONTEXT_KEY: RefCell<Option<CubebContextState>> = RefCell::new(None));
+
+fn cubeb_init_from_context_params() -> cubeb::Result<cubeb::Context> {
+    let params = super::G_CUBEB_CONTEXT_PARAMS.lock().unwrap();
+    let context_name = Some(params.context_name.as_c_str());
+    let backend_name = if let Some(ref name) = params.backend_name {
+        Some(name.as_c_str())
+    } else {
+        None
+    };
+    let r = cubeb::Context::init(context_name, backend_name);
+    r.map_err(|e| {
+        info!("cubeb::Context::init failed r={:?}", e);
+        e
+    })
+}
+
+fn with_local_context<T, F>(f: F) -> T
+where
+    F: FnOnce(&cubeb::Result<cubeb::Context>, &mut CubebDeviceCollectionManager) -> T,
+{
+    CONTEXT_KEY.with(|k| {
+        let mut state = k.borrow_mut();
+        if state.is_none() {
+            *state = Some(CubebContextState {
+                context: cubeb_init_from_context_params(),
+                manager: CubebDeviceCollectionManager::new(),
+            });
+        }
+        let CubebContextState { context, manager } = state.as_mut().unwrap();
+        // Always reattempt to initialize cubeb, OS config may have changed.
+        if context.is_err() {
+            *context = cubeb_init_from_context_params();
+        }
+        f(context, manager)
+    })
+}
+
+struct DeviceCollectionClient;
+
+impl rpc::Client for DeviceCollectionClient {
+    type Request = DeviceCollectionReq;
+    type Response = DeviceCollectionResp;
+    type Transport =
+        Framed<audioipc::AsyncMessageStream, LengthDelimitedCodec<Self::Request, Self::Response>>;
+}
+
+struct CallbackClient;
+
+impl rpc::Client for CallbackClient {
+    type Request = CallbackReq;
+    type Response = CallbackResp;
+    type Transport =
+        Framed<audioipc::AsyncMessageStream, LengthDelimitedCodec<Self::Request, Self::Response>>;
+}
+
+struct ServerStreamCallbacks {
+    /// Size of input frame in bytes
+    input_frame_size: u16,
+    /// Size of output frame in bytes
+    output_frame_size: u16,
+    /// Shared memory buffer for sending input data to client
+    input_shm: SharedMemWriter,
+    /// Shared memory buffer for receiving output data from client
+    output_shm: SharedMemReader,
+    /// RPC interface to callback server running in client
+    rpc: rpc::ClientProxy<CallbackReq, CallbackResp>,
+}
+
+impl ServerStreamCallbacks {
+    fn data_callback(&mut self, input: &[u8], output: &mut [u8], nframes: isize) -> isize {
+        trace!(
+            "Stream data callback: {} {} {}",
+            nframes,
+            input.len(),
+            output.len()
+        );
+
+        self.input_shm.write(input).unwrap();
+
+        let r = self
+            .rpc
+            .call(CallbackReq::Data {
+                nframes,
+                input_frame_size: self.input_frame_size as usize,
+                output_frame_size: self.output_frame_size as usize,
+            })
+            .wait();
+
+        match r {
+            Ok(CallbackResp::Data(frames)) => {
+                if frames >= 0 {
+                    let nbytes = frames as usize * self.output_frame_size as usize;
+                    trace!("Reslice output to {}", nbytes);
+                    self.output_shm.read(&mut output[..nbytes]).unwrap();
+                }
+                frames
+            }
+            _ => {
+                debug!("Unexpected message {:?} during data_callback", r);
+                // TODO: Return a CUBEB_ERROR result here once
+                // https://github.com/kinetiknz/cubeb/issues/553 is
+                // fixed.
+                0
+            }
+        }
+    }
+
+    fn state_callback(&mut self, state: cubeb::State) {
+        trace!("Stream state callback: {:?}", state);
+        let r = self.rpc.call(CallbackReq::State(state.into())).wait();
+        match r {
+            Ok(CallbackResp::State) => {}
+            _ => {
+                debug!("Unexpected message {:?} during state callback", r);
+            }
+        }
+    }
+
+    fn device_change_callback(&mut self) {
+        trace!("Stream device change callback");
+        let r = self.rpc.call(CallbackReq::DeviceChange).wait();
+        match r {
+            Ok(CallbackResp::DeviceChange) => {}
+            _ => {
+                debug!("Unexpected message {:?} during device change callback", r);
+            }
+        }
+    }
+}
+
+static SHM_ID: AtomicUsize = AtomicUsize::new(0);
+
+// Generate a temporary shm_id fragment that is unique to the process.  This
+// path is used temporarily to create a shm segment, which is then
+// immediately deleted from the filesystem while retaining handles to the
+// shm to be shared between the server and client.
+fn get_shm_id() -> String {
+    format!(
+        "cubeb-shm-{}-{}",
+        std::process::id(),
+        SHM_ID.fetch_add(1, Ordering::SeqCst)
+    )
+}
+
+struct ServerStream {
+    stream: Option<cubeb::Stream>,
+    cbs: Box<ServerStreamCallbacks>,
+}
+
+impl Drop for ServerStream {
+    fn drop(&mut self) {
+        // `stream` *must* be dropped before `cbs`.
+        drop(self.stream.take());
+    }
+}
+
+type StreamSlab = slab::Slab<ServerStream>;
+
+struct CubebServerCallbacks {
+    rpc: rpc::ClientProxy<DeviceCollectionReq, DeviceCollectionResp>,
+    devtype: cubeb::DeviceType,
+}
+
+impl CubebServerCallbacks {
+    fn device_collection_changed_callback(&mut self, device_type: ffi::cubeb_device_type) {
+        // TODO: Assert device_type is in devtype.
+        debug!(
+            "Sending device collection ({:?}) changed event",
+            device_type
+        );
+        let _ = self
+            .rpc
+            .call(DeviceCollectionReq::DeviceChange(device_type))
+            .wait();
+    }
+}
+
+pub struct CubebServer {
+    handle: current_thread::Handle,
+    streams: StreamSlab,
+    remote_pid: Option<u32>,
+    cbs: Option<Rc<RefCell<CubebServerCallbacks>>>,
+    devidmap: DevIdMap,
+}
+
+impl rpc::Server for CubebServer {
+    type Request = ServerMessage;
+    type Response = ClientMessage;
+    type Future = FutureResult<Self::Response, ()>;
+    type Transport = FramedWithPlatformHandles<
+        audioipc::AsyncMessageStream,
+        LengthDelimitedCodec<Self::Response, Self::Request>,
+    >;
+
+    fn process(&mut self, req: Self::Request) -> Self::Future {
+        if let ServerMessage::ClientConnect(pid) = req {
+            self.remote_pid = Some(pid);
+        }
+        let resp = with_local_context(|context, manager| match *context {
+            Err(_) => error(cubeb::Error::error()),
+            Ok(ref context) => self.process_msg(context, manager, &req),
+        });
+        future::ok(resp)
+    }
+}
+
+// Debugging for BMO 1594216/1612044.
+macro_rules! try_stream {
+    ($self:expr, $stm_tok:expr) => {
+        if $self.streams.contains($stm_tok) {
+            $self.streams[$stm_tok]
+                .stream
+                .as_mut()
+                .expect("uninitialized stream")
+        } else {
+            error!(
+                "{}:{}:{} - Stream({}): invalid token",
+                file!(),
+                line!(),
+                column!(),
+                $stm_tok
+            );
+            return error(cubeb::Error::invalid_parameter());
+        }
+    };
+}
+
+impl CubebServer {
+    pub fn new(handle: current_thread::Handle) -> Self {
+        CubebServer {
+            handle,
+            streams: StreamSlab::new(),
+            remote_pid: None,
+            cbs: None,
+            devidmap: DevIdMap::new(),
+        }
+    }
+
+    // Process a request coming from the client.
+    fn process_msg(
+        &mut self,
+        context: &cubeb::Context,
+        manager: &mut CubebDeviceCollectionManager,
+        msg: &ServerMessage,
+    ) -> ClientMessage {
+        let resp: ClientMessage = match *msg {
+            ServerMessage::ClientConnect(_) => {
+                // remote_pid is set before cubeb initialization, just verify here.
+                assert!(self.remote_pid.is_some());
+                ClientMessage::ClientConnected
+            }
+
+            ServerMessage::ClientDisconnect => {
+                // TODO:
+                //self.connection.client_disconnect();
+                ClientMessage::ClientDisconnected
+            }
+
+            ServerMessage::ContextGetBackendId => {
+                ClientMessage::ContextBackendId(context.backend_id().to_string())
+            }
+
+            ServerMessage::ContextGetMaxChannelCount => context
+                .max_channel_count()
+                .map(ClientMessage::ContextMaxChannelCount)
+                .unwrap_or_else(error),
+
+            ServerMessage::ContextGetMinLatency(ref params) => {
+                let format = cubeb::SampleFormat::from(params.format);
+                let layout = cubeb::ChannelLayout::from(params.layout);
+
+                let params = cubeb::StreamParamsBuilder::new()
+                    .format(format)
+                    .rate(params.rate)
+                    .channels(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::ContextGetDeviceEnumeration(device_type) => context
+                .enumerate_devices(cubeb::DeviceType::from_bits_truncate(device_type))
+                .map(|devices| {
+                    let v: Vec<DeviceInfo> = devices
+                        .iter()
+                        .map(|i| {
+                            let mut tmp: DeviceInfo = i.as_ref().into();
+                            // Replace each cubeb_devid with a unique handle suitable for IPC.
+                            tmp.devid = self.devidmap.to_handle(tmp.devid);
+                            tmp
+                        })
+                        .collect();
+                    ClientMessage::ContextEnumeratedDevices(v)
+                })
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamCreate(ref params) => self
+                .process_stream_create(params)
+                .unwrap_or_else(|_| error(cubeb::Error::error())),
+
+            ServerMessage::StreamInit(stm_tok, ref params) => self
+                .process_stream_init(context, stm_tok, params)
+                .unwrap_or_else(|_| error(cubeb::Error::error())),
+
+            ServerMessage::StreamDestroy(stm_tok) => {
+                if self.streams.contains(stm_tok) {
+                    debug!("Unregistering stream {:?}", stm_tok);
+                    self.streams.remove(stm_tok);
+                } else {
+                    // Debugging for BMO 1594216/1612044.
+                    error!("StreamDestroy({}): invalid token", stm_tok);
+                    return error(cubeb::Error::invalid_parameter());
+                }
+                ClientMessage::StreamDestroyed
+            }
+
+            ServerMessage::StreamStart(stm_tok) => try_stream!(self, stm_tok)
+                .start()
+                .map(|_| ClientMessage::StreamStarted)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamStop(stm_tok) => try_stream!(self, stm_tok)
+                .stop()
+                .map(|_| ClientMessage::StreamStopped)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamResetDefaultDevice(stm_tok) => try_stream!(self, stm_tok)
+                .reset_default_device()
+                .map(|_| ClientMessage::StreamDefaultDeviceReset)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamGetPosition(stm_tok) => try_stream!(self, stm_tok)
+                .position()
+                .map(ClientMessage::StreamPosition)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamGetLatency(stm_tok) => try_stream!(self, stm_tok)
+                .latency()
+                .map(ClientMessage::StreamLatency)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamGetInputLatency(stm_tok) => try_stream!(self, stm_tok)
+                .input_latency()
+                .map(ClientMessage::StreamInputLatency)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamSetVolume(stm_tok, volume) => try_stream!(self, stm_tok)
+                .set_volume(volume)
+                .map(|_| ClientMessage::StreamVolumeSet)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamSetName(stm_tok, ref name) => try_stream!(self, stm_tok)
+                .set_name(name)
+                .map(|_| ClientMessage::StreamNameSet)
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamGetCurrentDevice(stm_tok) => try_stream!(self, stm_tok)
+                .current_device()
+                .map(|device| ClientMessage::StreamCurrentDevice(Device::from(device)))
+                .unwrap_or_else(error),
+
+            ServerMessage::StreamRegisterDeviceChangeCallback(stm_tok, enable) => {
+                try_stream!(self, stm_tok)
+                    .register_device_changed_callback(if enable {
+                        Some(device_change_cb_c)
+                    } else {
+                        None
+                    })
+                    .map(|_| ClientMessage::StreamRegisterDeviceChangeCallback)
+                    .unwrap_or_else(error)
+            }
+
+            ServerMessage::ContextSetupDeviceCollectionCallback => {
+                if let Ok((ipc_server, ipc_client)) = MessageStream::anonymous_ipc_pair() {
+                    debug!(
+                        "Created device collection RPC pair: {:?}-{:?}",
+                        ipc_server, ipc_client
+                    );
+
+                    // 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 `handle`.
+                    let (tx, rx) = oneshot::channel();
+                    self.handle
+                        .spawn(futures::future::lazy(move || {
+                            let handle = reactor::Handle::default();
+                            let stream = ipc_server.into_tokio_ipc(&handle).unwrap();
+                            let transport = framed(stream, Default::default());
+                            let rpc = rpc::bind_client::<DeviceCollectionClient>(transport);
+                            drop(tx.send(rpc));
+                            Ok(())
+                        }))
+                        .expect("Failed to spawn DeviceCollectionClient");
+
+                    // TODO: The lowest comms layer expects exactly 3 PlatformHandles, but we only
+                    // need one here.  Send some dummy handles over for the other side to discard.
+                    let (dummy1, dummy2) =
+                        MessageStream::anonymous_ipc_pair().expect("need dummy IPC pair");
+                    if let Ok(rpc) = rx.wait() {
+                        self.cbs = Some(Rc::new(RefCell::new(CubebServerCallbacks {
+                            rpc,
+                            devtype: cubeb::DeviceType::empty(),
+                        })));
+                        let fds = RegisterDeviceCollectionChanged {
+                            platform_handles: [
+                                PlatformHandle::from(ipc_client),
+                                PlatformHandle::from(dummy1),
+                                PlatformHandle::from(dummy2),
+                            ],
+                            target_pid: self.remote_pid.unwrap(),
+                        };
+
+                        ClientMessage::ContextSetupDeviceCollectionCallback(fds)
+                    } else {
+                        warn!("Failed to setup RPC client");
+                        error(cubeb::Error::error())
+                    }
+                } else {
+                    warn!("Failed to create RPC pair");
+                    error(cubeb::Error::error())
+                }
+            }
+
+            ServerMessage::ContextRegisterDeviceCollectionChanged(device_type, enable) => self
+                .process_register_device_collection_changed(
+                    context,
+                    manager,
+                    cubeb::DeviceType::from_bits_truncate(device_type),
+                    enable,
+                )
+                .unwrap_or_else(error),
+
+            #[cfg(target_os = "linux")]
+            ServerMessage::PromoteThreadToRealTime(thread_info) => {
+                let info = RtPriorityThreadInfo::deserialize(thread_info);
+                match promote_thread_to_real_time(info, 0, 48000) {
+                    Ok(_) => {
+                        info!("Promotion of content process thread to real-time OK");
+                    }
+                    Err(_) => {
+                        warn!("Promotion of content process thread to real-time error");
+                    }
+                }
+                ClientMessage::ThreadPromoted
+            }
+        };
+
+        trace!("process_msg: req={:?}, resp={:?}", msg, resp);
+
+        resp
+    }
+
+    fn process_register_device_collection_changed(
+        &mut self,
+        context: &cubeb::Context,
+        manager: &mut CubebDeviceCollectionManager,
+        devtype: cubeb::DeviceType,
+        enable: bool,
+    ) -> cubeb::Result<ClientMessage> {
+        if devtype == cubeb::DeviceType::UNKNOWN {
+            return Err(cubeb::Error::invalid_parameter());
+        }
+
+        assert!(self.cbs.is_some());
+        let cbs = self.cbs.as_ref().unwrap();
+
+        if enable {
+            manager.register(context, cbs, devtype)
+        } else {
+            manager.unregister(context, cbs, devtype)
+        }
+        .map(|_| ClientMessage::ContextRegisteredDeviceCollectionChanged)
+    }
+
+    // Stream create is special, so it's been separated from process_msg.
+    fn process_stream_create(&mut self, params: &StreamCreateParams) -> Result<ClientMessage> {
+        fn frame_size_in_bytes(params: Option<&StreamParams>) -> u16 {
+            params
+                .map(|p| {
+                    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,
+                    };
+                    let channel_count = p.channels as u16;
+                    sample_size * channel_count
+                })
+                .unwrap_or(0u16)
+        }
+
+        // Create the callback handling struct which is attached the cubeb stream.
+        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 (ipc_server, ipc_client) = MessageStream::anonymous_ipc_pair()?;
+        debug!("Created callback pair: {:?}-{:?}", ipc_server, ipc_client);
+        let shm_id = get_shm_id();
+        let (input_shm, input_file) =
+            SharedMemWriter::new(&format!("{}-input", shm_id), audioipc::SHM_AREA_SIZE)?;
+        let (output_shm, output_file) =
+            SharedMemReader::new(&format!("{}-output", shm_id), audioipc::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 `handle`.
+        let (tx, rx) = oneshot::channel();
+        self.handle
+            .spawn(futures::future::lazy(move || {
+                let handle = reactor::Handle::default();
+                let stream = ipc_server.into_tokio_ipc(&handle).unwrap();
+                let transport = framed(stream, Default::default());
+                let rpc = rpc::bind_client::<CallbackClient>(transport);
+                drop(tx.send(rpc));
+                Ok(())
+            }))
+            .expect("Failed to spawn CallbackClient");
+
+        let rpc: rpc::ClientProxy<CallbackReq, CallbackResp> = match rx.wait() {
+            Ok(rpc) => rpc,
+            Err(_) => bail!("Failed to create callback rpc."),
+        };
+
+        let cbs = Box::new(ServerStreamCallbacks {
+            input_frame_size,
+            output_frame_size,
+            input_shm,
+            output_shm,
+            rpc,
+        });
+
+        let entry = self.streams.vacant_entry();
+        let key = entry.key();
+        debug!("Registering stream {:?}", key);
+
+        entry.insert(ServerStream { stream: None, cbs });
+
+        Ok(ClientMessage::StreamCreated(StreamCreate {
+            token: key,
+            platform_handles: [
+                PlatformHandle::from(ipc_client),
+                PlatformHandle::from(input_file),
+                PlatformHandle::from(output_file),
+            ],
+            target_pid: self.remote_pid.unwrap(),
+        }))
+    }
+
+    // Stream init is special, so it's been separated from process_msg.
+    fn process_stream_init(
+        &mut self,
+        context: &cubeb::Context,
+        stm_tok: usize,
+        params: &StreamInitParams,
+    ) -> Result<ClientMessage> {
+        // Create cubeb stream from params
+        let stream_name = params
+            .stream_name
+            .as_ref()
+            .and_then(|name| CStr::from_bytes_with_nul(name).ok());
+
+        // Map IPC handle back to cubeb_devid.
+        let input_device = self.devidmap.from_handle(params.input_device) as *const _;
+        let input_stream_params = params.input_stream_params.as_ref().map(|isp| unsafe {
+            cubeb::StreamParamsRef::from_ptr(isp as *const StreamParams as *mut _)
+        });
+
+        // Map IPC handle back to cubeb_devid.
+        let output_device = self.devidmap.from_handle(params.output_device) as *const _;
+        let output_stream_params = params.output_stream_params.as_ref().map(|osp| unsafe {
+            cubeb::StreamParamsRef::from_ptr(osp as *const StreamParams as *mut _)
+        });
+
+        let latency = params.latency_frames;
+
+        let server_stream = &mut self.streams[stm_tok];
+        assert!(size_of::<Box<ServerStreamCallbacks>>() == size_of::<usize>());
+        let user_ptr = server_stream.cbs.as_ref() as *const ServerStreamCallbacks as *mut c_void;
+
+        let stream = unsafe {
+            let stream = context.stream_init(
+                stream_name,
+                input_device,
+                input_stream_params,
+                output_device,
+                output_stream_params,
+                latency,
+                Some(data_cb_c),
+                Some(state_cb_c),
+                user_ptr,
+            );
+            match stream {
+                Ok(stream) => stream,
+                Err(e) => return Err(e.into()), // XXX full teardown of ServerStream?
+            }
+        };
+
+        server_stream.stream = Some(stream);
+
+        Ok(ClientMessage::StreamInitialized)
+    }
+}
+
+// C callable callbacks
+unsafe extern "C" fn data_cb_c(
+    _: *mut ffi::cubeb_stream,
+    user_ptr: *mut c_void,
+    input_buffer: *const c_void,
+    output_buffer: *mut c_void,
+    nframes: c_long,
+) -> c_long {
+    let ok = panic::catch_unwind(|| {
+        let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks);
+        let input = if input_buffer.is_null() {
+            &[]
+        } else {
+            let nbytes = nframes * c_long::from(cbs.input_frame_size);
+            slice::from_raw_parts(input_buffer as *const u8, nbytes as usize)
+        };
+        let output: &mut [u8] = if output_buffer.is_null() {
+            &mut []
+        } else {
+            let nbytes = nframes * c_long::from(cbs.output_frame_size);
+            slice::from_raw_parts_mut(output_buffer as *mut u8, nbytes as usize)
+        };
+        cbs.data_callback(input, output, nframes as isize) as c_long
+    });
+    // TODO: Return a CUBEB_ERROR result here once
+    // https://github.com/kinetiknz/cubeb/issues/553 is fixed.
+    ok.unwrap_or(0)
+}
+
+unsafe extern "C" fn state_cb_c(
+    _: *mut ffi::cubeb_stream,
+    user_ptr: *mut c_void,
+    state: ffi::cubeb_state,
+) {
+    let ok = panic::catch_unwind(|| {
+        let state = cubeb::State::from(state);
+        let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks);
+        cbs.state_callback(state);
+    });
+    ok.expect("State callback panicked");
+}
+
+unsafe extern "C" fn device_change_cb_c(user_ptr: *mut c_void) {
+    let ok = panic::catch_unwind(|| {
+        let cbs = &mut *(user_ptr as *mut ServerStreamCallbacks);
+        cbs.device_change_callback();
+    });
+    ok.expect("Device change callback panicked");
+}
+
+unsafe extern "C" fn device_collection_changed_input_cb_c(
+    _: *mut ffi::cubeb,
+    user_ptr: *mut c_void,
+) {
+    let ok = panic::catch_unwind(|| {
+        let manager = &mut *(user_ptr as *mut CubebDeviceCollectionManager);
+        manager.device_collection_changed_callback(ffi::CUBEB_DEVICE_TYPE_INPUT);
+    });
+    ok.expect("Collection changed (input) callback panicked");
+}
+
+unsafe extern "C" fn device_collection_changed_output_cb_c(
+    _: *mut ffi::cubeb,
+    user_ptr: *mut c_void,
+) {
+    let ok = panic::catch_unwind(|| {
+        let manager = &mut *(user_ptr as *mut CubebDeviceCollectionManager);
+        manager.device_collection_changed_callback(ffi::CUBEB_DEVICE_TYPE_OUTPUT);
+    });
+    ok.expect("Collection changed (output) callback panicked");
+}
diff --git a/media/audioipc/update.sh b/media/audioipc/update.sh
new file mode 100644
index 0000000000..72c871bd4a
--- /dev/null
+++ b/media/audioipc/update.sh
@@ -0,0 +1,37 @@
+# Usage: sh update.sh <upstream_src_directory>
+set -e
+
+cp -p $1/README.md .
+cp -p $1/Cargo.toml .
+
+for crate in audioipc client server; do
+    rm -fr $crate
+    mkdir $crate
+    cp -pr $1/$crate/Cargo.toml $crate
+    [ -e $1/$crate/cbindgen.toml ] && cp -pr $1/$crate/cbindgen.toml $crate
+    cp -pr $1/$crate/src $crate
+done
+
+rm -f audioipc/src/cmsghdr.c
+
+if [ -d $1/.git ]; then
+  rev=$(cd $1 && git rev-parse --verify HEAD)
+  date=$(cd $1 && git show -s --format=%ci HEAD)
+  dirty=$(cd $1 && git diff-index --name-only HEAD)
+fi
+
+if [ -n "$rev" ]; then
+  version=$rev
+  if [ -n "$dirty" ]; then
+    version=$version-dirty
+    echo "WARNING: updating from a dirty git repository."
+  fi
+  sed -i.bak -e "/The git commit ID used was/ s/[0-9a-f]\{40\}\(-dirty\)\{0,1\} .\{1,100\}/$version ($date)/" README_MOZILLA
+  rm -f README_MOZILLA.bak
+else
+  echo "Remember to update README_MOZILLA with the version details."
+fi
+
+echo "Applying gecko.patch on top of $rev"
+patch -p3 < gecko.patch
+