monorepo/music-player/server/src/playback.rs

use crate::{media::TrackId, Error, FatalError};
use flow::{ok, return_error, Flow};
use gstreamer::{format::ClockTime, prelude::*, MessageView, StateChangeError};
use std::{
    path::PathBuf,
    thread::{self, JoinHandle},
    time::Duration,
};
use tokio::sync::mpsc::{channel, Receiver, Sender};
use urlencoding::encode;

pub enum PlaybackControl {
    PlayTrack(TrackId),
    Stop,
    Exit,
}

pub enum PlaybackStatus {
    Stopped,
}

pub struct Playback {
    events_handle: tokio::task::JoinHandle<()>,
    pipeline: gstreamer::Element,
}

impl Playback {
    pub fn new(id: TrackId) -> Result<Self, Error> {
        let pb = PathBuf::from(id.as_ref());
        let path = pb
            .iter()
            .skip(1)
            .map(|component| encode(&component.to_string_lossy()).into_owned())
            .collect::<PathBuf>();
        let pipeline = gstreamer::parse_launch(&format!("playbin uri=file:///{}", path.display()))
            .map_err(|err| Error::GlibError(err))?;

        pipeline
            .set_state(gstreamer::State::Playing)
            .map_err(|err| Error::CannotPlay(err.to_string()))?;
        let events_handle = tokio::task::spawn_blocking({
            let pipeline = pipeline.clone();
            move || pipeline_events(pipeline)
        });

        Ok(Self {
            events_handle,
            pipeline,
        })
    }

    pub fn stop(&self) -> Result<(), Error> {
        self.events_handle.abort();
        self.pipeline
            .set_state(gstreamer::State::Paused)
            .map_err(|_| Error::CannotStop)?;
        Ok(())
    }

    pub fn position(&self) -> (Option<ClockTime>, Option<ClockTime>) {
        let position = self.pipeline.query_position();
        let duration = self.pipeline.query_duration();
        (position, duration)
    }
}

impl Drop for Playback {
    fn drop(&mut self) {
        self.stop();
    }
}

fn pipeline_events(pipeline: gstreamer::Element) {
    let bus = pipeline.bus().unwrap();
    for msg in bus.iter_timed(gstreamer::ClockTime::NONE) {
        match msg.view() {
            MessageView::Eos(_) => (),
            MessageView::Error(err) => {
                println!(
                    "Error from {:?}: {} ({:?})",
                    err.src().map(|s| s.path_string()),
                    err.error(),
                    err.debug()
                );
            }
            msg => println!("{:?}", msg),
        }
    }
}