Remove `UnknownTypeBuffer` in favour of specifying sample type.

This is an implementation of the planned changes described in #119.

For a quick overview of how the API has changed, check out the updated
examples.

**TODO:**

- [x] Update API.
- [x] Update examples.
- [ ] Remove `data_type` field from `Format` (see [here](https://github.com/RustAudio/cpal/issues/119#issuecomment-573788380)).
- Update backends:
  - [x] null
  - [x] ALSA
  - [ ] ASIO
  - [ ] WASAPI
  - [ ] CoreAudio
  - [ ] Emscripten

Closes #119
Closes #260

examples/beep.rs

@@ -1,56 +1,61 @@
 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| {
-        match data {
-            cpal::StreamData::Output { buffer: cpal::UnknownTypeOutputBuffer::U16(mut buffer) } => {
-                for sample in buffer.chunks_mut(channels as usize) {
-                    let value = ((next_value() * 0.5 + 0.5) * std::u16::MAX as f32) as u16;
-                    for out in sample.iter_mut() {
-                        *out = value;
-                    }
-                }
-            },
-            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);
-    })?;
+    };
+
+    let stream = match format.data_type {
+        cpal::SampleFormat::F32 => device.build_output_stream(
+            &format,
+            move |mut data| write_data::<f32>(&mut *data, channels, &mut next_value),
+            err_fn,
+        ),
+        cpal::SampleFormat::I16 => device.build_output_stream(
+            &format,
+            move |mut data| write_data::<i16>(&mut *data, channels, &mut next_value),
+            err_fn,
+        ),
+        cpal::SampleFormat::U16 => device.build_output_stream(
+            &format,
+            move |mut data| write_data::<u16>(&mut *data, channels, &mut next_value),
+            err_fn,
+        ),
+    }?;
+
     stream.play()?;
 
     std::thread::sleep(std::time::Duration::from_millis(1000));
 
     Ok(())
 }
+
+fn write_data<T>(output: &mut [T], channels: usize, next_sample: &mut dyn FnMut() -> f32)
+where
+    T: cpal::Sample,
+{
+    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

@@ -49,50 +49,44 @@ fn main() -> Result<(), anyhow::Error> {
 
     // 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(
-        match data {
+        &format,
-            cpal::StreamData::Input {
+        move |data| {
-                buffer: cpal::UnknownTypeInputBuffer::F32(buffer),
+            let mut output_fell_behind = false;
-            } => {
+            for &sample in data.iter() {
-                let mut output_fell_behind = false;
+                if producer.push(sample).is_err() {
-                for &sample in buffer.iter() {
+                    output_fell_behind = true;
-                    if producer.push(sample).is_err() {
-                        output_fell_behind = true;
-                    }
                 }
-                if output_fell_behind {
+            }
-                    eprintln!("output stream fell behind: try increasing latency");
+            if output_fell_behind {
-                }
+                eprintln!("output stream fell behind: try increasing latency");
-            },
+            }
-            _ => panic!("Expected input with f32 data"),
+        },
-        }
+        |err| {
-    }, move |err| {
+            eprintln!("an error occurred on stream: {}", err);
-        eprintln!("an error occurred on input stream: {}", err);
+        },
-    })?;
+    )?;
-    let output_stream = output_device.build_output_stream(&format, move |data| {
+    let output_stream = output_device.build_output_stream(
-        match data {
+        &format,
-            cpal::StreamData::Output {
+        move |mut data| {
-                buffer: cpal::UnknownTypeOutputBuffer::F32(mut buffer),
+            let mut input_fell_behind = None;
-            } => {
+            for sample in data.iter_mut() {
-                let mut input_fell_behind = None;
+                *sample = match consumer.pop() {
-                for sample in buffer.iter_mut() {
+                    Ok(s) => s,
-                    *sample = match consumer.pop() {
+                    Err(err) => {
-                        Ok(s) => s,
+                        input_fell_behind = Some(err);
-                        Err(err) => {
+                        0.0
-                            input_fell_behind = Some(err);
+                    },
-                            0.0
+                };
-                        },
+            }
-                    };
+            if let Some(err) = input_fell_behind {
-                }
+                eprintln!("input stream fell behind: {:?}: try increasing latency", err);
-                if let Some(err) = input_fell_behind {
+            }
-                    eprintln!("input stream fell behind: {:?}: try increasing latency", err);
+        },
-                }
+        move |err| {
-            },
+            eprintln!("an error occurred on output stream: {}", err);
-            _ => panic!("Expected output with f32 data"),
+        },
-        }
+    )?;
-    }, move |err| {
-        eprintln!("an error occurred on output stream: {}", err);
-    })?;
     println!("Successfully built streams.");
 
     // Play the streams.

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,36 @@ 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 stream = match format.data_type {
+        cpal::SampleFormat::F32 => device.build_input_stream(
+            &format,
+            move |mut data| write_input_data::<f32, f32>(&*data, &writer_2),
+            err_fn,
+        ),
+        cpal::SampleFormat::I16 => device.build_input_stream(
+            &format,
+            move |mut data| write_input_data::<i16, i16>(&*data, &writer_2),
+            err_fn,
+        ),
+        cpal::SampleFormat::U16 => device.build_input_stream(
+            &format,
+            move |mut data| write_input_data::<u16, i16>(&*data, &writer_2),
+            err_fn,
+        ),
+    }?;
+
     stream.play()?;
 
     // Let recording go for roughly three seconds.
@@ -100,3 +84,20 @@ 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: &[T], writer: &WavWriterHandle)
+where
+    T: cpal::Sample,
+    U: cpal::Sample + hound::Sample,
+{
+    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,30 @@
 extern crate alsa_sys as alsa;
 extern crate libc;
 
-use std::{cmp, ffi, io, mem, ptr};
+use crate::{
+    BackendSpecificError,
+    BuildStreamError,
+    ChannelCount,
+    DefaultFormatError,
+    DeviceNameError,
+    DevicesError,
+    Format,
+    InputData,
+    OutputData,
+    PauseStreamError,
+    PlayStreamError,
+    Sample,
+    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 +90,40 @@ 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<T, 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
+        T: Sample,
+        D: FnMut(InputData<T>) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
+        // TODO: Consider removing `data_type` field from `Format` and removing this.
+        assert_eq!(format.data_type, T::FORMAT, "sample format mismatch");
+        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<T, 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
+        T: Sample,
+        D: FnMut(OutputData<T>) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
+        // TODO: Consider removing `data_type` field from `Format` and removing this.
+        assert_eq!(format.data_type, T::FORMAT, "sample format mismatch");
+        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 +176,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 +543,7 @@ unsafe impl Send for StreamInner {}
 
 unsafe impl Sync for StreamInner {}
 
+#[derive(Debug, Eq, PartialEq)]
 enum StreamType { Input, Output }
 
 pub struct Stream {
@@ -524,183 +558,291 @@ 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<T>(
-        descriptors.reserve(stream.num_descriptors);
+    rx: TriggerReceiver,
-        let len = descriptors.len();
+    stream: &StreamInner,
-        let filled = unsafe {
+    data_callback: &mut (dyn FnMut(InputData<T>) + Send + 'static),
-            alsa::snd_pcm_poll_descriptors(
+    error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
+) where
+    T: Sample,
+{
+    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::<T>(
+                    stream,
+                    &mut ctxt.buffer,
+                    available_frames,
+                    data_callback,
+                    error_callback,
+                );
+            }
+        }
+    }
+}
+
+fn output_stream_worker<T>(
+    rx: TriggerReceiver,
+    stream: &StreamInner,
+    data_callback: &mut (dyn FnMut(OutputData<T>) + Send + 'static),
+    error_callback: &mut (dyn FnMut(StreamError) + Send + 'static),
+) where
+    T: Sample,
+{
+    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::<T>(
+                    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<T>(
+    stream: &StreamInner,
+    buffer: &mut [u8],
+    available_frames: usize,
+    data_callback: &mut (dyn FnMut(InputData<T>) + Send + 'static),
+    error_callback: &mut dyn FnMut(StreamError),
+) where
+    T: Sample,
+{
+    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 buffer = unsafe { cast_input_buffer::<T>(buffer) };
+    let input_data = InputData { buffer };
+    data_callback(input_data);
+}
+
+// Request data from the user's function and write it via ALSA.
+//
+// Returns `true`
+fn process_output<T>(
+    stream: &StreamInner,
+    buffer: &mut [u8],
+    available_frames: usize,
+    data_callback: &mut (dyn FnMut(OutputData<T>) + Send + 'static),
+    error_callback: &mut dyn FnMut(StreamError),
+) where
+    T: Sample,
+{
+    {
+        // We're now sure that we're ready to write data.
+        let buffer = unsafe { cast_output_buffer::<T>(buffer) };
+        let output_data = OutputData { buffer };
+        data_callback(output_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<T, 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
+        T: Sample,
+        D: FnMut(InputData<T>) + 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::<T>(rx, &*stream, &mut data_callback, &mut error_callback);
+        });
+        Stream {
+            thread: Some(thread),
+            inner,
+            trigger: tx,
+        }
+    }
+
+    fn new_output<T, D, E>(
+        inner: Arc<StreamInner>,
+        mut data_callback: D,
+        mut error_callback: E,
+    ) -> Stream
+    where
+        T: Sample,
+        D: FnMut(OutputData<T>) + 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::<T>(rx, &*stream, &mut data_callback, &mut error_callback);
         });
         Stream {
             thread: Some(thread),

src/host/null/mod.rs

@@ -1,14 +1,17 @@
-use BuildStreamError;
+use crate::{
-use DefaultFormatError;
+    BuildStreamError,
-use DevicesError;
+    DefaultFormatError,
-use DeviceNameError;
+    DevicesError,
-use Format;
+    DeviceNameError,
-use PauseStreamError;
+    Format,
-use PlayStreamError;
+    InputData,
-use StreamData;
+    OutputData,
-use StreamError;
+    PauseStreamError,
-use SupportedFormatsError;
+    PlayStreamError,
-use SupportedFormat;
+    StreamError,
+    SupportedFormatsError,
+    SupportedFormat,
+};
 use traits::{DeviceTrait, HostTrait, StreamTrait};
 
 #[derive(Default)]
@@ -68,14 +71,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<T, 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(InputData<T>) + 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<T, 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(OutputData<T>) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static,
+    {
         unimplemented!()
     }
 }

src/lib.rs

@@ -143,6 +143,7 @@
 //! # let format = device.default_output_format().unwrap();
 //! # let stream = device.build_output_stream(&format, move |_data| {}, move |_err| {}).unwrap();
 //! stream.pause().unwrap();
+//! ```
 
 #![recursion_limit = "512"]
 
@@ -202,25 +203,14 @@ pub struct SupportedFormat {
     pub data_type: SampleFormat,
 }
 
-/// Stream data passed to the `EventLoop::run` callback.
-#[derive(Debug)]
-pub enum StreamData<'a> {
-    Input {
-        buffer: UnknownTypeInputBuffer<'a>,
-    },
-    Output {
-        buffer: UnknownTypeOutputBuffer<'a>,
-    },
-}
-
 /// Represents a buffer containing audio data that may be read.
 ///
 /// 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.
 // TODO: explain audio stuff in general
-// TODO: remove the wrapper and just use slices in next major version
+// TODO: Consider making this an `enum` with `Interleaved` and `NonInterleaved` variants.
 #[derive(Debug)]
-pub struct InputBuffer<'a, T: 'a>
+pub struct InputData<'a, T: 'a>
 where
     T: Sample,
 {
@@ -233,42 +223,16 @@ where
 /// 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
+// TODO: Consider making this an `enum` with `Interleaved` and `NonInterleaved` variants.
 #[must_use]
 #[derive(Debug)]
-pub struct OutputBuffer<'a, T: 'a>
+pub struct OutputData<'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.
-///
-/// Since the type of data is only known at runtime, you have to read the right buffer.
-#[derive(Debug)]
-pub enum UnknownTypeInputBuffer<'a> {
-    /// Samples whose format is `u16`.
-    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.
-///
-/// Since the type of data is only known at runtime, you have to fill the right buffer.
-#[derive(Debug)]
-pub enum UnknownTypeOutputBuffer<'a> {
-    /// Samples whose format is `u16`.
-    U16(OutputBuffer<'a, u16>),
-    /// Samples whose format is `i16`.
-    I16(OutputBuffer<'a, i16>),
-    /// Samples whose format is `f32`.
-    F32(OutputBuffer<'a, f32>),
-}
-
 impl SupportedFormat {
     /// Turns this `SupportedFormat` into a `Format` corresponding to the maximum samples rate.
     #[inline]
@@ -352,8 +316,9 @@ impl SupportedFormat {
     }
 }
 
-impl<'a, T> Deref for InputBuffer<'a, T>
+impl<'a, T> Deref for InputData<'a, T>
-    where T: Sample
+where
+    T: Sample,
 {
     type Target = [T];
 
@@ -363,8 +328,9 @@ impl<'a, T> Deref for InputBuffer<'a, T>
     }
 }
 
-impl<'a, T> Deref for OutputBuffer<'a, T>
+impl<'a, T> Deref for OutputData<'a, T>
-    where T: Sample
+where
+    T: Sample,
 {
     type Target = [T];
 
@@ -374,8 +340,9 @@ impl<'a, T> Deref for OutputBuffer<'a, T>
     }
 }
 
-impl<'a, T> DerefMut for OutputBuffer<'a, T>
+impl<'a, T> DerefMut for OutputData<'a, T>
-    where T: Sample
+where
+    T: Sample,
 {
     #[inline]
     fn deref_mut(&mut self) -> &mut [T] {
@@ -383,30 +350,6 @@ impl<'a, T> DerefMut for OutputBuffer<'a, T>
     }
 }
 
-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,17 @@ 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<T, 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
+                T: crate::Sample,
+                D: FnMut(crate::InputData<T>) + 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 +275,17 @@ 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<T, 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
+                T: crate::Sample,
+                D: FnMut(crate::OutputData<T>) + 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

@@ -6,11 +6,13 @@ use {
     DeviceNameError,
     DevicesError,
     Format,
+    InputData,
     InputDevices,
+    OutputData,
     OutputDevices,
     PauseStreamError,
     PlayStreamError,
-    StreamData,
+    Sample,
     StreamError,
     SupportedFormat,
     SupportedFormatsError,
@@ -118,12 +120,28 @@ 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<T, 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
+        T: Sample,
+        D: FnMut(InputData<T>) + 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<T, 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
+        T: Sample,
+        D: FnMut(OutputData<T>) + Send + 'static,
+        E: FnMut(StreamError) + Send + 'static;
 }
 
 /// A stream created from `Device`, with methods to control playback.