yetangitu/librespot
1
0
Fork 0
mirror of https://github.com/librespot-org/librespot.git synced 2025-10-05 02:39:53 +02:00
Code Issues Releases Wiki Activity

Merge branch 'dev' into tokio_migration

Browse source
This commit is contained in:
johannesd3 2021-04-10 10:27:24 +02:00
commit 26c127c2ec
23 changed files with 961 additions and 800 deletions
Download patch file Download diff file Expand all files Collapse all files

85
playback/src/audio_backend/alsa.rs
View file

@ -1,5 +1,7 @@
use super::{Open, Sink};
use super::{Open, Sink, SinkAsBytes};
use crate::audio::AudioPacket;
use crate::config::AudioFormat;
use crate::player::{NUM_CHANNELS, SAMPLES_PER_SECOND, SAMPLE_RATE};
use alsa::device_name::HintIter;
use alsa::pcm::{Access, Format, Frames, HwParams, PCM};
use alsa::{Direction, Error, ValueOr};
@ -8,13 +10,14 @@ use std::ffi::CString;
use std::io;
use std::process::exit;
const PREFERED_PERIOD_SIZE: Frames = 5512; // Period of roughly 125ms
const BUFFERED_LATENCY: f32 = 0.125; // seconds
const BUFFERED_PERIODS: Frames = 4;
pub struct AlsaSink {
pcm: Option<PCM>,
format: AudioFormat,
device: String,
buffer: Vec<i16>,
buffer: Vec<u8>,
}
fn list_outputs() {
@ -34,23 +37,27 @@ fn list_outputs() {
}
}
fn open_device(dev_name: &str) -> Result<(PCM, Frames), Box<Error>> {
fn open_device(dev_name: &str, format: AudioFormat) -> Result<(PCM, Frames), Box<Error>> {
let pcm = PCM::new(dev_name, Direction::Playback, false)?;
let mut period_size = PREFERED_PERIOD_SIZE;
let alsa_format = match format {
AudioFormat::F32 => Format::float(),
AudioFormat::S32 => Format::s32(),
AudioFormat::S24 => Format::s24(),
AudioFormat::S24_3 => Format::S243LE,
AudioFormat::S16 => Format::s16(),
};
// http://www.linuxjournal.com/article/6735?page=0,1#N0x19ab2890.0x19ba78d8
// latency = period_size * periods / (rate * bytes_per_frame)
// For 16 Bit stereo data, one frame has a length of four bytes.
// 500ms = buffer_size / (44100 * 4)
// buffer_size_bytes = 0.5 * 44100 / 4
// buffer_size_frames = 0.5 * 44100 = 22050
// For stereo samples encoded as 32-bit float, one frame has a length of eight bytes.
let mut period_size = ((SAMPLES_PER_SECOND * format.size() as u32) as f32
* (BUFFERED_LATENCY / BUFFERED_PERIODS as f32)) as Frames;
{
// Set hardware parameters: 44100 Hz / Stereo / 16 bit
let hwp = HwParams::any(&pcm)?;
hwp.set_access(Access::RWInterleaved)?;
hwp.set_format(Format::s16())?;
hwp.set_rate(44100, ValueOr::Nearest)?;
hwp.set_channels(2)?;
hwp.set_format(alsa_format)?;
hwp.set_rate(SAMPLE_RATE, ValueOr::Nearest)?;
hwp.set_channels(NUM_CHANNELS as u32)?;
period_size = hwp.set_period_size_near(period_size, ValueOr::Greater)?;
hwp.set_buffer_size_near(period_size * BUFFERED_PERIODS)?;
pcm.hw_params(&hwp)?;
@ -64,12 +71,12 @@ fn open_device(dev_name: &str) -> Result<(PCM, Frames), Box<Error>> {
}
impl Open for AlsaSink {
fn open(device: Option<String>) -> AlsaSink {
info!("Using alsa sink");
fn open(device: Option<String>, format: AudioFormat) -> Self {
info!("Using Alsa sink with format: {:?}", format);
let name = match device.as_ref().map(AsRef::as_ref) {
Some("?") => {
println!("Listing available alsa outputs");
println!("Listing available Alsa outputs:");
list_outputs();
exit(0)
}
@ -78,8 +85,9 @@ impl Open for AlsaSink {
}
.to_string();
AlsaSink {
Self {
pcm: None,
format: format,
device: name,
buffer: vec![],
}
@ -89,12 +97,14 @@ impl Open for AlsaSink {
impl Sink for AlsaSink {
fn start(&mut self) -> io::Result<()> {
if self.pcm.is_none() {
let pcm = open_device(&self.device);
let pcm = open_device(&self.device, self.format);
match pcm {
Ok((p, period_size)) => {
self.pcm = Some(p);
// Create a buffer for all samples for a full period
self.buffer = Vec::with_capacity((period_size * 2) as usize);
self.buffer = Vec::with_capacity(
period_size as usize * BUFFERED_PERIODS as usize * self.format.size(),
);
}
Err(e) => {
error!("Alsa error PCM open {}", e);
@ -111,23 +121,22 @@ impl Sink for AlsaSink {
fn stop(&mut self) -> io::Result<()> {
{
let pcm = self.pcm.as_mut().unwrap();
// Write any leftover data in the period buffer
// before draining the actual buffer
let io = pcm.io_i16().unwrap();
match io.writei(&self.buffer[..]) {
Ok(_) => (),
Err(err) => pcm.try_recover(err, false).unwrap(),
}
self.write_bytes(&[]).expect("could not flush buffer");
let pcm = self.pcm.as_mut().unwrap();
pcm.drain().unwrap();
}
self.pcm = None;
Ok(())
}
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
sink_as_bytes!();
}
impl SinkAsBytes for AlsaSink {
fn write_bytes(&mut self, data: &[u8]) -> io::Result<()> {
let mut processed_data = 0;
let data = packet.samples();
while processed_data < data.len() {
let data_to_buffer = min(
self.buffer.capacity() - self.buffer.len(),
@ -137,12 +146,7 @@ impl Sink for AlsaSink {
.extend_from_slice(&data[processed_data..processed_data + data_to_buffer]);
processed_data += data_to_buffer;
if self.buffer.len() == self.buffer.capacity() {
let pcm = self.pcm.as_mut().unwrap();
let io = pcm.io_i16().unwrap();
match io.writei(&self.buffer) {
Ok(_) => (),
Err(err) => pcm.try_recover(err, false).unwrap(),
}
self.write_buf().expect("could not append to buffer");
self.buffer.clear();
}
}
@ -150,3 +154,16 @@ impl Sink for AlsaSink {
Ok(())
}
}
impl AlsaSink {
fn write_buf(&mut self) -> io::Result<()> {
let pcm = self.pcm.as_mut().unwrap();
let io = pcm.io_bytes();
match io.writei(&self.buffer) {
Ok(_) => (),
Err(err) => pcm.try_recover(err, false).unwrap(),
};
Ok(())
}
}

74
playback/src/audio_backend/gstreamer.rs
View file

@ -1,10 +1,13 @@
use super::{Open, Sink};
use super::{Open, Sink, SinkAsBytes};
use crate::audio::AudioPacket;
use crate::config::AudioFormat;
use crate::player::{NUM_CHANNELS, SAMPLE_RATE};
use gst::prelude::*;
use gstreamer as gst;
use gstreamer_app as gst_app;
use zerocopy::*;
use gst::prelude::*;
use zerocopy::AsBytes;
use std::sync::mpsc::{sync_channel, SyncSender};
use std::{io, thread};
@ -13,12 +16,27 @@ use std::{io, thread};
pub struct GstreamerSink {
tx: SyncSender<Vec<u8>>,
pipeline: gst::Pipeline,
format: AudioFormat,
}
impl Open for GstreamerSink {
fn open(device: Option<String>) -> GstreamerSink {
gst::init().expect("Failed to init gstreamer!");
let pipeline_str_preamble = r#"appsrc caps="audio/x-raw,format=S16LE,layout=interleaved,channels=2,rate=44100" block=true max-bytes=4096 name=appsrc0 "#;
fn open(device: Option<String>, format: AudioFormat) -> Self {
info!("Using GStreamer sink with format: {:?}", format);
gst::init().expect("failed to init GStreamer!");
// GStreamer calls S24 and S24_3 different from the rest of the world
let gst_format = match format {
AudioFormat::S24 => "S24_32".to_string(),
AudioFormat::S24_3 => "S24".to_string(),
_ => format!("{:?}", format),
};
let sample_size = format.size();
let gst_bytes = 2048 * sample_size;
let pipeline_str_preamble = format!(
"appsrc caps=\"audio/x-raw,format={}LE,layout=interleaved,channels={},rate={}\" block=true max-bytes={} name=appsrc0 ",
gst_format, NUM_CHANNELS, SAMPLE_RATE, gst_bytes
);
let pipeline_str_rest = r#" ! audioconvert ! autoaudiosink"#;
let pipeline_str: String = match device {
Some(x) => format!("{}{}", pipeline_str_preamble, x),
@ -30,27 +48,27 @@ impl Open for GstreamerSink {
let pipelinee = gst::parse_launch(&*pipeline_str).expect("Couldn't launch pipeline; likely a GStreamer issue or an error in the pipeline string you specified in the 'device' argument to librespot.");
let pipeline = pipelinee
.dynamic_cast::<gst::Pipeline>()
.expect("Couldn't cast pipeline element at runtime!");
let bus = pipeline.get_bus().expect("Couldn't get bus from pipeline");
.expect("couldn't cast pipeline element at runtime!");
let bus = pipeline.get_bus().expect("couldn't get bus from pipeline");
let mainloop = glib::MainLoop::new(None, false);
let appsrce: gst::Element = pipeline
.get_by_name("appsrc0")
.expect("Couldn't get appsrc from pipeline");
.expect("couldn't get appsrc from pipeline");
let appsrc: gst_app::AppSrc = appsrce
.dynamic_cast::<gst_app::AppSrc>()
.expect("Couldn't cast AppSrc element at runtime!");
.expect("couldn't cast AppSrc element at runtime!");
let bufferpool = gst::BufferPool::new();
let appsrc_caps = appsrc.get_caps().expect("Couldn't get appsrc caps");
let appsrc_caps = appsrc.get_caps().expect("couldn't get appsrc caps");
let mut conf = bufferpool.get_config();
conf.set_params(Some(&appsrc_caps), 8192, 0, 0);
conf.set_params(Some(&appsrc_caps), 4096 * sample_size as u32, 0, 0);
bufferpool
.set_config(conf)
.expect("Couldn't configure the buffer pool");
.expect("couldn't configure the buffer pool");
bufferpool
.set_active(true)
.expect("Couldn't activate buffer pool");
.expect("couldn't activate buffer pool");
let (tx, rx) = sync_channel::<Vec<u8>>(128);
let (tx, rx) = sync_channel::<Vec<u8>>(64 * sample_size);
thread::spawn(move || {
for data in rx {
let buffer = bufferpool.acquire_buffer(None);
@ -85,30 +103,32 @@ impl Open for GstreamerSink {
glib::Continue(true)
})
.expect("Failed to add bus watch");
.expect("failed to add bus watch");
thread_mainloop.run();
});
pipeline
.set_state(gst::State::Playing)
.expect("Unable to set the pipeline to the `Playing` state");
.expect("unable to set the pipeline to the `Playing` state");
GstreamerSink { tx, pipeline }
Self {
tx,
pipeline,
format,
}
}
}
impl Sink for GstreamerSink {
fn start(&mut self) -> io::Result<()> {
Ok(())
}
fn stop(&mut self) -> io::Result<()> {
Ok(())
}
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
start_stop_noop!();
sink_as_bytes!();
}
impl SinkAsBytes for GstreamerSink {
fn write_bytes(&mut self, data: &[u8]) -> io::Result<()> {
// Copy expensively (in to_vec()) to avoid thread synchronization
let deighta: &[u8] = packet.samples().as_bytes();
self.tx
.send(deighta.to_vec())
.send(data.to_vec())
.expect("tx send failed in write function");
Ok(())
}

37
playback/src/audio_backend/jackaudio.rs
View file

@ -1,5 +1,7 @@
use super::{Open, Sink};
use crate::audio::AudioPacket;
use crate::config::AudioFormat;
use crate::player::NUM_CHANNELS;
use jack::{
AsyncClient, AudioOut, Client, ClientOptions, Control, Port, ProcessHandler, ProcessScope,
};
@ -7,20 +9,18 @@ use std::io;
use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
pub struct JackSink {
send: SyncSender<i16>,
send: SyncSender<f32>,
// We have to keep hold of this object, or the Sink can't play...
#[allow(dead_code)]
active_client: AsyncClient<(), JackData>,
}
pub struct JackData {
rec: Receiver<i16>,
rec: Receiver<f32>,
port_l: Port<AudioOut>,
port_r: Port<AudioOut>,
}
fn pcm_to_f32(sample: i16) -> f32 {
sample as f32 / 32768.0
}
impl ProcessHandler for JackData {
fn process(&mut self, _: &Client, ps: &ProcessScope) -> Control {
// get output port buffers
@ -33,16 +33,19 @@ impl ProcessHandler for JackData {
let buf_size = buf_r.len();
for i in 0..buf_size {
buf_r[i] = pcm_to_f32(queue_iter.next().unwrap_or(0));
buf_l[i] = pcm_to_f32(queue_iter.next().unwrap_or(0));
buf_r[i] = queue_iter.next().unwrap_or(0.0);
buf_l[i] = queue_iter.next().unwrap_or(0.0);
}
Control::Continue
}
}
impl Open for JackSink {
fn open(client_name: Option<String>) -> JackSink {
info!("Using jack sink!");
fn open(client_name: Option<String>, format: AudioFormat) -> Self {
if format != AudioFormat::F32 {
warn!("JACK currently does not support {:?} output", format);
}
info!("Using JACK sink with format {:?}", AudioFormat::F32);
let client_name = client_name.unwrap_or("librespot".to_string());
let (client, _status) =
@ -50,7 +53,7 @@ impl Open for JackSink {
let ch_r = client.register_port("out_0", AudioOut::default()).unwrap();
let ch_l = client.register_port("out_1", AudioOut::default()).unwrap();
// buffer for samples from librespot (~10ms)
let (tx, rx) = sync_channel(2 * 1024 * 4);
let (tx, rx) = sync_channel::<f32>(NUM_CHANNELS as usize * 1024 * AudioFormat::F32.size());
let jack_data = JackData {
rec: rx,
port_l: ch_l,
@ -58,7 +61,7 @@ impl Open for JackSink {
};
let active_client = AsyncClient::new(client, (), jack_data).unwrap();
JackSink {
Self {
send: tx,
active_client,
}
@ -66,19 +69,13 @@ impl Open for JackSink {
}
impl Sink for JackSink {
fn start(&mut self) -> io::Result<()> {
Ok(())
}
fn stop(&mut self) -> io::Result<()> {
Ok(())
}
start_stop_noop!();
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
for s in packet.samples().iter() {
let res = self.send.send(*s);
if res.is_err() {
error!("jackaudio: cannot write to channel");
error!("cannot write to channel");
}
}
Ok(())

56
playback/src/audio_backend/mod.rs
View file

@ -1,8 +1,9 @@
use crate::audio::AudioPacket;
use crate::config::AudioFormat;
use std::io;
pub trait Open {
fn open(_: Option<String>) -> Self;
fn open(_: Option<String>, format: AudioFormat) -> Self;
}
pub trait Sink {
@ -11,10 +12,57 @@ pub trait Sink {
fn write(&mut self, packet: &AudioPacket) -> io::Result<()>;
}
pub type SinkBuilder = fn(Option<String>) -> Box<dyn Sink>;
pub type SinkBuilder = fn(Option<String>, AudioFormat) -> Box<dyn Sink>;
fn mk_sink<S: Sink + Open + 'static>(device: Option<String>) -> Box<dyn Sink> {
Box::new(S::open(device))
pub trait SinkAsBytes {
fn write_bytes(&mut self, data: &[u8]) -> io::Result<()>;
}
fn mk_sink<S: Sink + Open + 'static>(device: Option<String>, format: AudioFormat) -> Box<dyn Sink> {
Box::new(S::open(device, format))
}
// reuse code for various backends
macro_rules! sink_as_bytes {
() => {
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
use crate::audio::{i24, SamplesConverter};
use zerocopy::AsBytes;
match packet {
AudioPacket::Samples(samples) => match self.format {
AudioFormat::F32 => self.write_bytes(samples.as_bytes()),
AudioFormat::S32 => {
let samples_s32: &[i32] = &SamplesConverter::to_s32(samples);
self.write_bytes(samples_s32.as_bytes())
}
AudioFormat::S24 => {
let samples_s24: &[i32] = &SamplesConverter::to_s24(samples);
self.write_bytes(samples_s24.as_bytes())
}
AudioFormat::S24_3 => {
let samples_s24_3: &[i24] = &SamplesConverter::to_s24_3(samples);
self.write_bytes(samples_s24_3.as_bytes())
}
AudioFormat::S16 => {
let samples_s16: &[i16] = &SamplesConverter::to_s16(samples);
self.write_bytes(samples_s16.as_bytes())
}
},
AudioPacket::OggData(samples) => self.write_bytes(samples),
}
}
};
}
macro_rules! start_stop_noop {
() => {
fn start(&mut self) -> io::Result<()> {
Ok(())
}
fn stop(&mut self) -> io::Result<()> {
Ok(())
}
};
}
#[cfg(feature = "alsa-backend")]

54
playback/src/audio_backend/pipe.rs
View file

@ -1,46 +1,36 @@
use super::{Open, Sink};
use super::{Open, Sink, SinkAsBytes};
use crate::audio::AudioPacket;
use crate::config::AudioFormat;
use std::fs::OpenOptions;
use std::io::{self, Write};
use std::mem;
use std::slice;
pub struct StdoutSink(Box<dyn Write + Send>);
pub struct StdoutSink {
output: Box<dyn Write>,
format: AudioFormat,
}
impl Open for StdoutSink {
fn open(path: Option<String>) -> StdoutSink {
if let Some(path) = path {
let file = OpenOptions::new().write(true).open(path).unwrap();
StdoutSink(Box::new(file))
} else {
StdoutSink(Box::new(io::stdout()))
}
fn open(path: Option<String>, format: AudioFormat) -> Self {
info!("Using pipe sink with format: {:?}", format);
let output: Box<dyn Write> = match path {
Some(path) => Box::new(OpenOptions::new().write(true).open(path).unwrap()),
_ => Box::new(io::stdout()),
};
Self { output, format }
}
}
impl Sink for StdoutSink {
fn start(&mut self) -> io::Result<()> {
Ok(())
}
fn stop(&mut self) -> io::Result<()> {
Ok(())
}
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
let data: &[u8] = match packet {
AudioPacket::Samples(data) => unsafe {
slice::from_raw_parts(
data.as_ptr() as *const u8,
data.len() * mem::size_of::<i16>(),
)
},
AudioPacket::OggData(data) => data,
};
self.0.write_all(data)?;
self.0.flush()?;
start_stop_noop!();
sink_as_bytes!();
}
impl SinkAsBytes for StdoutSink {
fn write_bytes(&mut self, data: &[u8]) -> io::Result<()> {
self.output.write_all(data)?;
self.output.flush()?;
Ok(())
}
}

134
playback/src/audio_backend/portaudio.rs
View file

@ -1,5 +1,7 @@
use super::{Open, Sink};
use crate::audio::AudioPacket;
use crate::audio::{AudioPacket, SamplesConverter};
use crate::config::AudioFormat;
use crate::player::{NUM_CHANNELS, SAMPLE_RATE};
use portaudio_rs;
use portaudio_rs::device::{get_default_output_index, DeviceIndex, DeviceInfo};
use portaudio_rs::stream::*;
@ -7,10 +9,20 @@ use std::io;
use std::process::exit;
use std::time::Duration;
pub struct PortAudioSink<'a>(
Option<portaudio_rs::stream::Stream<'a, i16, i16>>,
StreamParameters<i16>,
);
pub enum PortAudioSink<'a> {
F32(
Option<portaudio_rs::stream::Stream<'a, f32, f32>>,
StreamParameters<f32>,
),
S32(
Option<portaudio_rs::stream::Stream<'a, i32, i32>>,
StreamParameters<i32>,
),
S16(
Option<portaudio_rs::stream::Stream<'a, i16, i16>>,
StreamParameters<i16>,
),
}
fn output_devices() -> Box<dyn Iterator<Item = (DeviceIndex, DeviceInfo)>> {
let count = portaudio_rs::device::get_count().unwrap();
@ -40,8 +52,11 @@ fn find_output(device: &str) -> Option<DeviceIndex> {
}
impl<'a> Open for PortAudioSink<'a> {
fn open(device: Option<String>) -> PortAudioSink<'a> {
debug!("Using PortAudio sink");
fn open(device: Option<String>, format: AudioFormat) -> PortAudioSink<'a> {
info!("Using PortAudio sink with format: {:?}", format);
warn!("This backend is known to panic on several platforms.");
warn!("Consider using some other backend, or better yet, contributing a fix.");
portaudio_rs::initialize().unwrap();
@ -53,7 +68,7 @@ impl<'a> Open for PortAudioSink<'a> {
Some(device) => find_output(device),
None => get_default_output_index(),
}
.expect("Could not find device");
.expect("could not find device");
let info = portaudio_rs::device::get_info(device_idx);
let latency = match info {
@ -61,46 +76,99 @@ impl<'a> Open for PortAudioSink<'a> {
None => Duration::new(0, 0),
};
let params = StreamParameters {
device: device_idx,
channel_count: 2,
suggested_latency: latency,
data: 0i16,
};
PortAudioSink(None, params)
macro_rules! open_sink {
($sink: expr, $type: ty) => {{
let params = StreamParameters {
device: device_idx,
channel_count: NUM_CHANNELS as u32,
suggested_latency: latency,
data: 0.0 as $type,
};
$sink(None, params)
}};
}
match format {
AudioFormat::F32 => open_sink!(Self::F32, f32),
AudioFormat::S32 => open_sink!(Self::S32, i32),
AudioFormat::S16 => open_sink!(Self::S16, i16),
_ => {
unimplemented!("PortAudio currently does not support {:?} output", format)
}
}
}
}
impl<'a> Sink for PortAudioSink<'a> {
fn start(&mut self) -> io::Result<()> {
if self.0.is_none() {
self.0 = Some(
Stream::open(
None,
Some(self.1),
44100.0,
FRAMES_PER_BUFFER_UNSPECIFIED,
StreamFlags::empty(),
None,
)
.unwrap(),
);
macro_rules! start_sink {
(ref mut $stream: ident, ref $parameters: ident) => {{
if $stream.is_none() {
*$stream = Some(
Stream::open(
None,
Some(*$parameters),
SAMPLE_RATE as f64,
FRAMES_PER_BUFFER_UNSPECIFIED,
StreamFlags::empty(),
None,
)
.unwrap(),
);
}
$stream.as_mut().unwrap().start().unwrap()
}};
}
self.0.as_mut().unwrap().start().unwrap();
match self {
Self::F32(stream, parameters) => start_sink!(ref mut stream, ref parameters),
Self::S32(stream, parameters) => start_sink!(ref mut stream, ref parameters),
Self::S16(stream, parameters) => start_sink!(ref mut stream, ref parameters),
};
Ok(())
}
fn stop(&mut self) -> io::Result<()> {
self.0.as_mut().unwrap().stop().unwrap();
self.0 = None;
macro_rules! stop_sink {
(ref mut $stream: ident) => {{
$stream.as_mut().unwrap().stop().unwrap();
*$stream = None;
}};
}
match self {
Self::F32(stream, _parameters) => stop_sink!(ref mut stream),
Self::S32(stream, _parameters) => stop_sink!(ref mut stream),
Self::S16(stream, _parameters) => stop_sink!(ref mut stream),
};
Ok(())
}
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
match self.0.as_mut().unwrap().write(packet.samples()) {
macro_rules! write_sink {
(ref mut $stream: expr, $samples: expr) => {
$stream.as_mut().unwrap().write($samples)
};
}
let samples = packet.samples();
let result = match self {
Self::F32(stream, _parameters) => {
write_sink!(ref mut stream, samples)
}
Self::S32(stream, _parameters) => {
let samples_s32: &[i32] = &SamplesConverter::to_s32(samples);
write_sink!(ref mut stream, samples_s32)
}
Self::S16(stream, _parameters) => {
let samples_s16: &[i16] = &SamplesConverter::to_s16(samples);
write_sink!(ref mut stream, samples_s16)
}
};
match result {
Ok(_) => (),
Err(portaudio_rs::PaError::OutputUnderflowed) => error!("PortAudio write underflow"),
Err(e) => panic!("PA Error {}", e),
Err(e) => panic!("PortAudio error {}", e),
};
Ok(())

47
playback/src/audio_backend/pulseaudio.rs
View file

@ -1,5 +1,7 @@
use super::{Open, Sink};
use super::{Open, Sink, SinkAsBytes};
use crate::audio::AudioPacket;
use crate::config::AudioFormat;
use crate::player::{NUM_CHANNELS, SAMPLE_RATE};
use libpulse_binding::{self as pulse, stream::Direction};
use libpulse_simple_binding::Simple;
use std::io;
@ -11,23 +13,34 @@ pub struct PulseAudioSink {
s: Option<Simple>,
ss: pulse::sample::Spec,
device: Option<String>,
format: AudioFormat,
}
impl Open for PulseAudioSink {
fn open(device: Option<String>) -> PulseAudioSink {
debug!("Using PulseAudio sink");
fn open(device: Option<String>, format: AudioFormat) -> Self {
info!("Using PulseAudio sink with format: {:?}", format);
// PulseAudio calls S24 and S24_3 different from the rest of the world
let pulse_format = match format {
AudioFormat::F32 => pulse::sample::Format::F32le,
AudioFormat::S32 => pulse::sample::Format::S32le,
AudioFormat::S24 => pulse::sample::Format::S24_32le,
AudioFormat::S24_3 => pulse::sample::Format::S24le,
AudioFormat::S16 => pulse::sample::Format::S16le,
};
let ss = pulse::sample::Spec {
format: pulse::sample::Format::S16le,
channels: 2, // stereo
rate: 44100,
format: pulse_format,
channels: NUM_CHANNELS,
rate: SAMPLE_RATE,
};
debug_assert!(ss.is_valid());
PulseAudioSink {
Self {
s: None,
ss,
device,
format,
}
}
}
@ -66,19 +79,13 @@ impl Sink for PulseAudioSink {
Ok(())
}
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
if let Some(s) = &self.s {
// SAFETY: An i16 consists of two bytes, so that the given slice can be interpreted
// as a byte array of double length. Each byte pointer is validly aligned, and so
// is the newly created slice.
let d: &[u8] = unsafe {
std::slice::from_raw_parts(
packet.samples().as_ptr() as *const u8,
packet.samples().len() * 2,
)
};
sink_as_bytes!();
}
match s.write(d) {
impl SinkAsBytes for PulseAudioSink {
fn write_bytes(&mut self, data: &[u8]) -> io::Result<()> {
if let Some(s) = &self.s {
match s.write(data) {
Ok(_) => Ok(()),
Err(e) => Err(io::Error::new(
io::ErrorKind::BrokenPipe,
@ -88,7 +95,7 @@ impl Sink for PulseAudioSink {
} else {
Err(io::Error::new(
io::ErrorKind::NotConnected,
"Not connected to pulseaudio",
"Not connected to PulseAudio",
))
}
}

61
playback/src/audio_backend/rodio.rs
View file

@ -5,7 +5,9 @@ use cpal::traits::{DeviceTrait, HostTrait};
use thiserror::Error;
use super::Sink;
use crate::audio::AudioPacket;
use crate::audio::{AudioPacket, SamplesConverter};
use crate::config::AudioFormat;
use crate::player::{NUM_CHANNELS, SAMPLE_RATE};
#[cfg(all(
feature = "rodiojack-backend",
@ -14,15 +16,16 @@ use crate::audio::AudioPacket;
compile_error!("Rodio JACK backend is currently only supported on linux.");
#[cfg(feature = "rodio-backend")]
pub fn mk_rodio(device: Option<String>) -> Box<dyn Sink> {
Box::new(open(cpal::default_host(), device))
pub fn mk_rodio(device: Option<String>, format: AudioFormat) -> Box<dyn Sink> {
Box::new(open(cpal::default_host(), device, format))
}
#[cfg(feature = "rodiojack-backend")]
pub fn mk_rodiojack(device: Option<String>) -> Box<dyn Sink> {
pub fn mk_rodiojack(device: Option<String>, format: AudioFormat) -> Box<dyn Sink> {
Box::new(open(
cpal::host_from_id(cpal::HostId::Jack).unwrap(),
device,
format,
))
}
@ -42,6 +45,7 @@ pub enum RodioError {
pub struct RodioSink {
rodio_sink: rodio::Sink,
format: AudioFormat,
_stream: rodio::OutputStream,
}
@ -147,33 +151,54 @@ fn create_sink(
Ok((sink, stream))
}
pub fn open(host: cpal::Host, device: Option<String>) -> RodioSink {
debug!("Using rodio sink with cpal host: {}", host.id().name());
pub fn open(host: cpal::Host, device: Option<String>, format: AudioFormat) -> RodioSink {
debug!(
"Using rodio sink with format {:?} and cpal host: {}",
format,
host.id().name()
);
match format {
AudioFormat::F32 => {
#[cfg(target_os = "linux")]
warn!("Rodio output to Alsa is known to cause garbled sound, consider using `--backend alsa`")
}
AudioFormat::S16 => (),
_ => unimplemented!("Rodio currently only supports F32 and S16 formats"),
}
let (sink, stream) = create_sink(&host, device).unwrap();
debug!("Rodio sink was created");
RodioSink {
rodio_sink: sink,
format,
_stream: stream,
}
}
impl Sink for RodioSink {
fn start(&mut self) -> io::Result<()> {
// More similar to an "unpause" than "play". Doesn't undo "stop".
// self.rodio_sink.play();
Ok(())
}
fn stop(&mut self) -> io::Result<()> {
// This will immediately stop playback, but the sink is then unusable.
// We just have to let the current buffer play till the end.
// self.rodio_sink.stop();
Ok(())
}
start_stop_noop!();
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
let samples = packet.samples();
match self.format {
AudioFormat::F32 => {
let source =
rodio::buffer::SamplesBuffer::new(NUM_CHANNELS as u16, SAMPLE_RATE, samples);
self.rodio_sink.append(source);
}
AudioFormat::S16 => {
let samples_s16: &[i16] = &SamplesConverter::to_s16(samples);
let source = rodio::buffer::SamplesBuffer::new(
NUM_CHANNELS as u16,
SAMPLE_RATE,
samples_s16,
);
self.rodio_sink.append(source);
}
_ => unreachable!(),
};
let source = rodio::buffer::SamplesBuffer::new(2, 44100, packet.samples());
self.rodio_sink.append(source);

103
playback/src/audio_backend/sdl.rs
View file

@ -1,57 +1,112 @@
use super::{Open, Sink};
use crate::audio::AudioPacket;
use crate::audio::{AudioPacket, SamplesConverter};
use crate::config::AudioFormat;
use crate::player::{NUM_CHANNELS, SAMPLE_RATE};
use sdl2::audio::{AudioQueue, AudioSpecDesired};
use std::{io, thread, time};
type Channel = i16;
pub struct SdlSink {
queue: AudioQueue<Channel>,
pub enum SdlSink {
F32(AudioQueue<f32>),
S32(AudioQueue<i32>),
S16(AudioQueue<i16>),
}
impl Open for SdlSink {
fn open(device: Option<String>) -> SdlSink {
debug!("Using SDL sink");
fn open(device: Option<String>, format: AudioFormat) -> Self {
info!("Using SDL sink with format: {:?}", format);
if device.is_some() {
panic!("SDL sink does not support specifying a device name");
warn!("SDL sink does not support specifying a device name");
}
let ctx = sdl2::init().expect("Could not init SDL");
let audio = ctx.audio().expect("Could not init SDL audio subsystem");
let ctx = sdl2::init().expect("could not initialize SDL");
let audio = ctx
.audio()
.expect("could not initialize SDL audio subsystem");
let desired_spec = AudioSpecDesired {
freq: Some(44_100),
channels: Some(2),
freq: Some(SAMPLE_RATE as i32),
channels: Some(NUM_CHANNELS),
samples: None,
};
let queue = audio
.open_queue(None, &desired_spec)
.expect("Could not open SDL audio device");
SdlSink { queue }
macro_rules! open_sink {
($sink: expr, $type: ty) => {{
let queue: AudioQueue<$type> = audio
.open_queue(None, &desired_spec)
.expect("could not open SDL audio device");
$sink(queue)
}};
}
match format {
AudioFormat::F32 => open_sink!(Self::F32, f32),
AudioFormat::S32 => open_sink!(Self::S32, i32),
AudioFormat::S16 => open_sink!(Self::S16, i16),
_ => {
unimplemented!("SDL currently does not support {:?} output", format)
}
}
}
}
impl Sink for SdlSink {
fn start(&mut self) -> io::Result<()> {
self.queue.clear();
self.queue.resume();
macro_rules! start_sink {
($queue: expr) => {{
$queue.clear();
$queue.resume();
}};
}
match self {
Self::F32(queue) => start_sink!(queue),
Self::S32(queue) => start_sink!(queue),
Self::S16(queue) => start_sink!(queue),
};
Ok(())
}
fn stop(&mut self) -> io::Result<()> {
self.queue.pause();
self.queue.clear();
macro_rules! stop_sink {
($queue: expr) => {{
$queue.pause();
$queue.clear();
}};
}
match self {
Self::F32(queue) => stop_sink!(queue),
Self::S32(queue) => stop_sink!(queue),
Self::S16(queue) => stop_sink!(queue),
};
Ok(())
}
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
while self.queue.size() > (2 * 2 * 44_100) {
// sleep and wait for sdl thread to drain the queue a bit
thread::sleep(time::Duration::from_millis(10));
macro_rules! drain_sink {
($queue: expr, $size: expr) => {{
// sleep and wait for sdl thread to drain the queue a bit
while $queue.size() > (NUM_CHANNELS as u32 * $size as u32 * SAMPLE_RATE) {
thread::sleep(time::Duration::from_millis(10));
}
}};
}
self.queue.queue(packet.samples());
let samples = packet.samples();
match self {
Self::F32(queue) => {
drain_sink!(queue, AudioFormat::F32.size());
queue.queue(samples)
}
Self::S32(queue) => {
let samples_s32: &[i32] = &SamplesConverter::to_s32(samples);
drain_sink!(queue, AudioFormat::S32.size());
queue.queue(samples_s32)
}
Self::S16(queue) => {
let samples_s16: &[i16] = &SamplesConverter::to_s16(samples);
drain_sink!(queue, AudioFormat::S16.size());
queue.queue(samples_s16)
}
};
Ok(())
}
}

25
playback/src/audio_backend/subprocess.rs
View file

@ -1,24 +1,26 @@
use super::{Open, Sink};
use super::{Open, Sink, SinkAsBytes};
use crate::audio::AudioPacket;
use crate::config::AudioFormat;
use shell_words::split;
use std::io::{self, Write};
use std::mem;
use std::process::{Child, Command, Stdio};
use std::slice;
pub struct SubprocessSink {
shell_command: String,
child: Option<Child>,
format: AudioFormat,
}
impl Open for SubprocessSink {
fn open(shell_command: Option<String>) -> SubprocessSink {
fn open(shell_command: Option<String>, format: AudioFormat) -> Self {
info!("Using subprocess sink with format: {:?}", format);
if let Some(shell_command) = shell_command {
SubprocessSink {
shell_command,
child: None,
format,
}
} else {
panic!("subprocess sink requires specifying a shell command");
@ -46,16 +48,15 @@ impl Sink for SubprocessSink {
Ok(())
}
fn write(&mut self, packet: &AudioPacket) -> io::Result<()> {
let data: &[u8] = unsafe {
slice::from_raw_parts(
packet.samples().as_ptr() as *const u8,
packet.samples().len() * mem::size_of::<i16>(),
)
};
sink_as_bytes!();
}
impl SinkAsBytes for SubprocessSink {
fn write_bytes(&mut self, data: &[u8]) -> io::Result<()> {
if let Some(child) = &mut self.child {
let child_stdin = child.stdin.as_mut().unwrap();
child_stdin.write_all(data)?;
child_stdin.flush()?;
}
Ok(())
}