cpal/src/platform/windows/asio/stream.rs

686 lines
29 KiB
Rust

extern crate asio_sys as sys;
extern crate num_traits;
use std;
use Format;
use CreationError;
use StreamData;
use super::Device;
use UnknownTypeInputBuffer;
use UnknownTypeOutputBuffer;
use std::sync::{Arc, Mutex};
use std::mem;
use std::sync::atomic::{AtomicUsize, Ordering};
use SampleFormat;
use super::asio_utils as au;
use self::num_traits::PrimInt;
pub struct EventLoop {
asio_streams: Arc<Mutex<sys::AsioStreams>>,
cpal_streams: Arc<Mutex<Vec<Option<Stream>>>>,
stream_count: AtomicUsize,
callbacks: Arc<Mutex<Vec<&'static mut (FnMut(StreamId, StreamData) + Send)>>>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct StreamId(usize);
pub struct InputBuffer<'a, T: 'a> {
buffer: &'a [T],
}
pub struct OutputBuffer<'a, T: 'a> {
buffer: &'a mut [T],
}
struct Stream{
playing: bool,
}
#[derive(Default)]
struct I16Buffer{
cpal: Vec<i16>,
channel: Vec<Vec<i16>>,
}
#[derive(Default)]
struct U16Buffer{
cpal: Vec<u16>,
channel: Vec<Vec<u16>>,
}
#[derive(Default)]
struct F32Buffer{
cpal: Vec<f32>,
channel: Vec<Vec<f32>>,
}
struct Buffers {
i16_buff: I16Buffer,
u16_buff: U16Buffer,
f32_buff: F32Buffer,
}
enum Endian {
Little,
Big,
}
impl EventLoop {
pub fn new() -> EventLoop {
EventLoop {
asio_streams: Arc::new(Mutex::new(sys::AsioStreams{input: None, output: None})),
cpal_streams: Arc::new(Mutex::new(Vec::new())),
stream_count: AtomicUsize::new(0),
callbacks: Arc::new(Mutex::new(Vec::new())),
}
}
/// Create a new CPAL Input Stream
/// If there is no ASIO Input Stream
/// it will be created
fn get_input_stream(&self, drivers: &sys::Drivers, format: &Format) -> Result<usize, CreationError> {
let Format {
channels,
sample_rate,
..
} = format;
let num_channels = *channels as usize;
let sample_rate = sample_rate.0;
let ref mut streams = *self.asio_streams.lock().unwrap();
if sample_rate != drivers.get_sample_rate().rate {
if drivers.can_sample_rate(sample_rate) {
drivers.set_sample_rate(sample_rate).expect("Unsupported sample rate");
} else {
panic!("This sample rate {:?} is not supported", sample_rate);
}
}
match streams.input {
Some(ref input) => Ok(input.buffer_size as usize),
None => {
let output = streams.output.take();
drivers.prepare_input_stream(output, num_channels)
.map(|new_streams| {
let bs = match new_streams.input {
Some(ref inp) => inp.buffer_size as usize,
None => unreachable!(),
};
*streams = new_streams;
bs
})
.map_err(|ref e| {
println!("Error preparing stream: {}", e);
CreationError::DeviceNotAvailable
})
}
}
}
fn get_output_stream(&self, drivers: &sys::Drivers, format: &Format) -> Result<usize, CreationError> {
let Format {
channels,
sample_rate,
..
} = format;
let num_channels = *channels as usize;
let sample_rate = sample_rate.0;
let ref mut streams = *self.asio_streams.lock().unwrap();
if sample_rate != drivers.get_sample_rate().rate {
if drivers.can_sample_rate(sample_rate) {
drivers.set_sample_rate(sample_rate).expect("Unsupported sample rate");
} else {
panic!("This sample rate {:?} is not supported", sample_rate);
}
}
match streams.output {
Some(ref output) => Ok(output.buffer_size as usize),
None => {
let input = streams.input.take();
drivers.prepare_output_stream(input, num_channels)
.map(|new_streams| {
let bs = match new_streams.output {
Some(ref out) => out.buffer_size as usize,
None => unreachable!(),
};
*streams = new_streams;
bs
})
.map_err(|ref e| {
println!("Error preparing stream: {}", e);
CreationError::DeviceNotAvailable
})
},
}
}
pub fn build_input_stream(
&self,
device: &Device,
format: &Format,
) -> Result<StreamId, CreationError> {
let Device {
drivers,
..
} = device;
let num_channels = format.channels.clone();
let stream_type = drivers.get_data_type().expect("Couldn't load data type");
self.get_input_stream(&drivers, format).map(|stream_buffer_size| {
let cpal_num_samples = stream_buffer_size * num_channels as usize;
let count = self.stream_count.load(Ordering::SeqCst);
self.stream_count.store(count + 1, Ordering::SeqCst);
let asio_streams = self.asio_streams.clone();
let cpal_streams = self.cpal_streams.clone();
let callbacks = self.callbacks.clone();
// Create buffers
let channel_len = cpal_num_samples
/ num_channels as usize;
let mut buffers = match format.data_type{
SampleFormat::I16 => {
Buffers{
i16_buff: I16Buffer{
cpal: vec![0 as i16; cpal_num_samples],
channel: (0..num_channels)
.map(|_| Vec::with_capacity(channel_len))
.collect()},
u16_buff: U16Buffer::default(),
f32_buff: F32Buffer::default(),
}
}
SampleFormat::U16 => {
Buffers{
i16_buff: I16Buffer::default(),
u16_buff: U16Buffer{
cpal: vec![0 as u16; cpal_num_samples],
channel: (0..num_channels)
.map(|_| Vec::with_capacity(channel_len))
.collect()},
f32_buff: F32Buffer::default(),
}
}
SampleFormat::F32 => {
Buffers{
i16_buff: I16Buffer::default(),
u16_buff: U16Buffer::default(),
f32_buff: F32Buffer{
cpal: vec![0 as f32; cpal_num_samples],
channel: (0..num_channels)
.map(|_| Vec::with_capacity(channel_len))
.collect()},
}
}
};
sys::set_callback(move |index| unsafe {
//if not playing return early
{
if let Some(s) = cpal_streams.lock().unwrap().get(count - 1){
if let Some(s) = s{
if !s.playing { return (); }
}
}
}
if let Some(ref asio_stream) = asio_streams.lock().unwrap().input {
// Number of samples needed total
let mut callbacks = callbacks.lock().unwrap();
// Theres only a single callback because theres only one event loop
match callbacks.first_mut() {
Some(callback) => {
macro_rules! convert_sample {
($AsioTypeIdent:ident,
u16,
$SampleTypeIdent:ident,
$Sample:expr
) => {
((*$Sample as f64 + $AsioTypeIdent::MAX as f64) /
(::std::u16::MAX as f64 /
::std::AsioTypeIdent::MAX as f64)) as u16
};
($AsioTypeIdent:ident,
$SampleType:ty,
$SampleTypeIdent:ident,
$Sample:expr
) => {
(*$Sample as i64 *
::std::$SampleTypeIdent::MAX as i64 /
::std::$AsioTypeIdent::MAX as i64) as $SampleType
};
};
macro_rules! try_callback {
($SampleFormat:ident,
$SampleType:ty,
$SampleTypeIdent:ident,
$AsioType:ty,
$AsioTypeIdent:ident,
$Buffers:expr,
$BuffersType:ty,
$BuffersTypeIdent:ident,
$Endianness:expr,
$ConvertEndian:expr
) => {
// For each channel write the cpal data to
// the asio buffer
// Also need to check for Endian
for (i, channel) in $Buffers.channel.iter_mut().enumerate(){
let buff_ptr = asio_stream
.buffer_infos[i]
.buffers[index as usize] as *mut $AsioType;
let asio_buffer: &'static [$AsioType] =
std::slice::from_raw_parts(
buff_ptr,
asio_stream.buffer_size as usize);
for asio_s in asio_buffer.iter(){
channel.push( $ConvertEndian(convert_sample!(
$AsioTypeIdent,
$SampleType,
$SampleTypeIdent,
asio_s), $Endianness));
}
}
// interleave all the channels
{
let $BuffersTypeIdent {
cpal: ref mut c_buffer,
channel: ref mut channels,
} = $Buffers;
au::interleave(&channels, c_buffer);
for c in channels.iter_mut() {
c.clear();
}
}
let buff = InputBuffer{
buffer: &mut $Buffers.cpal,
};
callback(
StreamId(count),
StreamData::Input{
buffer: UnknownTypeInputBuffer::$SampleFormat(
::InputBuffer{
buffer: Some(super::super::InputBuffer::Asio(buff))
})
}
);
}
};
// Generic over types
// TODO check for endianess
match stream_type {
sys::AsioSampleType::ASIOSTInt32LSB => {
try_callback!(I16, i16, i16, i32, i32,
buffers.i16_buff, I16Buffer, I16Buffer,
Endian::Little, convert_endian_to);
}
sys::AsioSampleType::ASIOSTInt16LSB => {
try_callback!(I16, i16, i16, i16, i16,
buffers.i16_buff, I16Buffer, I16Buffer,
Endian::Little, convert_endian_to);
}
sys::AsioSampleType::ASIOSTInt32MSB => {
try_callback!(I16, i16, i16, i32, i32,
buffers.i16_buff, I16Buffer, I16Buffer,
Endian::Big, convert_endian_to);
}
sys::AsioSampleType::ASIOSTInt16MSB => {
try_callback!(I16, i16, i16, i16, i16,
buffers.i16_buff, I16Buffer, I16Buffer,
Endian::Big, convert_endian_to);
}
sys::AsioSampleType::ASIOSTFloat32LSB => {
try_callback!(F32, f32, f32, f32, f32,
buffers.f32_buff, F32Buffer, F32Buffer,
Endian::Little, |a, _| a);
}
sys::AsioSampleType::ASIOSTFloat64LSB => {
try_callback!(F32, f32, f32, f64, f64,
buffers.f32_buff, F32Buffer, F32Buffer,
Endian::Little, |a, _| a);
}
sys::AsioSampleType::ASIOSTFloat32MSB => {
try_callback!(F32, f32, f32, f32, f32,
buffers.f32_buff, F32Buffer, F32Buffer,
Endian::Big, |a, _| a);
}
sys::AsioSampleType::ASIOSTFloat64MSB => {
try_callback!(F32, f32, f32, f64, f64,
buffers.f32_buff, F32Buffer, F32Buffer,
Endian::Big, |a, _| a);
}
_ => println!("unsupported format {:?}", stream_type),
}
}
None => return (),
}
}
});
self.cpal_streams.lock().unwrap().push(Some(Stream{ playing: false }));
StreamId(count)
})
}
pub fn build_output_stream(
&self,
device: &Device,
format: &Format,
) -> Result<StreamId, CreationError> {
let Device {
drivers,
..
} = device;
let num_channels = format.channels.clone();
let stream_type = drivers.get_data_type().expect("Couldn't load data type");
self.get_output_stream(&drivers, format).map(|stream_buffer_size| {
let cpal_num_samples = stream_buffer_size * num_channels as usize;
let count = self.stream_count.load(Ordering::SeqCst);
self.stream_count.store(count + 1, Ordering::SeqCst);
let asio_streams = self.asio_streams.clone();
let cpal_streams = self.cpal_streams.clone();
let callbacks = self.callbacks.clone();
// Create buffers
let channel_len = cpal_num_samples
/ num_channels as usize;
let mut re_buffers = match format.data_type{
SampleFormat::I16 => {
Buffers{
i16_buff: I16Buffer{
cpal: vec![0 as i16; cpal_num_samples],
channel: (0..num_channels)
.map(|_| Vec::with_capacity(channel_len))
.collect()},
u16_buff: U16Buffer::default(),
f32_buff: F32Buffer::default(),
}
}
SampleFormat::U16 => {
Buffers{
i16_buff: I16Buffer::default(),
u16_buff: U16Buffer{
cpal: vec![0 as u16; cpal_num_samples],
channel: (0..num_channels)
.map(|_| Vec::with_capacity(channel_len))
.collect()},
f32_buff: F32Buffer::default(),
}
}
SampleFormat::F32 => {
Buffers{
i16_buff: I16Buffer::default(),
u16_buff: U16Buffer::default(),
f32_buff: F32Buffer{
cpal: vec![0 as f32; cpal_num_samples],
channel: (0..num_channels)
.map(|_| Vec::with_capacity(channel_len))
.collect()},
}
}
};
sys::set_callback(move |index| unsafe {
//if not playing return early
{
if let Some(s) = cpal_streams.lock().unwrap().get(count - 1){
if let Some(s) = s{
if !s.playing { return (); }
}
}
}
if let Some(ref asio_stream) = asio_streams.lock().unwrap().output {
// Number of samples needed total
let mut callbacks = callbacks.lock().unwrap();
macro_rules! convert_sample {
($AsioTypeIdent:ident,
$AsioType:ty,
u16,
$Sample:expr
) => {
((*$Sample as i64 *
::std::$AsioTypeIdent::MAX as i64 /
::std::u16::MAX as i64) - $AsioTypeIdent::MAX as i64) as $AsioType
};
($AsioTypeIdent:ident,
$AsioType:ty,
$SampleTypeIdent:ident,
$Sample:expr
) => {
(*$Sample as i64 *
::std::$AsioTypeIdent::MAX as i64 /
::std::$SampleTypeIdent::MAX as i64) as $AsioType
};
};
// Theres only a single callback because theres only one event loop
match callbacks.first_mut() {
Some(callback) => {
macro_rules! try_callback {
($SampleFormat:ident,
$SampleType:ty,
$SampleTypeIdent:ident,
$AsioType:ty,
$AsioTypeIdent:ident,
$Buffers:expr,
$BuffersType:ty,
$BuffersTypeIdent:ident,
$Endianness:expr,
$ConvertEndian:expr
) => {
let mut my_buffers = $Buffers;
{
let buff = OutputBuffer{
buffer: &mut my_buffers.cpal
};
callback(
StreamId(count),
StreamData::Output{
buffer: UnknownTypeOutputBuffer::$SampleFormat(
::OutputBuffer{
target: Some(super::super::OutputBuffer::Asio(buff))
})
}
);
}
// Deinter all the channels
{
let $BuffersTypeIdent {
cpal: ref mut c_buffer,
channel: ref mut channels,
} = my_buffers;
au::deinterleave(&c_buffer[..], channels);
}
let silence = match index {
0 =>{
if !sys::SILENCE_FIRST.load(Ordering::SeqCst) {
sys::SILENCE_FIRST.store(true, Ordering::SeqCst);
sys::SILENCE_SECOND.store(false, Ordering::SeqCst);
true
}else{false}
},
1 =>{
if !sys::SILENCE_SECOND.load(Ordering::SeqCst) {
sys::SILENCE_SECOND.store(true, Ordering::SeqCst);
sys::SILENCE_FIRST.store(false, Ordering::SeqCst);
true
}else{false}
},
_ => unreachable!(),
};
// For each channel write the cpal data to
// the asio buffer
for (i, channel) in my_buffers.channel.iter().enumerate(){
let buff_ptr = asio_stream
.buffer_infos[i]
.buffers[index as usize] as *mut $AsioType;
let asio_buffer: &'static mut [$AsioType] =
std::slice::from_raw_parts_mut(
buff_ptr,
asio_stream.buffer_size as usize);
for (asio_s, cpal_s) in asio_buffer.iter_mut()
.zip(channel){
if silence { *asio_s = 0.0 as $AsioType; }
*asio_s += $ConvertEndian(convert_sample!(
$AsioTypeIdent,
$AsioType,
$SampleTypeIdent,
cpal_s
),
$Endianness);
}
}
};
}
// Generic over types
match stream_type {
sys::AsioSampleType::ASIOSTInt32LSB => {
try_callback!(I16, i16, i16, i32, i32,
&mut re_buffers.i16_buff, I16Buffer, I16Buffer,
Endian::Little, convert_endian_from);
}
sys::AsioSampleType::ASIOSTInt16LSB => {
try_callback!(I16, i16, i16, i16, i16,
&mut re_buffers.i16_buff, I16Buffer, I16Buffer,
Endian::Little, convert_endian_from);
}
sys::AsioSampleType::ASIOSTInt32MSB => {
try_callback!(I16, i16, i16, i32, i32,
&mut re_buffers.i16_buff, I16Buffer, I16Buffer,
Endian::Big, convert_endian_from);
}
sys::AsioSampleType::ASIOSTInt16MSB => {
try_callback!(I16, i16, i16, i16, i16,
&mut re_buffers.i16_buff, I16Buffer, I16Buffer,
Endian::Big, convert_endian_from);
}
sys::AsioSampleType::ASIOSTFloat32LSB => {
try_callback!(F32, f32, f32, f32, f32,
&mut re_buffers.f32_buff, F32Buffer, F32Buffer,
Endian::Little, |a, _| a);
}
sys::AsioSampleType::ASIOSTFloat64LSB => {
try_callback!(F32, f32, f32, f64, f64,
&mut re_buffers.f32_buff, F32Buffer, F32Buffer,
Endian::Little, |a, _| a);
}
sys::AsioSampleType::ASIOSTFloat32MSB => {
try_callback!(F32, f32, f32, f32, f32,
&mut re_buffers.f32_buff, F32Buffer, F32Buffer,
Endian::Big, |a, _| a);
}
sys::AsioSampleType::ASIOSTFloat64MSB => {
try_callback!(F32, f32, f32, f64, f64,
&mut re_buffers.f32_buff, F32Buffer, F32Buffer,
Endian::Big, |a, _| a);
}
_ => println!("unsupported format {:?}", stream_type),
}
}
None => return (),
}
}
});
self.cpal_streams.lock().unwrap().push(Some(Stream{ playing: false }));
StreamId(count)
})
}
pub fn play_stream(&self, stream_id: StreamId) {
let mut streams = self.cpal_streams.lock().unwrap();
if let Some(s) = streams.get_mut(stream_id.0).expect("Bad play stream index") {
s.playing = true;
}
// Calling play when already playing is a no-op
sys::play();
}
pub fn pause_stream(&self, stream_id: StreamId) {
let mut streams = self.cpal_streams.lock().unwrap();
if let Some(s) = streams.get_mut(stream_id.0).expect("Bad pause stream index") {
s.playing = false;
}
let any_playing = streams
.iter()
.filter(|s| if let Some(s) = s {
s.playing
} else {false} )
.next();
if let None = any_playing {
sys::stop();
}
}
pub fn destroy_stream(&self, stream_id: StreamId) {
let mut streams = self.cpal_streams.lock().unwrap();
streams.get_mut(stream_id.0).take();
let count = self.stream_count.load(Ordering::SeqCst);
self.stream_count.store(count - 1, Ordering::SeqCst);
if count == 1 {
*self.asio_streams.lock().unwrap() = sys::AsioStreams{ output: None, input: None };
sys::clean_up();
}
}
pub fn run<F>(&self, mut callback: F) -> !
where
F: FnMut(StreamId, StreamData) + Send,
{
let callback: &mut (FnMut(StreamId, StreamData) + Send) = &mut callback;
self.callbacks
.lock()
.unwrap()
.push(unsafe { mem::transmute(callback) });
loop {
// Might need a sleep here to prevent the loop being
// removed in --release
}
}
}
impl Drop for EventLoop {
fn drop(&mut self) {
sys::clean_up();
}
}
impl<'a, T> InputBuffer<'a, T> {
pub fn buffer(&self) -> &[T] {
&self.buffer
}
pub fn finish(self) {
}
}
impl<'a, T> OutputBuffer<'a, T> {
pub fn buffer(&mut self) -> &mut [T] {
&mut self.buffer
}
pub fn len(&self) -> usize {
self.buffer.len()
}
pub fn finish(self) {}
}
fn convert_endian_to<T: PrimInt>(sample: T, endian: Endian) -> T {
match endian {
Endian::Big => sample.to_be(),
Endian::Little => sample.to_le(),
}
}
fn convert_endian_from<T: PrimInt>(sample: T, endian: Endian) -> T {
match endian {
Endian::Big => T::from_be(sample),
Endian::Little => T::from_le(sample),
}
}