Skip to content

Commit

Permalink
Fix compile errors in backends
Browse files Browse the repository at this point in the history
fe37186 added the restriction that `Sink`s must be `Send`. It turned
out later that this restrictions was unnecessary, and since some
`Sink`s aren't `Send` yet, this restriction is lifted again.

librespot-org#601 refactored the `RodioSink` in order to make
it `Send`. These changes are partly reverted in favour of the initial
simpler design.

Furthermore, there were some compile errors in the gstreamer backend
which are hereby fixed.
  • Loading branch information
Johannesd3 committed Apr 10, 2021
1 parent 9a3a666 commit 5435ab3
Show file tree
Hide file tree
Showing 5 changed files with 18 additions and 36 deletions.
15 changes: 8 additions & 7 deletions playback/src/audio_backend/gstreamer.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
use super::{Open, Sink};
use crate::audio::AudioPacket;

use gst::prelude::*;
use gst::*;
use gstreamer as gst;
use gstreamer_app as gst_app;
use zerocopy::*;

use std::sync::mpsc::{sync_channel, SyncSender};
Expand Down Expand Up @@ -52,14 +54,13 @@ impl Open for GstreamerSink {
thread::spawn(move || {
for data in rx {
let buffer = bufferpool.acquire_buffer(None);
if !buffer.is_err() {
let mut okbuffer = buffer.unwrap();
let mutbuf = okbuffer.make_mut();
if let Ok(mut buffer) = buffer {
let mutbuf = buffer.make_mut();
mutbuf.set_size(data.len());
mutbuf
.copy_from_slice(0, data.as_bytes())
.expect("Failed to copy from slice");
let _eat = appsrc.push_buffer(okbuffer);
let _eat = appsrc.push_buffer(buffer);
}
}
});
Expand All @@ -69,8 +70,8 @@ impl Open for GstreamerSink {
let watch_mainloop = thread_mainloop.clone();
bus.add_watch(move |_, msg| {
match msg.view() {
MessageView::Eos(..) => watch_mainloop.quit(),
MessageView::Error(err) => {
gst::MessageView::Eos(..) => watch_mainloop.quit(),
gst::MessageView::Error(err) => {
println!(
"Error from {:?}: {} ({:?})",
err.get_src().map(|s| s.get_path_string()),
Expand Down
4 changes: 2 additions & 2 deletions playback/src/audio_backend/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,9 @@ pub trait Sink {
fn write(&mut self, packet: &AudioPacket) -> io::Result<()>;
}

pub type SinkBuilder = fn(Option<String>) -> Box<dyn Sink + Send>;
pub type SinkBuilder = fn(Option<String>) -> Box<dyn Sink>;

fn mk_sink<S: Sink + Open + Send + 'static>(device: Option<String>) -> Box<dyn Sink + Send> {
fn mk_sink<S: Sink + Open + 'static>(device: Option<String>) -> Box<dyn Sink> {
Box::new(S::open(device))
}

Expand Down
29 changes: 5 additions & 24 deletions playback/src/audio_backend/rodio.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
use std::process::exit;
use std::{convert::Infallible, sync::mpsc};
use std::{io, thread, time};

use cpal::traits::{DeviceTrait, HostTrait};
Expand All @@ -15,12 +14,12 @@ 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 + Send> {
pub fn mk_rodio(device: Option<String>) -> Box<dyn Sink> {
Box::new(open(cpal::default_host(), device))
}

#[cfg(feature = "rodiojack-backend")]
pub fn mk_rodiojack(device: Option<String>) -> Box<dyn Sink + Send> {
pub fn mk_rodiojack(device: Option<String>) -> Box<dyn Sink> {
Box::new(open(
cpal::host_from_id(cpal::HostId::Jack).unwrap(),
device,
Expand All @@ -43,8 +42,7 @@ pub enum RodioError {

pub struct RodioSink {
rodio_sink: rodio::Sink,
// will produce a TryRecvError on the receiver side when it is dropped.
_close_tx: mpsc::SyncSender<Infallible>,
_stream: rodio::OutputStream,
}

fn list_formats(device: &rodio::Device) {
Expand Down Expand Up @@ -152,29 +150,12 @@ fn create_sink(
pub fn open(host: cpal::Host, device: Option<String>) -> RodioSink {
debug!("Using rodio sink with cpal host: {}", host.id().name());

let (sink_tx, sink_rx) = mpsc::sync_channel(1);
let (close_tx, close_rx) = mpsc::sync_channel(1);

std::thread::spawn(move || match create_sink(&host, device) {
Ok((sink, stream)) => {
sink_tx.send(Ok(sink)).unwrap();

close_rx.recv().unwrap_err(); // This will fail as soon as the sender is dropped
debug!("drop rodio::OutputStream");
drop(stream);
}
Err(e) => {
sink_tx.send(Err(e)).unwrap();
}
});

// Instead of the second `unwrap`, better error handling could be introduced
let sink = sink_rx.recv().unwrap().unwrap();
let (sink, stream) = create_sink(&host, device).unwrap();

debug!("Rodio sink was created");
RodioSink {
rodio_sink: sink,
_close_tx: close_tx,
_stream: stream,
}
}

Expand Down
4 changes: 2 additions & 2 deletions playback/src/player.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ struct PlayerInternal {

state: PlayerState,
preload: PlayerPreload,
sink: Box<dyn Sink + Send>,
sink: Box<dyn Sink>,
sink_status: SinkStatus,
sink_event_callback: Option<SinkEventCallback>,
audio_filter: Option<Box<dyn AudioFilter + Send>>,
Expand Down Expand Up @@ -242,7 +242,7 @@ impl Player {
sink_builder: F,
) -> (Player, PlayerEventChannel)
where
F: FnOnce() -> Box<dyn Sink + Send> + Send + 'static,
F: FnOnce() -> Box<dyn Sink> + Send + 'static,
{
let (cmd_tx, cmd_rx) = mpsc::unbounded_channel();
let (event_sender, event_receiver) = mpsc::unbounded_channel();
Expand Down
2 changes: 1 addition & 1 deletion src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -105,7 +105,7 @@ fn print_version() {

#[derive(Clone)]
struct Setup {
backend: fn(Option<String>) -> Box<dyn Sink + Send + 'static>,
backend: fn(Option<String>) -> Box<dyn Sink + 'static>,
device: Option<String>,

mixer: fn(Option<MixerConfig>) -> Box<dyn Mixer>,
Expand Down

0 comments on commit 5435ab3

Please sign in to comment.