Merge pull request #359 from mitchmindtree/remove-unknown-buffer-type

Remove `UnknownTypeBuffer` in favour of specifying sample type.

asio-sys/Cargo.toml

@@ -10,7 +10,7 @@ keywords = ["audio", "sound", "asio", "steinberg"]
 build = "build.rs"
 
 [target.'cfg(any(target_os = "windows"))'.build-dependencies]
-bindgen = "0.42.0"
+bindgen = "0.51.0"
 walkdir = "2"
 cc = "1.0.25"
 

asio-sys/src/bindings/mod.rs

@@ -319,9 +319,7 @@ impl Asio {
         // Make owned CString to send to load driver
         let driver_name_cstring = CString::new(driver_name)
             .expect("failed to create `CString` from driver name");
-        let mut driver_info = ai::ASIODriverInfo {
+        let mut driver_info = std::mem::MaybeUninit::<ai::ASIODriverInfo>::uninit();
-            _bindgen_opaque_blob: [0u32; 43],
-        };
 
         unsafe {
             // TODO: Check that a driver of the same name does not already exist?
@@ -329,7 +327,8 @@ impl Asio {
                 false => Err(LoadDriverError::LoadDriverFailed),
                 true => {
                     // Initialize ASIO.
-                    asio_result!(ai::ASIOInit(&mut driver_info))?;
+                    asio_result!(ai::ASIOInit(driver_info.as_mut_ptr()))?;
+                    let _driver_info = driver_info.assume_init();
                     let state = Mutex::new(DriverState::Initialized);
                     let name = driver_name.to_string();
                     let destroyed = false;

examples/beep.rs

@@ -1,56 +1,50 @@
 extern crate anyhow;
 extern crate cpal;
 
-use cpal::traits::{DeviceTrait, StreamTrait, HostTrait};
+use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
 
 fn main() -> Result<(), anyhow::Error> {
     let host = cpal::default_host();
-    let device = host.default_output_device().expect("failed to find a default output device");
+    let device = host
+        .default_output_device()
+        .expect("failed to find a default output device");
     let format = device.default_output_format()?;
     let sample_rate = format.sample_rate.0 as f32;
-    let channels = format.channels;
+    let channels = format.channels as usize;
-    let mut sample_clock = 0f32;
 
     // Produce a sinusoid of maximum amplitude.
+    let mut sample_clock = 0f32;
     let mut next_value = move || {
         sample_clock = (sample_clock + 1.0) % sample_rate;
         (sample_clock * 440.0 * 2.0 * 3.141592 / sample_rate).sin()
     };
 
-    let stream = device.build_output_stream(&format, move |data| {
+    let err_fn = |err| eprintln!("an error occurred on stream: {}", err);
-        match data {
+
-            cpal::StreamData::Output { buffer: cpal::UnknownTypeOutputBuffer::U16(mut buffer) } => {
+    let data_fn = move |data: &mut cpal::Data| match data.sample_format() {
-                for sample in buffer.chunks_mut(channels as usize) {
+        cpal::SampleFormat::F32 => write_data::<f32>(data, channels, &mut next_value),
-                    let value = ((next_value() * 0.5 + 0.5) * std::u16::MAX as f32) as u16;
+        cpal::SampleFormat::I16 => write_data::<i16>(data, channels, &mut next_value),
-                    for out in sample.iter_mut() {
+        cpal::SampleFormat::U16 => write_data::<u16>(data, channels, &mut next_value),
-                        *out = value;
+    };
-                    }
+
-                }
+    let stream = device.build_output_stream(&format, data_fn, err_fn)?;
-            },
+
-            cpal::StreamData::Output { buffer: cpal::UnknownTypeOutputBuffer::I16(mut buffer) } => {
-                for sample in buffer.chunks_mut(channels as usize) {
-                    let value = (next_value() * std::i16::MAX as f32) as i16;
-                    for out in sample.iter_mut() {
-                        *out = value;
-                    }
-                }
-            },
-            cpal::StreamData::Output { buffer: cpal::UnknownTypeOutputBuffer::F32(mut buffer) } => {
-                for sample in buffer.chunks_mut(channels as usize) {
-                    let value = next_value();
-                    for out in sample.iter_mut() {
-                        *out = value;
-                    }
-                }
-            },
-            _ => (),
-        }
-    }, move |err| {
-        eprintln!("an error occurred on stream: {}", err);
-    })?;
     stream.play()?;
 
     std::thread::sleep(std::time::Duration::from_millis(1000));
 
     Ok(())
 }
+
+fn write_data<T>(output: &mut cpal::Data, channels: usize, next_sample: &mut dyn FnMut() -> f32)
+where
+    T: cpal::Sample,
+{
+    let output = output.as_slice_mut::<T>().expect("unexpected sample type");
+    for frame in output.chunks_mut(channels) {
+        let value: T = cpal::Sample::from::<f32>(&next_sample());
+        for sample in frame.iter_mut() {
+            *sample = value;
+        }
+    }
+}

examples/feedback.rs

@@ -47,52 +47,40 @@ fn main() -> Result<(), anyhow::Error> {
         producer.push(0.0).unwrap();
     }
 
+    let input_data_fn = move |data: &cpal::Data| {
+        let mut output_fell_behind = false;
+        let data = data.as_slice::<f32>().expect("unexpected sample type");
+        for &sample in data {
+            if producer.push(sample).is_err() {
+                output_fell_behind = true;
+            }
+        }
+        if output_fell_behind {
+            eprintln!("output stream fell behind: try increasing latency");
+        }
+    };
+
+    let output_data_fn = move |data: &mut cpal::Data| {
+        let mut input_fell_behind = None;
+        let data = data.as_slice_mut::<f32>().expect("unexpected sample type");
+        for sample in data {
+            *sample = match consumer.pop() {
+                Ok(s) => s,
+                Err(err) => {
+                    input_fell_behind = Some(err);
+                    0.0
+                },
+            };
+        }
+        if let Some(err) = input_fell_behind {
+            eprintln!("input stream fell behind: {:?}: try increasing latency", err);
+        }
+    };
+
     // Build streams.
     println!("Attempting to build both streams with `{:?}`.", format);
-    let input_stream = input_device.build_input_stream(&format, move |data| {
+    let input_stream = input_device.build_input_stream(&format, input_data_fn, err_fn)?;
-        match data {
+    let output_stream = output_device.build_output_stream(&format, output_data_fn, err_fn)?;
-            cpal::StreamData::Input {
-                buffer: cpal::UnknownTypeInputBuffer::F32(buffer),
-            } => {
-                let mut output_fell_behind = false;
-                for &sample in buffer.iter() {
-                    if producer.push(sample).is_err() {
-                        output_fell_behind = true;
-                    }
-                }
-                if output_fell_behind {
-                    eprintln!("output stream fell behind: try increasing latency");
-                }
-            },
-            _ => panic!("Expected input with f32 data"),
-        }
-    }, move |err| {
-        eprintln!("an error occurred on input stream: {}", err);
-    })?;
-    let output_stream = output_device.build_output_stream(&format, move |data| {
-        match data {
-            cpal::StreamData::Output {
-                buffer: cpal::UnknownTypeOutputBuffer::F32(mut buffer),
-            } => {
-                let mut input_fell_behind = None;
-                for sample in buffer.iter_mut() {
-                    *sample = match consumer.pop() {
-                        Ok(s) => s,
-                        Err(err) => {
-                            input_fell_behind = Some(err);
-                            0.0
-                        },
-                    };
-                }
-                if let Some(err) = input_fell_behind {
-                    eprintln!("input stream fell behind: {:?}: try increasing latency", err);
-                }
-            },
-            _ => panic!("Expected output with f32 data"),
-        }
-    }, move |err| {
-        eprintln!("an error occurred on output stream: {}", err);
-    })?;
     println!("Successfully built streams.");
 
     // Play the streams.
@@ -111,3 +99,7 @@ fn main() -> Result<(), anyhow::Error> {
     println!("Done!");
     Ok(())
 }
+
+fn err_fn(err: cpal::StreamError) {
+    eprintln!("an error occurred on stream: {}", err);
+}

examples/record_wav.rs

@@ -7,6 +7,9 @@ extern crate cpal;
 extern crate hound;
 
 use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
+use std::sync::{Arc, Mutex};
+use std::fs::File;
+use std::io::BufWriter;
 
 fn main() -> Result<(), anyhow::Error> {
     // Use the default host for working with audio devices.
@@ -25,55 +28,28 @@ fn main() -> Result<(), anyhow::Error> {
     const PATH: &'static str = concat!(env!("CARGO_MANIFEST_DIR"), "/recorded.wav");
     let spec = wav_spec_from_format(&format);
     let writer = hound::WavWriter::create(PATH, spec)?;
-    let writer = std::sync::Arc::new(std::sync::Mutex::new(Some(writer)));
+    let writer = Arc::new(Mutex::new(Some(writer)));
 
     // A flag to indicate that recording is in progress.
     println!("Begin recording...");
 
     // Run the input stream on a separate thread.
     let writer_2 = writer.clone();
-    let stream = device.build_input_stream(&format, move |data| {
+
-        // Otherwise write to the wav writer.
+    let err_fn = move |err| {
-        match data {
-            cpal::StreamData::Input {
-                buffer: cpal::UnknownTypeInputBuffer::U16(buffer),
-            } => {
-                if let Ok(mut guard) = writer_2.try_lock() {
-                    if let Some(writer) = guard.as_mut() {
-                        for sample in buffer.iter() {
-                            let sample = cpal::Sample::to_i16(sample);
-                            writer.write_sample(sample).ok();
-                        }
-                    }
-                }
-            },
-            cpal::StreamData::Input {
-                buffer: cpal::UnknownTypeInputBuffer::I16(buffer),
-            } => {
-                if let Ok(mut guard) = writer_2.try_lock() {
-                    if let Some(writer) = guard.as_mut() {
-                        for &sample in buffer.iter() {
-                            writer.write_sample(sample).ok();
-                        }
-                    }
-                }
-            },
-            cpal::StreamData::Input {
-                buffer: cpal::UnknownTypeInputBuffer::F32(buffer),
-            } => {
-                if let Ok(mut guard) = writer_2.try_lock() {
-                    if let Some(writer) = guard.as_mut() {
-                        for &sample in buffer.iter() {
-                            writer.write_sample(sample).ok();
-                        }
-                    }
-                }
-            },
-            _ => (),
-        }
-    }, move |err| {
         eprintln!("an error occurred on stream: {}", err);
-    })?;
+    };
+
+    let data_fn = move |data: &cpal::Data| {
+        match data.sample_format() {
+            cpal::SampleFormat::F32 => write_input_data::<f32, f32>(data, &writer_2),
+            cpal::SampleFormat::I16 => write_input_data::<i16, i16>(data, &writer_2),
+            cpal::SampleFormat::U16 => write_input_data::<u16, i16>(data, &writer_2),
+        }
+    };
+
+    let stream = device.build_input_stream(&format, data_fn, err_fn)?;
+
     stream.play()?;
 
     // Let recording go for roughly three seconds.
@@ -100,3 +76,21 @@ fn wav_spec_from_format(format: &cpal::Format) -> hound::WavSpec {
         sample_format: sample_format(format.data_type),
     }
 }
+
+type WavWriterHandle = Arc<Mutex<Option<hound::WavWriter<BufWriter<File>>>>>;
+
+fn write_input_data<T, U>(input: &cpal::Data, writer: &WavWriterHandle)
+where
+    T: cpal::Sample,
+    U: cpal::Sample + hound::Sample,
+{
+    let input = input.as_slice::<T>().expect("unexpected sample format");
+    if let Ok(mut guard) = writer.try_lock() {
+        if let Some(writer) = guard.as_mut() {
+            for &sample in input.iter() {
+                let sample: U = cpal::Sample::from(&sample);
+                writer.write_sample(sample).ok();
+            }
+        }
+    }
+}

src/host/alsa/mod.rs

@@ -1,29 +1,28 @@
 extern crate alsa_sys as alsa;
 extern crate libc;
 
-use std::{cmp, ffi, io, mem, ptr};
+use crate::{
+    BackendSpecificError,
+    BuildStreamError,
+    ChannelCount,
+    Data,
+    DefaultFormatError,
+    DeviceNameError,
+    DevicesError,
+    Format,
+    PauseStreamError,
+    PlayStreamError,
+    SampleFormat,
+    SampleRate,
+    StreamError,
+    SupportedFormat,
+    SupportedFormatsError,
+};
+use std::{cmp, ffi, io, ptr};
 use std::sync::Arc;
 use std::thread::{self, JoinHandle};
 use std::vec::IntoIter as VecIntoIter;
-
-use BackendSpecificError;
-use BuildStreamError;
-use ChannelCount;
-use DefaultFormatError;
-use DeviceNameError;
-use DevicesError;
-use Format;
-use PauseStreamError;
-use PlayStreamError;
-use SampleFormat;
-use SampleRate;
-use StreamData;
-use StreamError;
-use SupportedFormat;
-use SupportedFormatsError;
 use traits::{DeviceTrait, HostTrait, StreamTrait};
-use UnknownTypeInputBuffer;
-use UnknownTypeOutputBuffer;
 
 pub use self::enumerate::{default_input_device, default_output_device, Devices};
 
@@ -89,12 +88,34 @@ impl DeviceTrait for Device {
         Device::default_output_format(self)
     }
 
-    fn build_input_stream<D, E>(&self, format: &Format, data_callback: D, error_callback: E) -> Result<Self::Stream, BuildStreamError> where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static {
+    fn build_input_stream<D, E>(
-        Ok(Stream::new(Arc::new(self.build_stream_inner(format, alsa::SND_PCM_STREAM_CAPTURE)?), data_callback, error_callback))
+        &self,
+        format: &Format,
+        data_callback: D,
+        error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
+    where
+        D: FnMut(&Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
+        let stream_inner = self.build_stream_inner(format, alsa::SND_PCM_STREAM_CAPTURE)?;
+        let stream = Stream::new_input(Arc::new(stream_inner), data_callback, error_callback);
+        Ok(stream)
     }
 
-    fn build_output_stream<D, E>(&self, format: &Format, data_callback: D, error_callback: E) -> Result<Self::Stream, BuildStreamError> where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static {
+    fn build_output_stream<D, E>(
-        Ok(Stream::new(Arc::new(self.build_stream_inner(format, alsa::SND_PCM_STREAM_PLAYBACK)?), data_callback, error_callback))
+        &self,
+        format: &Format,
+        data_callback: D,
+        error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
+    where
+        D: FnMut(&mut Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
+        let stream_inner = self.build_stream_inner(format, alsa::SND_PCM_STREAM_PLAYBACK)?;
+        let stream = Stream::new_output(Arc::new(stream_inner), data_callback, error_callback);
+        Ok(stream)
     }
 }
 
@@ -147,7 +168,11 @@ impl Drop for TriggerReceiver {
 pub struct Device(String);
 
 impl Device {
-    fn build_stream_inner(&self, format: &Format, stream_type: alsa::snd_pcm_stream_t) -> Result<StreamInner, BuildStreamError> {
+    fn build_stream_inner(
+        &self,
+        format: &Format,
+        stream_type: alsa::snd_pcm_stream_t,
+    ) -> Result<StreamInner, BuildStreamError> {
         let name = ffi::CString::new(self.0.clone()).expect("unable to clone device");
 
         let handle = unsafe {
@@ -510,6 +535,7 @@ unsafe impl Send for StreamInner {}
 
 unsafe impl Sync for StreamInner {}
 
+#[derive(Debug, Eq, PartialEq)]
 enum StreamType { Input, Output }
 
 pub struct Stream {
@@ -524,183 +550,289 @@ pub struct Stream {
     trigger: TriggerSender,
 }
 
-/// The inner body of the audio processing thread. Takes the polymorphic
+#[derive(Default)]
-/// callback to avoid generating too much generic code.
+struct StreamWorkerContext {
-fn stream_worker(rx: TriggerReceiver,
+    descriptors: Vec<libc::pollfd>,
-                 stream: &StreamInner,
+    buffer: Vec<u8>,
-                 data_callback: &mut (dyn FnMut(StreamData) + Send + 'static),
+}
-                 error_callback: &mut (dyn FnMut(StreamError) + Send + 'static)) {
-    let mut descriptors = Vec::new();
-    let mut buffer = Vec::new();
-    loop {
-        descriptors.clear();
-        // Add the self-pipe for signaling termination.
-        descriptors.push(libc::pollfd {
-            fd: rx.0,
-            events: libc::POLLIN,
-            revents: 0,
-        });
 
-        // Add ALSA polling fds.
+fn input_stream_worker(
-        descriptors.reserve(stream.num_descriptors);
+    rx: TriggerReceiver,
-        let len = descriptors.len();
+    stream: &StreamInner,
-        let filled = unsafe {
+    data_callback: &mut (dyn FnMut(&Data) + Send + 'static),
-            alsa::snd_pcm_poll_descriptors(
+    error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
+) {
+    let mut ctxt = StreamWorkerContext::default();
+    loop {
+        match poll_descriptors_and_prepare_buffer(&rx, stream, &mut ctxt, error_callback) {
+            PollDescriptorsFlow::Continue => continue,
+            PollDescriptorsFlow::Return => return,
+            PollDescriptorsFlow::Ready { available_frames, stream_type } => {
+                assert_eq!(
+                    stream_type,
+                    StreamType::Input,
+                    "expected input stream, but polling descriptors indicated output",
+                );
+                process_input(
+                    stream,
+                    &mut ctxt.buffer,
+                    available_frames,
+                    data_callback,
+                    error_callback,
+                );
+            }
+        }
+    }
+}
+
+fn output_stream_worker(
+    rx: TriggerReceiver,
+    stream: &StreamInner,
+    data_callback: &mut (dyn FnMut(&mut Data) + Send + 'static),
+    error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
+) {
+    let mut ctxt = StreamWorkerContext::default();
+    loop {
+        match poll_descriptors_and_prepare_buffer(&rx, stream, &mut ctxt, error_callback) {
+            PollDescriptorsFlow::Continue => continue,
+            PollDescriptorsFlow::Return => return,
+            PollDescriptorsFlow::Ready { available_frames, stream_type } => {
+                assert_eq!(
+                    stream_type,
+                    StreamType::Output,
+                    "expected output stream, but polling descriptors indicated input",
+                );
+                process_output(
+                    stream,
+                    &mut ctxt.buffer,
+                    available_frames,
+                    data_callback,
+                    error_callback,
+                );
+            }
+        }
+    }
+}
+
+enum PollDescriptorsFlow {
+    Continue,
+    Return,
+    Ready {
+        stream_type: StreamType,
+        available_frames: usize,
+    }
+}
+
+// This block is shared between both input and output stream worker functions.
+fn poll_descriptors_and_prepare_buffer(
+    rx: &TriggerReceiver,
+    stream: &StreamInner,
+    ctxt: &mut StreamWorkerContext,
+    error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
+) -> PollDescriptorsFlow {
+    let StreamWorkerContext {
+        ref mut descriptors,
+        ref mut buffer,
+    } = *ctxt;
+
+    descriptors.clear();
+
+    // Add the self-pipe for signaling termination.
+    descriptors.push(libc::pollfd {
+        fd: rx.0,
+        events: libc::POLLIN,
+        revents: 0,
+    });
+
+    // Add ALSA polling fds.
+    descriptors.reserve(stream.num_descriptors);
+    let len = descriptors.len();
+    let filled = unsafe {
+        alsa::snd_pcm_poll_descriptors(
+            stream.channel,
+            descriptors[len..].as_mut_ptr(),
+            stream.num_descriptors as libc::c_uint,
+        )
+    };
+    debug_assert_eq!(filled, stream.num_descriptors as libc::c_int);
+    unsafe {
+        descriptors.set_len(len + stream.num_descriptors);
+    }
+
+    let res = unsafe {
+        // Don't timeout, wait forever.
+        libc::poll(descriptors.as_mut_ptr(), descriptors.len() as libc::nfds_t, -1)
+    };
+    if res < 0 {
+        let description = format!("`libc::poll()` failed: {}", io::Error::last_os_error());
+        error_callback(BackendSpecificError { description }.into());
+        return PollDescriptorsFlow::Continue;
+    } else if res == 0 {
+        let description = String::from("`libc::poll()` spuriously returned");
+        error_callback(BackendSpecificError { description }.into());
+        return PollDescriptorsFlow::Continue;
+    }
+
+    if descriptors[0].revents != 0 {
+        // The stream has been requested to be destroyed.
+        rx.clear_pipe();
+        return PollDescriptorsFlow::Return;
+    }
+
+    let stream_type = match check_for_pollout_or_pollin(stream, descriptors[1..].as_mut_ptr()) {
+        Ok(Some(ty)) => ty,
+        Ok(None) => {
+            // Nothing to process, poll again
+            return PollDescriptorsFlow::Continue;
+        },
+        Err(err) => {
+            error_callback(err.into());
+            return PollDescriptorsFlow::Continue;
+        }
+    };
+    // Get the number of available samples for reading/writing.
+    let available_samples = match get_available_samples(stream) {
+        Ok(n) => n,
+        Err(err) => {
+            let description = format!("Failed to query the number of available samples: {}", err);
+            error_callback(BackendSpecificError { description }.into());
+            return PollDescriptorsFlow::Continue;
+        }
+    };
+
+    // Only go on if there is at least `stream.period_len` samples.
+    if available_samples < stream.period_len {
+        return PollDescriptorsFlow::Continue;
+    }
+
+    // Prepare the data buffer.
+    let buffer_size = stream.sample_format.sample_size() * available_samples;
+    buffer.resize(buffer_size, 0u8);
+    let available_frames = available_samples / stream.num_channels as usize;
+
+    PollDescriptorsFlow::Ready {
+        stream_type,
+        available_frames,
+    }
+}
+
+// Read input data from ALSA and deliver it to the user.
+fn process_input(
+    stream: &StreamInner,
+    buffer: &mut [u8],
+    available_frames: usize,
+    data_callback: &mut (dyn FnMut(&Data) + Send + 'static),
+    error_callback: &mut dyn FnMut(StreamError),
+) {
+    let result = unsafe {
+        alsa::snd_pcm_readi(
+            stream.channel,
+            buffer.as_mut_ptr() as *mut _,
+            available_frames as alsa::snd_pcm_uframes_t,
+        )
+    };
+    if let Err(err) = check_errors(result as _) {
+        let description = format!("`snd_pcm_readi` failed: {}", err);
+        error_callback(BackendSpecificError { description }.into());
+        return;
+    }
+    let sample_format = stream.sample_format;
+    let data = buffer.as_mut_ptr() as *mut ();
+    let len = buffer.len() / sample_format.sample_size();
+    let data = unsafe {
+        Data::from_parts(data, len, sample_format)
+    };
+    data_callback(&data);
+}
+
+// Request data from the user's function and write it via ALSA.
+//
+// Returns `true`
+fn process_output(
+    stream: &StreamInner,
+    buffer: &mut [u8],
+    available_frames: usize,
+    data_callback: &mut (dyn FnMut(&mut Data) + Send + 'static),
+    error_callback: &mut dyn FnMut(StreamError),
+) {
+    {
+        // We're now sure that we're ready to write data.
+        let sample_format = stream.sample_format;
+        let data = buffer.as_mut_ptr() as *mut ();
+        let len = buffer.len() / sample_format.sample_size();
+        let mut data = unsafe {
+            Data::from_parts(data, len, sample_format)
+        };
+        data_callback(&mut data);
+    }
+    loop {
+        let result = unsafe {
+            alsa::snd_pcm_writei(
                 stream.channel,
-                descriptors[len..].as_mut_ptr(),
+                buffer.as_ptr() as *const _,
-                stream.num_descriptors as libc::c_uint,
+                available_frames as alsa::snd_pcm_uframes_t,
             )
         };
-        debug_assert_eq!(filled, stream.num_descriptors as libc::c_int);
+        if result == -libc::EPIPE as i64 {
-        unsafe {
+            // buffer underrun
-            descriptors.set_len(len + stream.num_descriptors);
+            // TODO: Notify the user of this.
-        }
+            unsafe { alsa::snd_pcm_recover(stream.channel, result as i32, 0) };
-
+        } else if let Err(err) = check_errors(result as _) {
-        let res = unsafe {
+            let description = format!("`snd_pcm_writei` failed: {}", err);
-            // Don't timeout, wait forever.
-            libc::poll(descriptors.as_mut_ptr(), descriptors.len() as libc::nfds_t, -1)
-        };
-        if res < 0 {
-            let description = format!("`libc::poll()` failed: {}", io::Error::last_os_error());
             error_callback(BackendSpecificError { description }.into());
             continue;
-        } else if res == 0 {
+        } else if result as usize != available_frames {
-            let description = String::from("`libc::poll()` spuriously returned");
+            let description = format!(
+                "unexpected number of frames written: expected {}, \
+                            result {} (this should never happen)",
+                available_frames,
+                result,
+            );
             error_callback(BackendSpecificError { description }.into());
             continue;
-        }
+        } else {
-
+            break;
-        if descriptors[0].revents != 0 {
-            // The stream has been requested to be destroyed.
-            rx.clear_pipe();
-            return;
-        }
-
-        let stream_type = match check_for_pollout_or_pollin(stream, descriptors[1..].as_mut_ptr()) {
-            Ok(Some(ty)) => ty,
-            Ok(None) => {
-                // Nothing to process, poll again
-                continue;
-            },
-            Err(err) => {
-                error_callback(err.into());
-                continue;
-            }
-        };
-        // Get the number of available samples for reading/writing.
-        let available_samples = match get_available_samples(stream) {
-            Ok(n) => n,
-            Err(err) => {
-                let description = format!("Failed to query the number of available samples: {}", err);
-                error_callback(BackendSpecificError { description }.into());
-                continue;
-            }
-        };
-
-        // Only go on if there is at least `stream.period_len` samples.
-        if available_samples < stream.period_len {
-            continue;
-        }
-
-        // Prepare the data buffer.
-        let buffer_size = stream.sample_format.sample_size() * available_samples;
-        buffer.resize(buffer_size, 0u8);
-        let available_frames = available_samples / stream.num_channels as usize;
-
-        match stream_type {
-            StreamType::Input => {
-                let result = unsafe {
-                    alsa::snd_pcm_readi(
-                        stream.channel,
-                        buffer.as_mut_ptr() as *mut _,
-                        available_frames as alsa::snd_pcm_uframes_t,
-                    )
-                };
-                if let Err(err) = check_errors(result as _) {
-                    let description = format!("`snd_pcm_readi` failed: {}", err);
-                    error_callback(BackendSpecificError { description }.into());
-                    continue;
-                }
-
-                let input_buffer = match stream.sample_format {
-                    SampleFormat::I16 => UnknownTypeInputBuffer::I16(::InputBuffer {
-                        buffer: unsafe { cast_input_buffer(&mut buffer) },
-                    }),
-                    SampleFormat::U16 => UnknownTypeInputBuffer::U16(::InputBuffer {
-                        buffer: unsafe { cast_input_buffer(&mut buffer) },
-                    }),
-                    SampleFormat::F32 => UnknownTypeInputBuffer::F32(::InputBuffer {
-                        buffer: unsafe { cast_input_buffer(&mut buffer) },
-                    }),
-                };
-                let stream_data = StreamData::Input {
-                    buffer: input_buffer,
-                };
-                data_callback(stream_data);
-            },
-            StreamType::Output => {
-                {
-                    // We're now sure that we're ready to write data.
-                    let output_buffer = match stream.sample_format {
-                        SampleFormat::I16 => UnknownTypeOutputBuffer::I16(::OutputBuffer {
-                            buffer: unsafe { cast_output_buffer(&mut buffer) },
-                        }),
-                        SampleFormat::U16 => UnknownTypeOutputBuffer::U16(::OutputBuffer {
-                            buffer: unsafe { cast_output_buffer(&mut buffer) },
-                        }),
-                        SampleFormat::F32 => UnknownTypeOutputBuffer::F32(::OutputBuffer {
-                            buffer: unsafe { cast_output_buffer(&mut buffer) },
-                        }),
-                    };
-
-                    let stream_data = StreamData::Output {
-                        buffer: output_buffer,
-                    };
-                    data_callback(stream_data);
-                }
-                loop {
-                    let result = unsafe {
-                        alsa::snd_pcm_writei(
-                            stream.channel,
-                            buffer.as_ptr() as *const _,
-                            available_frames as alsa::snd_pcm_uframes_t,
-                        )
-                    };
-
-                    if result == -libc::EPIPE as i64 {
-                        // buffer underrun
-                        // TODO: Notify the user of this.
-                        unsafe { alsa::snd_pcm_recover(stream.channel, result as i32, 0) };
-                    } else if let Err(err) = check_errors(result as _) {
-                        let description = format!("`snd_pcm_writei` failed: {}", err);
-                        error_callback(BackendSpecificError { description }.into());
-                        continue;
-                    } else if result as usize != available_frames {
-                        let description = format!(
-                            "unexpected number of frames written: expected {}, \
-                                        result {} (this should never happen)",
-                            available_frames,
-                            result,
-                        );
-                        error_callback(BackendSpecificError { description }.into());
-                        continue;
-                    } else {
-                        break;
-                    }
-                }
-            },
         }
     }
 }
 
 impl Stream {
-    fn new<D, E>(inner: Arc<StreamInner>, mut data_callback: D, mut error_callback: E) -> Stream
+    fn new_input<D, E>(
-        where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static {
+        inner: Arc<StreamInner>,
+        mut data_callback: D,
+        mut error_callback: E,
+    ) -> Stream
+    where
+        D: FnMut(&Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
         let (tx, rx) = trigger();
         // Clone the handle for passing into worker thread.
         let stream = inner.clone();
         let thread = thread::spawn(move || {
-            stream_worker(rx, &*stream, &mut data_callback, &mut error_callback);
+            input_stream_worker(rx, &*stream, &mut data_callback, &mut error_callback);
+        });
+        Stream {
+            thread: Some(thread),
+            inner,
+            trigger: tx,
+        }
+    }
+
+    fn new_output<D, E>(
+        inner: Arc<StreamInner>,
+        mut data_callback: D,
+        mut error_callback: E,
+    ) -> Stream
+    where
+        D: FnMut(&mut Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
+        let (tx, rx) = trigger();
+        // Clone the handle for passing into worker thread.
+        let stream = inner.clone();
+        let thread = thread::spawn(move || {
+            output_stream_worker(rx, &*stream, &mut data_callback, &mut error_callback);
         });
         Stream {
             thread: Some(thread),
@@ -938,17 +1070,3 @@ fn check_errors(err: libc::c_int) -> Result<(), String> {
 
     Ok(())
 }
-
-/// Cast a byte slice into a (immutable) slice of desired type.
-/// Safety: it's up to the caller to ensure that the input slice has valid bit representations.
-unsafe fn cast_input_buffer<T>(v: &[u8]) -> &[T] {
-    debug_assert!(v.len() % std::mem::size_of::<T>() == 0);
-    std::slice::from_raw_parts(v.as_ptr() as *const T, v.len() / std::mem::size_of::<T>())
-}
-
-/// Cast a byte slice into a mutable slice of desired type.
-/// Safety: it's up to the caller to ensure that the input slice has valid bit representations.
-unsafe fn cast_output_buffer<T>(v: &mut [u8]) -> &mut [T] {
-    debug_assert!(v.len() % std::mem::size_of::<T>() == 0);
-    std::slice::from_raw_parts_mut(v.as_mut_ptr() as *mut T, v.len() / std::mem::size_of::<T>())
-}

src/host/asio/mod.rs

@@ -1,17 +1,17 @@
 extern crate asio_sys as sys;
 extern crate parking_lot;
 
-use {
+use crate::{
     BuildStreamError,
+    Data,
     DefaultFormatError,
     DeviceNameError,
     DevicesError,
     Format,
     PauseStreamError,
     PlayStreamError,
-    SupportedFormatsError,
-    StreamData,
     StreamError,
+    SupportedFormatsError,
 };
 use traits::{
     DeviceTrait,
@@ -89,16 +89,28 @@ impl DeviceTrait for Device {
         Device::default_output_format(self)
     }
 
-    fn build_input_stream<D, E>(&self, format: &Format, data_callback: D, error_callback: E) -> Result<Self::Stream, BuildStreamError>
+    fn build_input_stream<D, E>(
+        &self,
+        format: &Format,
+        data_callback: D,
+        error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
     where
-        D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static
+        D: FnMut(&Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static
     {
         Device::build_input_stream(self, format, data_callback, error_callback)
     }
 
-    fn build_output_stream<D, E>(&self, format: &Format, data_callback: D, error_callback: E) -> Result<Self::Stream, BuildStreamError>
+    fn build_output_stream<D, E>(
+        &self,
+        format: &Format,
+        data_callback: D,
+        error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
     where
-        D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static
+        D: FnMut(&mut Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static
     {
         Device::build_output_stream(self, format, data_callback, error_callback)
     }

src/host/asio/stream.rs

@@ -9,13 +9,12 @@ use std::sync::Arc;
 use super::parking_lot::Mutex;
 use BackendSpecificError;
 use BuildStreamError;
+use Data;
 use Format;
 use PauseStreamError;
 use PlayStreamError;
+use Sample;
 use SampleFormat;
-use StreamData;
-use UnknownTypeInputBuffer;
-use UnknownTypeOutputBuffer;
 use StreamError;
 
 /// Sample types whose constant silent value is known.
@@ -23,14 +22,11 @@ trait Silence {
     const SILENCE: Self;
 }
 
-/// Constraints on the interleaved sample buffer format required by the CPAL API.
-trait InterleavedSample: Clone + Copy + Silence {
-    fn unknown_type_input_buffer(&[Self]) -> UnknownTypeInputBuffer;
-    fn unknown_type_output_buffer(&mut [Self]) -> UnknownTypeOutputBuffer;
-}
-
 /// Constraints on the ASIO sample types.
-trait AsioSample: Clone + Copy + Silence + std::ops::Add<Self, Output = Self> {}
+trait AsioSample: Clone + Copy + Silence + std::ops::Add<Self, Output = Self> {
+    fn to_cpal_sample<T: Sample>(&self) -> T;
+    fn from_cpal_sample<T: Sample>(&T) -> Self;
+}
 
 // Used to keep track of whether or not the current current asio stream buffer requires
 // being silencing before summing audio.
@@ -68,8 +64,8 @@ impl Device {
         _error_callback: E,
     ) -> Result<Stream, BuildStreamError>
     where
-        D: FnMut(StreamData) + Send + 'static,
+        D: FnMut(&Data) + Send + 'static,
-        E: FnMut(StreamError) + Send + 'static
+        E: FnMut(StreamError) + Send + 'static,
     {
         let stream_type = self.driver.input_data_type().map_err(build_stream_err)?;
 
@@ -109,20 +105,18 @@ impl Device {
 
             /// 1. Write from the ASIO buffer to the interleaved CPAL buffer.
             /// 2. Deliver the CPAL buffer to the user callback.
-            unsafe fn process_input_callback<A, B, D, F, G>(
+            unsafe fn process_input_callback<A, B, D, F>(
                 callback: &mut D,
                 interleaved: &mut [u8],
                 asio_stream: &sys::AsioStream,
                 buffer_index: usize,
                 from_endianness: F,
-                to_cpal_sample: G,
             )
             where
                 A: AsioSample,
-                B: InterleavedSample,
+                B: Sample,
-                D: FnMut(StreamData) + Send + 'static,
+                D: FnMut(&Data) + Send + 'static,
                 F: Fn(A) -> A,
-                G: Fn(A) -> B,
             {
                 // 1. Write the ASIO channels to the CPAL buffer.
                 let interleaved: &mut [B] = cast_slice_mut(interleaved);
@@ -130,35 +124,34 @@ impl Device {
                 for ch_ix in 0..n_channels {
                     let asio_channel = asio_channel_slice::<A>(asio_stream, buffer_index, ch_ix);
                     for (frame, s_asio) in interleaved.chunks_mut(n_channels).zip(asio_channel) {
-                        frame[ch_ix] = to_cpal_sample(from_endianness(*s_asio));
+                        frame[ch_ix] = from_endianness(*s_asio).to_cpal_sample();
                     }
                 }
 
                 // 2. Deliver the interleaved buffer to the callback.
-                callback(
+                let data = interleaved.as_mut_ptr() as *mut ();
-                    StreamData::Input { buffer: B::unknown_type_input_buffer(interleaved) },
+                let len = interleaved.len();
-                );
+                let data = Data::from_parts(data, len, B::FORMAT);
+                callback(&data);
             }
 
             match (&stream_type, data_type) {
                 (&sys::AsioSampleType::ASIOSTInt16LSB, SampleFormat::I16) => {
-                    process_input_callback::<i16, i16, _, _, _>(
+                    process_input_callback::<i16, i16, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         asio_stream,
                         buffer_index as usize,
                         from_le,
-                        std::convert::identity::<i16>,
                     );
                 }
                 (&sys::AsioSampleType::ASIOSTInt16MSB, SampleFormat::I16) => {
-                    process_input_callback::<i16, i16, _, _, _>(
+                    process_input_callback::<i16, i16, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         asio_stream,
                         buffer_index as usize,
                         from_be,
-                        std::convert::identity::<i16>,
                     );
                 }
 
@@ -166,13 +159,12 @@ impl Device {
                 // trait for the `to_le` and `to_be` methods, but this does not support floats.
                 (&sys::AsioSampleType::ASIOSTFloat32LSB, SampleFormat::F32) |
                 (&sys::AsioSampleType::ASIOSTFloat32MSB, SampleFormat::F32) => {
-                    process_input_callback::<f32, f32, _, _, _>(
+                    process_input_callback::<f32, f32, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         asio_stream,
                         buffer_index as usize,
                         std::convert::identity::<f32>,
-                        std::convert::identity::<f32>,
                     );
                 }
 
@@ -180,36 +172,33 @@ impl Device {
                 // `process_output_callback` function above by removing the unnecessary sample
                 // conversion function.
                 (&sys::AsioSampleType::ASIOSTInt32LSB, SampleFormat::I16) => {
-                    process_input_callback::<i32, i16, _, _, _>(
+                    process_input_callback::<i32, i16, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         asio_stream,
                         buffer_index as usize,
                         from_le,
-                        |s| (s >> 16) as i16,
                     );
                 }
                 (&sys::AsioSampleType::ASIOSTInt32MSB, SampleFormat::I16) => {
-                    process_input_callback::<i32, i16, _, _, _>(
+                    process_input_callback::<i32, i16, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         asio_stream,
                         buffer_index as usize,
                         from_be,
-                        |s| (s >> 16) as i16,
                     );
                 }
                 // TODO: Handle endianness conversion for floats? We currently use the `PrimInt`
                 // trait for the `to_le` and `to_be` methods, but this does not support floats.
                 (&sys::AsioSampleType::ASIOSTFloat64LSB, SampleFormat::F32) |
                 (&sys::AsioSampleType::ASIOSTFloat64MSB, SampleFormat::F32) => {
-                    process_input_callback::<f64, f32, _, _, _>(
+                    process_input_callback::<f64, f32, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         asio_stream,
                         buffer_index as usize,
                         std::convert::identity::<f64>,
-                        |s| s as f32,
                     );
                 }
 
@@ -241,7 +230,7 @@ impl Device {
         _error_callback: E,
     ) -> Result<Stream, BuildStreamError>
     where
-        D: FnMut(StreamData) + Send + 'static,
+        D: FnMut(&mut Data) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
         let stream_type = self.driver.output_data_type().map_err(build_stream_err)?;
@@ -305,26 +294,26 @@ impl Device {
             /// 2. If required, silence the ASIO buffer.
             /// 3. Finally, write the interleaved data to the non-interleaved ASIO buffer,
             ///    performing endianness conversions as necessary.
-            unsafe fn process_output_callback<A, B, D, F, G>(
+            unsafe fn process_output_callback<A, B, D, F>(
                 callback: &mut D,
                 interleaved: &mut [u8],
                 silence_asio_buffer: bool,
                 asio_stream: &sys::AsioStream,
                 buffer_index: usize,
-                to_asio_sample: F,
+                to_endianness: F,
-                to_endianness: G,
             )
             where
-                A: InterleavedSample,
+                A: Sample,
                 B: AsioSample,
-                D: FnMut(StreamData) + Send + 'static,
+                D: FnMut(&mut Data) + Send + 'static,
-                F: Fn(A) -> B,
+                F: Fn(B) -> B,
-                G: Fn(B) -> B,
             {
                 // 1. Render interleaved buffer from callback.
                 let interleaved: &mut [A] = cast_slice_mut(interleaved);
-                let buffer = A::unknown_type_output_buffer(interleaved);
+                let data = interleaved.as_mut_ptr() as *mut ();
-                callback(StreamData::Output { buffer });
+                let len = interleaved.len();
+                let mut data = Data::from_parts(data, len, A::FORMAT);
+                callback(&mut data);
 
                 // 2. Silence ASIO channels if necessary.
                 let n_channels = interleaved.len() / asio_stream.buffer_size as usize;
@@ -341,31 +330,29 @@ impl Device {
                     let asio_channel =
                         asio_channel_slice_mut::<B>(asio_stream, buffer_index, ch_ix);
                     for (frame, s_asio) in interleaved.chunks(n_channels).zip(asio_channel) {
-                        *s_asio = *s_asio + to_endianness(to_asio_sample(frame[ch_ix]));
+                        *s_asio = *s_asio + to_endianness(B::from_cpal_sample(&frame[ch_ix]));
                     }
                 }
             }
 
             match (data_type, &stream_type) {
                 (SampleFormat::I16, &sys::AsioSampleType::ASIOSTInt16LSB) => {
-                    process_output_callback::<i16, i16, _, _, _>(
+                    process_output_callback::<i16, i16, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         silence,
                         asio_stream,
                         buffer_index as usize,
-                        std::convert::identity::<i16>,
                         to_le,
                     );
                 }
                 (SampleFormat::I16, &sys::AsioSampleType::ASIOSTInt16MSB) => {
-                    process_output_callback::<i16, i16, _, _, _>(
+                    process_output_callback::<i16, i16, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         silence,
                         asio_stream,
                         buffer_index as usize,
-                        std::convert::identity::<i16>,
                         to_be,
                     );
                 }
@@ -374,14 +361,13 @@ impl Device {
                 // trait for the `to_le` and `to_be` methods, but this does not support floats.
                 (SampleFormat::F32, &sys::AsioSampleType::ASIOSTFloat32LSB) |
                 (SampleFormat::F32, &sys::AsioSampleType::ASIOSTFloat32MSB) => {
-                    process_output_callback::<f32, f32, _, _, _>(
+                    process_output_callback::<f32, f32, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         silence,
                         asio_stream,
                         buffer_index as usize,
                         std::convert::identity::<f32>,
-                        std::convert::identity::<f32>,
                     );
                 }
 
@@ -389,24 +375,22 @@ impl Device {
                 // `process_output_callback` function above by removing the unnecessary sample
                 // conversion function.
                 (SampleFormat::I16, &sys::AsioSampleType::ASIOSTInt32LSB) => {
-                    process_output_callback::<i16, i32, _, _, _>(
+                    process_output_callback::<i16, i32, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         silence,
                         asio_stream,
                         buffer_index as usize,
-                        |s| (s as i32) << 16,
                         to_le,
                     );
                 }
                 (SampleFormat::I16, &sys::AsioSampleType::ASIOSTInt32MSB) => {
-                    process_output_callback::<i16, i32, _, _, _>(
+                    process_output_callback::<i16, i32, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         silence,
                         asio_stream,
                         buffer_index as usize,
-                        |s| (s as i32) << 16,
                         to_be,
                     );
                 }
@@ -414,13 +398,12 @@ impl Device {
                 // trait for the `to_le` and `to_be` methods, but this does not support floats.
                 (SampleFormat::F32, &sys::AsioSampleType::ASIOSTFloat64LSB) |
                 (SampleFormat::F32, &sys::AsioSampleType::ASIOSTFloat64MSB) => {
-                    process_output_callback::<f32, f64, _, _, _>(
+                    process_output_callback::<f32, f64, _, _>(
                         &mut data_callback,
                         &mut interleaved,
                         silence,
                         asio_stream,
                         buffer_index as usize,
-                        |s| s as f64,
                         std::convert::identity::<f64>,
                     );
                 }
@@ -549,33 +532,45 @@ impl Silence for f64 {
     const SILENCE: Self = 0.0;
 }
 
-impl InterleavedSample for i16 {
+impl AsioSample for i16 {
-    fn unknown_type_input_buffer(buffer: &[Self]) -> UnknownTypeInputBuffer {
+    fn to_cpal_sample<T: Sample>(&self) -> T {
-        UnknownTypeInputBuffer::I16(::InputBuffer { buffer })
+        T::from(self)
     }
-
+    fn from_cpal_sample<T: Sample>(t: &T) -> Self {
-    fn unknown_type_output_buffer(buffer: &mut [Self]) -> UnknownTypeOutputBuffer {
+        Sample::from(t)
-        UnknownTypeOutputBuffer::I16(::OutputBuffer { buffer })
     }
 }
 
-impl InterleavedSample for f32 {
+impl AsioSample for i32 {
-    fn unknown_type_input_buffer(buffer: &[Self]) -> UnknownTypeInputBuffer {
+    fn to_cpal_sample<T: Sample>(&self) -> T {
-        UnknownTypeInputBuffer::F32(::InputBuffer { buffer })
+        let s = (*self >> 16) as i16;
+        s.to_cpal_sample()
     }
-
+    fn from_cpal_sample<T: Sample>(t: &T) -> Self {
-    fn unknown_type_output_buffer(buffer: &mut [Self]) -> UnknownTypeOutputBuffer {
+        let s = i16::from_cpal_sample(t);
-        UnknownTypeOutputBuffer::F32(::OutputBuffer { buffer })
+        (s as i32) << 16
     }
 }
 
-impl AsioSample for i16 {}
+impl AsioSample for f32 {
+    fn to_cpal_sample<T: Sample>(&self) -> T {
+        T::from(self)
+    }
+    fn from_cpal_sample<T: Sample>(t: &T) -> Self {
+        Sample::from(t)
+    }
+}
 
-impl AsioSample for i32 {}
+impl AsioSample for f64 {
-
+    fn to_cpal_sample<T: Sample>(&self) -> T {
-impl AsioSample for f32 {}
+        let f = *self as f32;
-
+        f.to_cpal_sample()
-impl AsioSample for f64 {}
+    }
+    fn from_cpal_sample<T: Sample>(t: &T) -> Self {
+        let f = f32::from_cpal_sample(t);
+        f as f64
+    }
+}
 
 /// Check whether or not the desired format is supported by the stream.
 ///

src/host/coreaudio/mod.rs

@@ -1,35 +1,24 @@
 extern crate coreaudio;
 extern crate core_foundation_sys;
 
-use ChannelCount;
+use crate::{
-use BackendSpecificError;
+    ChannelCount,
-use BuildStreamError;
+    BackendSpecificError,
-use DefaultFormatError;
+    BuildStreamError,
-use DeviceNameError;
+    Data,
-use DevicesError;
+    DefaultFormatError,
-use Format;
+    DeviceNameError,
-use PauseStreamError;
+    DevicesError,
-use PlayStreamError;
+    Format,
-use SupportedFormatsError;
+    PauseStreamError,
-use SampleFormat;
+    PlayStreamError,
-use SampleRate;
+    SampleFormat,
-use StreamData;
+    SampleRate,
-use StreamError;
+    StreamError,
-use SupportedFormat;
+    SupportedFormat,
-use UnknownTypeInputBuffer;
+    SupportedFormatsError,
-use UnknownTypeOutputBuffer;
+};
-use traits::{DeviceTrait, HostTrait, StreamTrait};
+use crate::traits::{DeviceTrait, HostTrait, StreamTrait};
-
-use std::ffi::CStr;
-use std::fmt;
-use std::mem;
-use std::cell::RefCell;
-use std::os::raw::c_char;
-use std::ptr::null;
-use std::slice;
-use std::thread;
-use std::time::Duration;
-
 use self::coreaudio::audio_unit::{AudioUnit, Scope, Element};
 use self::coreaudio::audio_unit::render_callback::{self, data};
 use self::coreaudio::sys::{
@@ -69,6 +58,15 @@ use self::core_foundation_sys::string::{
     CFStringRef,
     CFStringGetCStringPtr,
 };
+use std::ffi::CStr;
+use std::fmt;
+use std::mem;
+use std::cell::RefCell;
+use std::os::raw::c_char;
+use std::ptr::null;
+use std::slice;
+use std::thread;
+use std::time::Duration;
 
 mod enumerate;
 
@@ -131,11 +129,29 @@ impl DeviceTrait for Device {
         Device::default_output_format(self)
     }
 
-    fn build_input_stream<D, E>(&self, format: &Format, data_callback: D, error_callback: E) -> Result<Self::Stream, BuildStreamError> where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static {
+    fn build_input_stream<D, E>(
+        &self,
+        format: &Format,
+        data_callback: D,
+        error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
+    where
+        D: FnMut(&Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
         Device::build_input_stream(self, format, data_callback, error_callback)
     }
 
-    fn build_output_stream<D, E>(&self, format: &Format, data_callback: D, error_callback: E) -> Result<Self::Stream, BuildStreamError> where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static {
+    fn build_output_stream<D, E>(
+        &self,
+        format: &Format,
+        data_callback: D,
+        error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
+    where
+        D: FnMut(&mut Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
         Device::build_output_stream(self, format, data_callback, error_callback)
     }
 }
@@ -478,7 +494,16 @@ fn audio_unit_from_device(device: &Device, input: bool) -> Result<AudioUnit, cor
 }
 
 impl Device {
-    fn build_input_stream<D, E>(&self, format: &Format, mut data_callback: D, _error_callback: E) -> Result<Stream, BuildStreamError> where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static {
+    fn build_input_stream<D, E>(
+        &self,
+        format: &Format,
+        mut data_callback: D,
+        _error_callback: E,
+    ) -> Result<Stream, BuildStreamError>
+    where
+        D: FnMut(&Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
         // The scope and element for working with a device's input stream.
         let scope = Scope::Output;
         let element = Element::Input;
@@ -626,7 +651,7 @@ impl Device {
         // Register the callback that is being called by coreaudio whenever it needs data to be
         // fed to the audio buffer.
         let sample_format = format.data_type;
-        let bytes_per_channel = format.data_type.sample_size();
+        let bytes_per_channel = sample_format.sample_size();
         type Args = render_callback::Args<data::Raw>;
         audio_unit.set_input_callback(move |args: Args| unsafe {
             let ptr = (*args.data.data).mBuffers.as_ptr() as *const AudioBuffer;
@@ -640,23 +665,10 @@ impl Device {
                 mData: data
             } = buffers[0];
 
-            // A small macro to simplify handling the callback for different sample types.
+            let data = data as *mut ();
-            macro_rules! try_callback {
+            let len = (data_byte_size as usize / bytes_per_channel) as usize;
-                ($SampleFormat:ident, $SampleType:ty) => {{
+            let data = Data::from_parts(data, len, sample_format);
-                    let data_len = (data_byte_size as usize / bytes_per_channel) as usize;
+            data_callback(&data);
-                    let data_slice = slice::from_raw_parts(data as *const $SampleType, data_len);
-                    let unknown_type_buffer = UnknownTypeInputBuffer::$SampleFormat(::InputBuffer { buffer: data_slice });
-                    let stream_data = StreamData::Input { buffer: unknown_type_buffer };
-                    data_callback(stream_data);
-                }};
-            }
-
-            match sample_format {
-                SampleFormat::F32 => try_callback!(F32, f32),
-                SampleFormat::I16 => try_callback!(I16, i16),
-                SampleFormat::U16 => try_callback!(U16, u16),
-            }
-
             Ok(())
         })?;
 
@@ -669,7 +681,16 @@ impl Device {
         }))
     }
 
-    fn build_output_stream<D, E>(&self, format: &Format, mut data_callback: D, _error_callback: E) -> Result<Stream, BuildStreamError> where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static {
+    fn build_output_stream<D, E>(
+        &self,
+        format: &Format,
+        mut data_callback: D,
+        _error_callback: E,
+    ) -> Result<Stream, BuildStreamError>
+    where
+        D: FnMut(&mut Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
         let mut audio_unit = audio_unit_from_device(self, false)?;
 
         // The scope and element for working with a device's output stream.
@@ -683,7 +704,7 @@ impl Device {
         // Register the callback that is being called by coreaudio whenever it needs data to be
         // fed to the audio buffer.
         let sample_format = format.data_type;
-        let bytes_per_channel = format.data_type.sample_size();
+        let bytes_per_channel = sample_format.sample_size();
         type Args = render_callback::Args<data::Raw>;
         audio_unit.set_render_callback(move |args: Args| unsafe {
             // If `run()` is currently running, then a callback will be available from this list.
@@ -695,23 +716,10 @@ impl Device {
                 mData: data
             } = (*args.data.data).mBuffers[0];
 
-            // A small macro to simplify handling the callback for different sample types.
+            let data = data as *mut ();
-            macro_rules! try_callback {
+            let len = (data_byte_size as usize / bytes_per_channel) as usize;
-                ($SampleFormat:ident, $SampleType:ty, $equilibrium:expr) => {{
+            let mut data = Data::from_parts(data, len, sample_format);
-                    let data_len = (data_byte_size as usize / bytes_per_channel) as usize;
+            data_callback(&mut data);
-                    let data_slice = slice::from_raw_parts_mut(data as *mut $SampleType, data_len);
-                    let unknown_type_buffer = UnknownTypeOutputBuffer::$SampleFormat(::OutputBuffer { buffer: data_slice });
-                    let stream_data = StreamData::Output { buffer: unknown_type_buffer };
-                    data_callback(stream_data);
-                }};
-            }
-
-            match sample_format {
-                SampleFormat::F32 => try_callback!(F32, f32, 0.0),
-                SampleFormat::I16 => try_callback!(I16, i16, 0),
-                SampleFormat::U16 => try_callback!(U16, u16, ::std::u16::MAX / 2),
-            }
-
             Ok(())
         })?;
 

src/host/emscripten/mod.rs

@@ -7,18 +7,20 @@ use stdweb::unstable::TryInto;
 use stdweb::web::TypedArray;
 use stdweb::web::set_timeout;
 
-use BuildStreamError;
+use crate::{
-use DefaultFormatError;
+    BuildStreamError,
-use DeviceNameError;
+    Data,
-use DevicesError;
+    DefaultFormatError,
-use Format;
+    DeviceNameError,
-use PauseStreamError;
+    DevicesError,
-use PlayStreamError;
+    Format,
-use SupportedFormatsError;
+    PauseStreamError,
-use StreamData;
+    PlayStreamError,
-use StreamError;
+    SampleFormat,
-use SupportedFormat;
+    StreamError,
-use UnknownTypeOutputBuffer;
+    SupportedFormat,
+    SupportedFormatsError,
+};
 use traits::{DeviceTrait, HostTrait, StreamTrait};
 
 // The emscripten backend currently works by instantiating an `AudioContext` object per `Stream`.
@@ -163,7 +165,7 @@ impl DeviceTrait for Device {
         _error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        D: FnMut(StreamData) + Send + 'static,
+        D: FnMut(&Data) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
         unimplemented!()
@@ -171,14 +173,20 @@ impl DeviceTrait for Device {
 
     fn build_output_stream<D, E>(
         &self,
-        _format: &Format,
+        format: &Format,
         data_callback: D,
         error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        D: FnMut(StreamData) + Send + 'static,
+        D: FnMut(&mut Data) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
+        assert_eq!(
+            format.data_type,
+            SampleFormat::F32,
+            "emscripten backend currently only supports `f32` data",
+        );
+
         // Create the stream.
         let audio_ctxt_ref = js!(return new AudioContext()).into_reference().unwrap();
         let stream = Stream { audio_ctxt_ref };
@@ -217,7 +225,7 @@ impl StreamTrait for Stream {
 // and to the `callback` parameter that was passed to `run`.
 fn audio_callback_fn<D, E>(user_data_ptr: *mut c_void)
 where
-    D: FnMut(StreamData) + Send + 'static,
+    D: FnMut(&mut Data) + Send + 'static,
     E: FnMut(StreamError) + Send + 'static,
 {
     unsafe {
@@ -230,9 +238,11 @@ where
         let mut temporary_buffer = vec![0.0; 44100 * 2 / 3];
 
         {
-            let buffer = UnknownTypeOutputBuffer::F32(::OutputBuffer { buffer: &mut temporary_buffer });
+            let len = temporary_buffer.len();
-            let data = StreamData::Output { buffer: buffer };
+            let data = temporary_buffer.as_mut_ptr() as *mut ();
-            data_cb(data);
+            let sample_format = SampleFormat::F32;
+            let mut data = Data::from_parts(data, len, sample_format);
+            data_cb(&mut data);
         }
 
         // TODO: directly use a TypedArray<f32> once this is supported by stdweb

src/host/null/mod.rs

@@ -1,14 +1,16 @@
-use BuildStreamError;
+use crate::{
-use DefaultFormatError;
+    BuildStreamError,
-use DevicesError;
+    Data,
-use DeviceNameError;
+    DefaultFormatError,
-use Format;
+    DevicesError,
-use PauseStreamError;
+    DeviceNameError,
-use PlayStreamError;
+    Format,
-use StreamData;
+    PauseStreamError,
-use StreamError;
+    PlayStreamError,
-use SupportedFormatsError;
+    StreamError,
-use SupportedFormat;
+    SupportedFormatsError,
+    SupportedFormat,
+};
 use traits::{DeviceTrait, HostTrait, StreamTrait};
 
 #[derive(Default)]
@@ -68,14 +70,30 @@ impl DeviceTrait for Device {
         unimplemented!()
     }
 
-    fn build_input_stream<D, E>(&self, _format: &Format, _data_callback: D, _error_callback: E) -> Result<Self::Stream, BuildStreamError>
+    fn build_input_stream<D, E>(
-        where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static {
+        &self,
+        _format: &Format,
+        _data_callback: D,
+        _error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
+    where
+        D: FnMut(&Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
         unimplemented!()
     }
 
     /// Create an output stream.
-    fn build_output_stream<D, E>(&self, _format: &Format, _data_callback: D, _error_callback: E) -> Result<Self::Stream, BuildStreamError>
+    fn build_output_stream<D, E>(
-        where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static{
+        &self,
+        _format: &Format,
+        _data_callback: D,
+        _error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
+    where
+        D: FnMut(&mut Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
         unimplemented!()
     }
 }

src/host/wasapi/device.rs

@@ -1,3 +1,16 @@
+use crate::{
+    BackendSpecificError,
+    Data,
+    DefaultFormatError,
+    DeviceNameError,
+    DevicesError,
+    Format,
+    SampleFormat,
+    SampleRate,
+    SupportedFormat,
+    SupportedFormatsError,
+    COMMON_SAMPLE_RATES,
+};
 use std;
 use std::ffi::OsString;
 use std::fmt;
@@ -9,17 +22,6 @@ use std::ptr;
 use std::slice;
 use std::sync::{Arc, Mutex, MutexGuard, atomic::Ordering};
 
-use BackendSpecificError;
-use DefaultFormatError;
-use DeviceNameError;
-use DevicesError;
-use Format;
-use SampleFormat;
-use SampleRate;
-use SupportedFormat;
-use SupportedFormatsError;
-use COMMON_SAMPLE_RATES;
-
 use super::check_result;
 use super::check_result_backend_specific;
 use super::com;
@@ -54,7 +56,7 @@ use super::{
     stream::{AudioClientFlow, Stream, StreamInner},
     winapi::um::synchapi,
 };
-use crate::{traits::DeviceTrait, BuildStreamError, StreamData, StreamError};
+use crate::{traits::DeviceTrait, BuildStreamError, StreamError};
 
 pub type SupportedInputFormats = std::vec::IntoIter<SupportedFormat>;
 pub type SupportedOutputFormats = std::vec::IntoIter<SupportedFormat>;
@@ -109,14 +111,11 @@ impl DeviceTrait for Device {
         error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        D: FnMut(StreamData) + Send + 'static,
+        D: FnMut(&Data) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
-        Ok(Stream::new(
+        let stream_inner = self.build_input_stream_inner(format)?;
-            self.build_input_stream_inner(format)?,
+        Ok(Stream::new_input(stream_inner, data_callback, error_callback))
-            data_callback,
-            error_callback,
-        ))
     }
 
     fn build_output_stream<D, E>(
@@ -126,14 +125,11 @@ impl DeviceTrait for Device {
         error_callback: E,
     ) -> Result<Self::Stream, BuildStreamError>
     where
-        D: FnMut(StreamData) + Send + 'static,
+        D: FnMut(&mut Data) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
-        Ok(Stream::new(
+        let stream_inner = self.build_output_stream_inner(format)?;
-            self.build_output_stream_inner(format)?,
+        Ok(Stream::new_output(stream_inner, data_callback, error_callback))
-            data_callback,
-            error_callback,
-        ))
     }
 }
 

src/host/wasapi/stream.rs

@@ -1,3 +1,16 @@
+use crate::{
+    BackendSpecificError,
+    Data,
+    PauseStreamError,
+    PlayStreamError,
+    SampleFormat,
+    StreamError,
+};
+use crate::traits::StreamTrait;
+use std::mem;
+use std::ptr;
+use std::sync::mpsc::{channel, Receiver, Sender};
+use std::thread::{self, JoinHandle};
 use super::check_result;
 use super::winapi::shared::basetsd::UINT32;
 use super::winapi::shared::minwindef::{BYTE, FALSE, WORD};
@@ -7,23 +20,6 @@ use super::winapi::um::synchapi;
 use super::winapi::um::winbase;
 use super::winapi::um::winnt;
 
-use std::mem;
-use std::ptr;
-use std::slice;
-use std::sync::mpsc::{channel, Receiver, Sender};
-
-use crate::traits::StreamTrait;
-use std::thread::{self, JoinHandle};
-
-use BackendSpecificError;
-use PauseStreamError;
-use PlayStreamError;
-use SampleFormat;
-use StreamData;
-use StreamError;
-use UnknownTypeInputBuffer;
-use UnknownTypeOutputBuffer;
-
 pub struct Stream {
     /// The high-priority audio processing thread calling callbacks.
     /// Option used for moving out in destructor.
@@ -86,13 +82,13 @@ pub struct StreamInner {
 }
 
 impl Stream {
-    pub(crate) fn new<D, E>(
+    pub(crate) fn new_input<D, E>(
         stream_inner: StreamInner,
         mut data_callback: D,
         mut error_callback: E,
     ) -> Stream
     where
-        D: FnMut(StreamData) + Send + 'static,
+        D: FnMut(&Data) + Send + 'static,
         E: FnMut(StreamError) + Send + 'static,
     {
         let pending_scheduled_event =
@@ -106,7 +102,36 @@ impl Stream {
         };
 
         let thread =
-            thread::spawn(move || run_inner(run_context, &mut data_callback, &mut error_callback));
+            thread::spawn(move || run_input(run_context, &mut data_callback, &mut error_callback));
+
+        Stream {
+            thread: Some(thread),
+            commands: tx,
+            pending_scheduled_event,
+        }
+    }
+
+    pub(crate) fn new_output<D, E>(
+        stream_inner: StreamInner,
+        mut data_callback: D,
+        mut error_callback: E,
+    ) -> Stream
+    where
+        D: FnMut(&mut Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
+        let pending_scheduled_event =
+            unsafe { synchapi::CreateEventA(ptr::null_mut(), 0, 0, ptr::null()) };
+        let (tx, rx) = channel();
+
+        let run_context = RunContext {
+            handles: vec![pending_scheduled_event, stream_inner.event],
+            stream: stream_inner,
+            commands: rx,
+        };
+
+        let thread =
+            thread::spawn(move || run_output(run_context, &mut data_callback, &mut error_callback));
 
         Stream {
             thread: Some(thread),
@@ -255,160 +280,185 @@ fn stream_error_from_hresult(hresult: winnt::HRESULT) -> Result<(), StreamError>
     Ok(())
 }
 
-fn run_inner(
+fn run_input(
-    mut run_context: RunContext,
+    mut run_ctxt: RunContext,
-    data_callback: &mut dyn FnMut(StreamData),
+    data_callback: &mut dyn FnMut(&Data),
     error_callback: &mut dyn FnMut(StreamError),
 ) {
+    loop {
+        match process_commands_and_await_signal(&mut run_ctxt, error_callback) {
+            Some(ControlFlow::Break) => break,
+            Some(ControlFlow::Continue) => continue,
+            None => (),
+        }
+        let capture_client = match run_ctxt.stream.client_flow {
+            AudioClientFlow::Capture { capture_client } => capture_client,
+            _ => unreachable!(),
+        };
+        match process_input(&mut run_ctxt.stream, capture_client, data_callback, error_callback) {
+            ControlFlow::Break => break,
+            ControlFlow::Continue => continue,
+        }
+    }
+}
+
+fn run_output(
+    mut run_ctxt: RunContext,
+    data_callback: &mut dyn FnMut(&mut Data),
+    error_callback: &mut dyn FnMut(StreamError),
+) {
+    loop {
+        match process_commands_and_await_signal(&mut run_ctxt, error_callback) {
+            Some(ControlFlow::Break) => break,
+            Some(ControlFlow::Continue) => continue,
+            None => (),
+        }
+        let render_client = match run_ctxt.stream.client_flow {
+            AudioClientFlow::Render { render_client } => render_client,
+            _ => unreachable!(),
+        };
+        match process_output(&mut run_ctxt.stream, render_client, data_callback, error_callback) {
+            ControlFlow::Break => break,
+            ControlFlow::Continue => continue,
+        }
+    }
+}
+
+enum ControlFlow {
+    Break,
+    Continue,
+}
+
+fn process_commands_and_await_signal(
+    run_context: &mut RunContext,
+    error_callback: &mut dyn FnMut(StreamError),
+) -> Option<ControlFlow> {
+    // Process queued commands.
+    match process_commands(run_context) {
+        Ok(true) => (),
+        Ok(false) => return Some(ControlFlow::Break),
+        Err(err) => {
+            error_callback(err);
+            return Some(ControlFlow::Break);
+        }
+    };
+
+    // Wait for any of the handles to be signalled.
+    let handle_idx = match wait_for_handle_signal(&run_context.handles) {
+        Ok(idx) => idx,
+        Err(err) => {
+            error_callback(err.into());
+            return Some(ControlFlow::Break);
+        }
+    };
+
+    // If `handle_idx` is 0, then it's `pending_scheduled_event` that was signalled in
+    // order for us to pick up the pending commands. Otherwise, a stream needs data.
+    if handle_idx == 0 {
+        return Some(ControlFlow::Continue);
+    }
+
+    None
+}
+
+// The loop for processing pending input data.
+fn process_input(
+    stream: &StreamInner,
+    capture_client: *mut audioclient::IAudioCaptureClient,
+    data_callback: &mut dyn FnMut(&Data),
+    error_callback: &mut dyn FnMut(StreamError),
+) -> ControlFlow {
+    let mut frames_available = 0;
     unsafe {
-        'stream_loop: loop {
+        // Get the available data in the shared buffer.
-            // Process queued commands.
+        let mut buffer: *mut BYTE = mem::uninitialized();
-            match process_commands(&mut run_context) {
+        let mut flags = mem::uninitialized();
-                Ok(true) => (),
+        loop {
-                Ok(false) => break,
+            let hresult = (*capture_client).GetNextPacketSize(&mut frames_available);
-                Err(err) => {
+            if let Err(err) = stream_error_from_hresult(hresult) {
-                    error_callback(err);
+                error_callback(err);
-                    break 'stream_loop;
+                return ControlFlow::Break;
-                }
+            }
-            };
+            if frames_available == 0 {
+                return ControlFlow::Continue;
+            }
+            let hresult = (*capture_client).GetBuffer(
+                &mut buffer,
+                &mut frames_available,
+                &mut flags,
+                ptr::null_mut(),
+                ptr::null_mut(),
+            );
 
-            // Wait for any of the handles to be signalled.
+            // TODO: Can this happen?
-            let handle_idx = match wait_for_handle_signal(&run_context.handles) {
+            if hresult == AUDCLNT_S_BUFFER_EMPTY {
-                Ok(idx) => idx,
-                Err(err) => {
-                    error_callback(err.into());
-                    break 'stream_loop;
-                }
-            };
-
-            // If `handle_idx` is 0, then it's `pending_scheduled_event` that was signalled in
-            // order for us to pick up the pending commands. Otherwise, a stream needs data.
-            if handle_idx == 0 {
                 continue;
+            } else if let Err(err) = stream_error_from_hresult(hresult) {
+                error_callback(err);
+                return ControlFlow::Break;
             }
 
-            let stream = &mut run_context.stream;
+            debug_assert!(!buffer.is_null());
-            let sample_size = stream.sample_format.sample_size();
 
-            // Obtaining a pointer to the buffer.
+            let data = buffer as *mut ();
-            match stream.client_flow {
+            let len = frames_available as usize
-                AudioClientFlow::Capture { capture_client } => {
+                * stream.bytes_per_frame as usize
-                    let mut frames_available = 0;
+                / stream.sample_format.sample_size();
-                    // Get the available data in the shared buffer.
+            let data = Data::from_parts(data, len, stream.sample_format);
-                    let mut buffer: *mut BYTE = mem::uninitialized();
+            data_callback(&data);
-                    let mut flags = mem::uninitialized();
-                    loop {
-                        let hresult = (*capture_client).GetNextPacketSize(&mut frames_available);
-                        if let Err(err) = stream_error_from_hresult(hresult) {
-                            error_callback(err);
-                            break 'stream_loop;
-                        }
-                        if frames_available == 0 {
-                            break;
-                        }
-                        let hresult = (*capture_client).GetBuffer(
-                            &mut buffer,
-                            &mut frames_available,
-                            &mut flags,
-                            ptr::null_mut(),
-                            ptr::null_mut(),
-                        );
 
-                        // TODO: Can this happen?
+            // Release the buffer.
-                        if hresult == AUDCLNT_S_BUFFER_EMPTY {
+            let hresult = (*capture_client).ReleaseBuffer(frames_available);
-                            continue;
+            if let Err(err) = stream_error_from_hresult(hresult) {
-                        } else if let Err(err) = stream_error_from_hresult(hresult) {
+                error_callback(err);
-                            error_callback(err);
+                return ControlFlow::Break;
-                            break 'stream_loop;
-                        }
-
-                        debug_assert!(!buffer.is_null());
-
-                        let buffer_len = frames_available as usize
-                            * stream.bytes_per_frame as usize
-                            / sample_size;
-
-                        // Simplify the capture callback sample format branches.
-                        macro_rules! capture_callback {
-                            ($T:ty, $Variant:ident) => {{
-                                let buffer_data = buffer as *mut _ as *const $T;
-                                let slice = slice::from_raw_parts(buffer_data, buffer_len);
-                                let unknown_buffer =
-                                    UnknownTypeInputBuffer::$Variant(::InputBuffer {
-                                        buffer: slice,
-                                    });
-                                let data = StreamData::Input {
-                                    buffer: unknown_buffer,
-                                };
-                                data_callback(data);
-                                // Release the buffer.
-                                let hresult = (*capture_client).ReleaseBuffer(frames_available);
-                                if let Err(err) = stream_error_from_hresult(hresult) {
-                                    error_callback(err);
-                                    break 'stream_loop;
-                                }
-                            }};
-                        }
-
-                        match stream.sample_format {
-                            SampleFormat::F32 => capture_callback!(f32, F32),
-                            SampleFormat::I16 => capture_callback!(i16, I16),
-                            SampleFormat::U16 => capture_callback!(u16, U16),
-                        }
-                    }
-                }
-
-                AudioClientFlow::Render { render_client } => {
-                    // The number of frames available for writing.
-                    let frames_available = match get_available_frames(&stream) {
-                        Ok(0) => continue, // TODO: Can this happen?
-                        Ok(n) => n,
-                        Err(err) => {
-                            error_callback(err);
-                            break 'stream_loop;
-                        }
-                    };
-
-                    let mut buffer: *mut BYTE = mem::uninitialized();
-                    let hresult =
-                        (*render_client).GetBuffer(frames_available, &mut buffer as *mut *mut _);
-
-                    if let Err(err) = stream_error_from_hresult(hresult) {
-                        error_callback(err);
-                        break 'stream_loop;
-                    }
-
-                    debug_assert!(!buffer.is_null());
-                    let buffer_len =
-                        frames_available as usize * stream.bytes_per_frame as usize / sample_size;
-
-                    // Simplify the render callback sample format branches.
-                    macro_rules! render_callback {
-                        ($T:ty, $Variant:ident) => {{
-                            let buffer_data = buffer as *mut $T;
-                            let slice = slice::from_raw_parts_mut(buffer_data, buffer_len);
-                            let unknown_buffer =
-                                UnknownTypeOutputBuffer::$Variant(::OutputBuffer { buffer: slice });
-                            let data = StreamData::Output {
-                                buffer: unknown_buffer,
-                            };
-                            data_callback(data);
-                            let hresult =
-                                (*render_client).ReleaseBuffer(frames_available as u32, 0);
-                            if let Err(err) = stream_error_from_hresult(hresult) {
-                                error_callback(err);
-                                break 'stream_loop;
-                            }
-                        }};
-                    }
-
-                    match stream.sample_format {
-                        SampleFormat::F32 => render_callback!(f32, F32),
-                        SampleFormat::I16 => render_callback!(i16, I16),
-                        SampleFormat::U16 => render_callback!(u16, U16),
-                    }
-                }
             }
         }
     }
 }
+
+// The loop for writing output data.
+fn process_output(
+    stream: &StreamInner,
+    render_client: *mut audioclient::IAudioRenderClient,
+    data_callback: &mut dyn FnMut(&mut Data),
+    error_callback: &mut dyn FnMut(StreamError),
+) -> ControlFlow {
+    // The number of frames available for writing.
+    let frames_available = match get_available_frames(&stream) {
+        Ok(0) => return ControlFlow::Continue, // TODO: Can this happen?
+        Ok(n) => n,
+        Err(err) => {
+            error_callback(err);
+            return ControlFlow::Break;
+        }
+    };
+
+    unsafe {
+        let mut buffer: *mut BYTE = mem::uninitialized();
+        let hresult =
+            (*render_client).GetBuffer(frames_available, &mut buffer as *mut *mut _);
+
+        if let Err(err) = stream_error_from_hresult(hresult) {
+            error_callback(err);
+            return ControlFlow::Break;
+        }
+
+        debug_assert!(!buffer.is_null());
+
+        let data = buffer as *mut ();
+        let len = frames_available as usize
+            * stream.bytes_per_frame as usize
+            / stream.sample_format.sample_size();
+        let mut data = Data::from_parts(data, len, stream.sample_format);
+        data_callback(&mut data);
+
+        let hresult = (*render_client).ReleaseBuffer(frames_available as u32, 0);
+        if let Err(err) = stream_error_from_hresult(hresult) {
+            error_callback(err);
+            return ControlFlow::Break;
+        }
+    }
+
+    ControlFlow::Continue
+}

src/lib.rs

@@ -55,13 +55,14 @@
 //! Now that we have everything for the stream, we are ready to create it from our selected device:
 //!
 //! ```no_run
+//! use cpal::Data;
 //! use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
 //! # let host = cpal::default_host();
 //! # let device = host.default_output_device().unwrap();
 //! # let format = device.default_output_format().unwrap();
 //! let stream = device.build_output_stream(
 //!     &format,
-//!     move |data| {
+//!     move |data: &mut Data| {
 //!         // react to stream events and read or write stream data here.
 //!     },
 //!     move |err| {
@@ -71,10 +72,8 @@
 //! ```
 //!
 //! While the stream is running, the selected audio device will periodically call the data callback
-//! that was passed to the function. The callback is passed an instance of type `StreamData` that
+//! that was passed to the function. The callback is passed an instance of either `&Data` or
-//! represents the data that must be read from or written to. The inner `UnknownTypeOutputBuffer`
+//! `&mut Data` depending on whether the stream is an input stream or output stream respectively.
-//! can be one of `I16`, `U16` or `F32` depending on the format that was passed to
-//! `build_output_stream`.
 //!
 //! > **Note**: Creating and running a stream will *not* block the thread. On modern platforms, the
 //! > given callback is called by a dedicated, high-priority thread responsible for delivering
@@ -85,40 +84,28 @@
 //! > please share your issue and use-case with the CPAL team on the github issue tracker for
 //! > consideration.*
 //!
-//! In this example, we simply fill the given output buffer with zeroes.
+//! In this example, we simply fill the given output buffer with silence.
 //!
 //! ```no_run
-//! use cpal::{StreamData, UnknownTypeOutputBuffer};
+//! use cpal::{Data, Sample, SampleFormat};
 //! use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
 //! # let host = cpal::default_host();
 //! # let device = host.default_output_device().unwrap();
 //! # let format = device.default_output_format().unwrap();
-//! let stream = device.build_output_stream(
+//! let err_fn = |err| eprintln!("an error occurred on the output audio stream: {}", err);
-//!     &format,
+//! let data_fn = move |data: &mut Data| match data.sample_format() {
-//!     move |data| {
+//!     SampleFormat::F32 => write_silence::<f32>(data),
-//!         match data {
+//!     SampleFormat::I16 => write_silence::<i16>(data),
-//!             StreamData::Output { buffer: UnknownTypeOutputBuffer::U16(mut buffer) } => {
+//!     SampleFormat::U16 => write_silence::<u16>(data),
-//!                 for elem in buffer.iter_mut() {
+//! };
-//!                     *elem = u16::max_value() / 2;
+//! let stream = device.build_output_stream(&format, data_fn, err_fn).unwrap();
-//!                 }
+//!
-//!             },
+//! fn write_silence<T: Sample>(data: &mut Data) {
-//!             StreamData::Output { buffer: UnknownTypeOutputBuffer::I16(mut buffer) } => {
+//!     let data = data.as_slice_mut::<T>().unwrap();
-//!                 for elem in buffer.iter_mut() {
+//!     for sample in data.iter_mut() {
-//!                     *elem = 0;
+//!         *sample = Sample::from(&0.0);
-//!                 }
+//!     }
-//!             },
+//! }
-//!             StreamData::Output { buffer: UnknownTypeOutputBuffer::F32(mut buffer) } => {
-//!                 for elem in buffer.iter_mut() {
-//!                     *elem = 0.0;
-//!                 }
-//!             },
-//!             _ => (),
-//!         }
-//!     },
-//!     move |err| {
-//!         eprintln!("an error occurred on the output audio stream: {}", err);
-//!     },
-//! );
 //! ```
 //!
 //! Not all platforms automatically run the stream upon creation. To ensure the stream has started,
@@ -129,7 +116,9 @@
 //! # let host = cpal::default_host();
 //! # let device = host.default_output_device().unwrap();
 //! # let format = device.default_output_format().unwrap();
-//! # let stream = device.build_output_stream(&format, move |_data| {}, move |_err| {}).unwrap();
+//! # let data_fn = move |_data: &mut cpal::Data| {};
+//! # let err_fn = move |_err| {};
+//! # let stream = device.build_output_stream(&format, data_fn, err_fn).unwrap();
 //! stream.play().unwrap();
 //! ```
 //!
@@ -141,8 +130,11 @@
 //! # let host = cpal::default_host();
 //! # let device = host.default_output_device().unwrap();
 //! # let format = device.default_output_format().unwrap();
-//! # let stream = device.build_output_stream(&format, move |_data| {}, move |_err| {}).unwrap();
+//! # let data_fn = move |_data: &mut cpal::Data| {};
+//! # let err_fn = move |_err| {};
+//! # let stream = device.build_output_stream(&format, data_fn, err_fn).unwrap();
 //! stream.pause().unwrap();
+//! ```
 
 #![recursion_limit = "512"]
 
@@ -161,7 +153,6 @@ pub use platform::{
     HostId, Stream, SupportedInputFormats, SupportedOutputFormats,
 };
 pub use samples_formats::{Sample, SampleFormat};
-use std::ops::{Deref, DerefMut};
 
 mod error;
 mod host;
@@ -202,71 +193,106 @@ pub struct SupportedFormat {
     pub data_type: SampleFormat,
 }
 
-/// Stream data passed to the `EventLoop::run` callback.
+/// Represents a buffer of audio data, delivered via a user's stream data callback function.
+///
+/// Input stream callbacks receive `&Data`, while output stream callbacks expect `&mut Data`.
 #[derive(Debug)]
-pub enum StreamData<'a> {
+pub struct Data {
-    Input {
+    data: *mut (),
-        buffer: UnknownTypeInputBuffer<'a>,
+    len: usize,
-    },
+    sample_format: SampleFormat,
-    Output {
-        buffer: UnknownTypeOutputBuffer<'a>,
-    },
 }
 
-/// Represents a buffer containing audio data that may be read.
+impl Data {
-///
+    // Internal constructor for host implementations to use.
-/// This struct implements the `Deref` trait targeting `[T]`. Therefore this buffer can be read the
+    //
-/// same way as reading from a `Vec` or any other kind of Rust array.
+    // The following requirements must be met in order for the safety of `Data`'s public API.
-// TODO: explain audio stuff in general
+    //
-// TODO: remove the wrapper and just use slices in next major version
+    // - The `data` pointer must point to the first sample in the slice containing all samples.
-#[derive(Debug)]
+    // - The `len` must describe the length of the buffer as a number of samples in the expected
-pub struct InputBuffer<'a, T: 'a>
+    //   format specified via the `sample_format` argument.
-where
+    // - The `sample_format` must correctly represent the underlying sample data delivered/expected
-    T: Sample,
+    //   by the stream.
-{
+    pub(crate) unsafe fn from_parts(
-    buffer: &'a [T],
+        data: *mut (),
-}
+        len: usize,
+        sample_format: SampleFormat,
+    ) -> Self {
+        Data { data, len, sample_format }
+    }
 
-/// Represents a buffer that must be filled with audio data. The buffer in unfilled state may
+    /// The sample format of the internal audio data.
-/// contain garbage values.
+    pub fn sample_format(&self) -> SampleFormat {
-///
+        self.sample_format
-/// This struct implements the `Deref` and `DerefMut` traits to `[T]`. Therefore writing to this
+    }
-/// buffer is done in the same way as writing to a `Vec` or any other kind of Rust array.
-// TODO: explain audio stuff in general
-// TODO: remove the wrapper and just use slices
-#[must_use]
-#[derive(Debug)]
-pub struct OutputBuffer<'a, T: 'a>
-where
-    T: Sample,
-{
-    buffer: &'a mut [T],
-}
 
-/// This is the struct that is provided to you by cpal when you want to read samples from a buffer.
+    /// The full length of the buffer in samples.
-///
+    ///
-/// Since the type of data is only known at runtime, you have to read the right buffer.
+    /// The returned length is the same length as the slice of type `T` that would be returned via
-#[derive(Debug)]
+    /// `as_slice` given a sample type that matches the inner sample format.
-pub enum UnknownTypeInputBuffer<'a> {
+    pub fn len(&self) -> usize {
-    /// Samples whose format is `u16`.
+        self.len
-    U16(InputBuffer<'a, u16>),
+    }
-    /// Samples whose format is `i16`.
-    I16(InputBuffer<'a, i16>),
-    /// Samples whose format is `f32`.
-    F32(InputBuffer<'a, f32>),
-}
 
-/// This is the struct that is provided to you by cpal when you want to write samples to a buffer.
+    /// The raw slice of memory representing the underlying audio data as a slice of bytes.
-///
+    ///
-/// Since the type of data is only known at runtime, you have to fill the right buffer.
+    /// It is up to the user to interpret the slice of memory based on `Data::sample_format`.
-#[derive(Debug)]
+    pub fn bytes(&self) -> &[u8] {
-pub enum UnknownTypeOutputBuffer<'a> {
+        let len = self.len * self.sample_format.sample_size();
-    /// Samples whose format is `u16`.
+        // The safety of this block relies on correct construction of the `Data` instance. See
-    U16(OutputBuffer<'a, u16>),
+        // the unsafe `from_parts` constructor for these requirements.
-    /// Samples whose format is `i16`.
+        unsafe {
-    I16(OutputBuffer<'a, i16>),
+            std::slice::from_raw_parts(self.data as *const u8, len)
-    /// Samples whose format is `f32`.
+        }
-    F32(OutputBuffer<'a, f32>),
+    }
+
+    /// The raw slice of memory representing the underlying audio data as a slice of bytes.
+    ///
+    /// It is up to the user to interpret the slice of memory based on `Data::sample_format`.
+    pub fn bytes_mut(&mut self) -> &mut [u8] {
+        let len = self.len * self.sample_format.sample_size();
+        // The safety of this block relies on correct construction of the `Data` instance. See
+        // the unsafe `from_parts` constructor for these requirements.
+        unsafe {
+            std::slice::from_raw_parts_mut(self.data as *mut u8, len)
+        }
+    }
+
+    /// Access the data as a slice of sample type `T`.
+    ///
+    /// Returns `None` if the sample type does not match the expected sample format.
+    pub fn as_slice<T>(&self) -> Option<&[T]>
+    where
+        T: Sample,
+    {
+        if T::FORMAT == self.sample_format {
+            // The safety of this block relies on correct construction of the `Data` instance. See
+            // the unsafe `from_parts` constructor for these requirements.
+            unsafe {
+                Some(std::slice::from_raw_parts(self.data as *const T, self.len))
+            }
+        } else {
+            None
+        }
+    }
+
+    /// Access the data as a slice of sample type `T`.
+    ///
+    /// Returns `None` if the sample type does not match the expected sample format.
+    pub fn as_slice_mut<T>(&mut self) -> Option<&mut [T]>
+    where
+        T: Sample,
+    {
+        if T::FORMAT == self.sample_format {
+            // The safety of this block relies on correct construction of the `Data` instance. See
+            // the unsafe `from_parts` constructor for these requirements.
+            unsafe {
+                Some(std::slice::from_raw_parts_mut(self.data as *mut T, self.len))
+            }
+        } else {
+            None
+        }
+    }
 }
 
 impl SupportedFormat {
@@ -352,61 +378,6 @@ impl SupportedFormat {
     }
 }
 
-impl<'a, T> Deref for InputBuffer<'a, T>
-    where T: Sample
-{
-    type Target = [T];
-
-    #[inline]
-    fn deref(&self) -> &[T] {
-        self.buffer
-    }
-}
-
-impl<'a, T> Deref for OutputBuffer<'a, T>
-    where T: Sample
-{
-    type Target = [T];
-
-    #[inline]
-    fn deref(&self) -> &[T] {
-        self.buffer
-    }
-}
-
-impl<'a, T> DerefMut for OutputBuffer<'a, T>
-    where T: Sample
-{
-    #[inline]
-    fn deref_mut(&mut self) -> &mut [T] {
-        self.buffer
-    }
-}
-
-impl<'a> UnknownTypeInputBuffer<'a> {
-    /// Returns the length of the buffer in number of samples.
-    #[inline]
-    pub fn len(&self) -> usize {
-        match self {
-            &UnknownTypeInputBuffer::U16(ref buf) => buf.len(),
-            &UnknownTypeInputBuffer::I16(ref buf) => buf.len(),
-            &UnknownTypeInputBuffer::F32(ref buf) => buf.len(),
-        }
-    }
-}
-
-impl<'a> UnknownTypeOutputBuffer<'a> {
-    /// Returns the length of the buffer in number of samples.
-    #[inline]
-    pub fn len(&self) -> usize {
-        match self {
-            &UnknownTypeOutputBuffer::U16(ref buf) => buf.len(),
-            &UnknownTypeOutputBuffer::I16(ref buf) => buf.len(),
-            &UnknownTypeOutputBuffer::F32(ref buf) => buf.len(),
-        }
-    }
-}
-
 impl From<Format> for SupportedFormat {
     #[inline]
     fn from(format: Format) -> SupportedFormat {

src/platform/mod.rs

@@ -255,8 +255,16 @@ macro_rules! impl_platform_host {
                 }
             }
 
-            fn build_input_stream<D, E>(&self, format: &crate::Format, data_callback: D, error_callback: E) -> Result<Self::Stream, crate::BuildStreamError>
+            fn build_input_stream<D, E>(
-                where D: FnMut(crate::StreamData) + Send + 'static, E: FnMut(crate::StreamError) + Send + 'static {
+                &self,
+                format: &crate::Format,
+                data_callback: D,
+                error_callback: E,
+            ) -> Result<Self::Stream, crate::BuildStreamError>
+            where
+                D: FnMut(&crate::Data) + Send + 'static,
+                E: FnMut(crate::StreamError) + Send + 'static,
+            {
                 match self.0 {
                     $(
                         DeviceInner::$HostVariant(ref d) => d.build_input_stream(format, data_callback, error_callback)
@@ -266,8 +274,16 @@ macro_rules! impl_platform_host {
                 }
             }
 
-            fn build_output_stream<D, E>(&self, format: &crate::Format, data_callback: D, error_callback: E) -> Result<Self::Stream, crate::BuildStreamError>
+            fn build_output_stream<D, E>(
-                where D: FnMut(crate::StreamData) + Send + 'static, E: FnMut(crate::StreamError) + Send + 'static {
+                &self,
+                format: &crate::Format,
+                data_callback: D,
+                error_callback: E,
+            ) -> Result<Self::Stream, crate::BuildStreamError>
+            where
+                D: FnMut(&mut crate::Data) + Send + 'static,
+                E: FnMut(crate::StreamError) + Send + 'static,
+            {
                 match self.0 {
                     $(
                         DeviceInner::$HostVariant(ref d) => d.build_output_stream(format, data_callback, error_callback)

src/samples_formats.rs

@@ -21,20 +21,12 @@ impl SampleFormat {
             &SampleFormat::F32 => mem::size_of::<f32>(),
         }
     }
-
-    /// Deprecated. Use `sample_size` instead.
-    #[inline]
-    #[deprecated]
-    pub fn get_sample_size(&self) -> usize {
-        self.sample_size()
-    }
 }
 
 /// Trait for containers that contain PCM data.
 pub unsafe trait Sample: Copy + Clone {
-    /// Returns the `SampleFormat` corresponding to this data type.
+    /// The `SampleFormat` corresponding to this data type.
-    // TODO: rename to `format()`. Requires a breaking change.
+    const FORMAT: SampleFormat;
-    fn get_format() -> SampleFormat;
 
     /// Turns the sample into its equivalent as a floating-point.
     fn to_f32(&self) -> f32;
@@ -49,10 +41,7 @@ pub unsafe trait Sample: Copy + Clone {
 }
 
 unsafe impl Sample for u16 {
-    #[inline]
+    const FORMAT: SampleFormat = SampleFormat::U16;
-    fn get_format() -> SampleFormat {
-        SampleFormat::U16
-    }
 
     #[inline]
     fn to_f32(&self) -> f32 {
@@ -82,10 +71,7 @@ unsafe impl Sample for u16 {
 }
 
 unsafe impl Sample for i16 {
-    #[inline]
+    const FORMAT: SampleFormat = SampleFormat::I16;
-    fn get_format() -> SampleFormat {
-        SampleFormat::I16
-    }
 
     #[inline]
     fn to_f32(&self) -> f32 {
@@ -119,10 +105,7 @@ unsafe impl Sample for i16 {
 }
 
 unsafe impl Sample for f32 {
-    #[inline]
+    const FORMAT: SampleFormat = SampleFormat::F32;
-    fn get_format() -> SampleFormat {
-        SampleFormat::F32
-    }
 
     #[inline]
     fn to_f32(&self) -> f32 {

src/traits.rs

@@ -2,6 +2,7 @@
 
 use {
     BuildStreamError,
+    Data,
     DefaultFormatError,
     DeviceNameError,
     DevicesError,
@@ -10,7 +11,6 @@ use {
     OutputDevices,
     PauseStreamError,
     PlayStreamError,
-    StreamData,
     StreamError,
     SupportedFormat,
     SupportedFormatsError,
@@ -118,12 +118,26 @@ pub trait DeviceTrait {
     fn default_output_format(&self) -> Result<Format, DefaultFormatError>;
 
     /// Create an input stream.
-    fn build_input_stream<D, E>(&self, format: &Format, data_callback: D, error_callback: E) -> Result<Self::Stream, BuildStreamError>
+    fn build_input_stream<D, E>(
-        where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static;
+        &self,
+        format: &Format,
+        data_callback: D,
+        error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
+    where
+        D: FnMut(&Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static;
 
     /// Create an output stream.
-    fn build_output_stream<D, E>(&self, format: &Format, data_callback: D, error_callback: E) -> Result<Self::Stream, BuildStreamError>
+    fn build_output_stream<D, E>(
-        where D: FnMut(StreamData) + Send + 'static, E: FnMut(StreamError) + Send + 'static;
+        &self,
+        format: &Format,
+        data_callback: D,
+        error_callback: E,
+    ) -> Result<Self::Stream, BuildStreamError>
+    where
+        D: FnMut(&mut Data) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static;
 }
 
 /// A stream created from `Device`, with methods to control playback.