clusterfuck
/
crabidy
4
1
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects 1 Releases 5 Wiki Activity

Update all dependencies to current versions 

- tonic 0.9 -> 0.14 (tonic-prost/tonic-prost-build split), prost 0.14
- ratatui 0.20 -> 0.30 (Frame no longer generic, Line instead of Spans),
  crossterm 0.29
- rodio 0.17 -> 0.22: replace the custom symphonia decoder with rodio's
  built-in decoder, seeking (try_seek) and position tracking (get_pos);
  end-of-stream is now signalled via an EmptyCallback source with a
  generation counter so a replaced track can never emit a stale EOS
- replace the vendored stream-download crate with the published
  stream-download 0.24 (rustls), with a 30s open timeout
- reqwest 0.12->0.13 (rustls/webpki-roots/query features), base64 0.22
  Engine API, rand 0.10, flume 0.12, thiserror 2, dirs 6, toml 1
- unify everything under [workspace.dependencies]; drop unused deps
  (once_cell, serde_json in server; confique, secrecy in tidaldy)
- devenv: add protobuf (protoc) for prost-build

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
AI User 2026-07-19 21:41:10 +02:00
parent 91716daf84
commit 6eb5a87b15
33 changed files with 3283 additions and 2682 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4201
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

61
Cargo.toml
View File

@ -1,9 +1,68 @@
[workspace]
resolver = "2"
members = [
"audio-player",
"cbd-tui",
"crabidy-core",
"crabidy-server",
"stream-download",
"tidaldy",
]
[workspace.package]
version = "0.1.0"
edition = "2021"
[workspace.dependencies]
anyhow = "1"
async-trait = "0.1"
base64 = "0.22"
chrono = { version = "0.4", default-features = false, features = ["clock"] }
clap = { version = "4", features = ["derive"] }
clap-serde-derive = "0.2"
crossterm = "0.29"
dirs = "6"
flume = "0.12"
futures = "0.3"
log = "0.4"
notify-rust = "4"
prost = "0.14"
rand = "0.10"
ratatui = "0.30"
reqwest = { version = "0.13", default-features = false, features = [
"json",
"query",
"rustls",
"webpki-roots",
"http2",
"hickory-dns",
"stream",
] }
rodio = { version = "0.22", default-features = false, features = [
"playback",
"symphonia-all",
] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
serde_urlencoded = "0.7"
stream-download = { version = "0.24", default-features = false, features = [
"reqwest",
"reqwest-rustls",
"temp-storage",
] }
thiserror = "2"
tokio = "1"
tokio-stream = "0.1"
toml = "1"
tonic = "0.14"
tonic-prost = "0.14"
tonic-prost-build = "0.14"
tracing = "0.1"
tracing-appender = "0.2"
tracing-log = "0.2"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
url = "2"
# Local crates
audio-player = { path = "audio-player" }
crabidy-core = { path = "crabidy-core" }
tidaldy = { path = "tidaldy" }

24
audio-player/Cargo.toml
View File

@ -1,19 +1,17 @@
[package]
name = "audio-player"
version = "0.1.0"
edition = "2021"
version.workspace = true
edition.workspace = true
[dependencies]
rodio = { version = "0.17.1", default-features = false, features = [
"symphonia-all",
] }
symphonia = { version = "0.5.3", features = ["all"] }
stream-download = { path = "../stream-download" }
anyhow = "1.0.71"
url = "2.4.0"
flume = "0.10.14"
thiserror = "1.0.40"
tracing = "0.1.37"
anyhow.workspace = true
flume.workspace = true
rodio.workspace = true
stream-download.workspace = true
thiserror.workspace = true
tokio = { workspace = true, features = ["rt", "rt-multi-thread"] }
tracing.workspace = true
url.workspace = true
[dev-dependencies]
tokio = { version = "1", features = ["full"] }
tokio = { workspace = true, features = ["full"] }

318
audio-player/src/decoder.rs
View File

@ -1,318 +0,0 @@
use std::error::Error;
use std::fmt;
use std::time::Duration;
use flume::Sender;
use rodio::Source;
use symphonia::{
core::{
audio::{AudioBufferRef, SampleBuffer, SignalSpec},
codecs::{Decoder, DecoderOptions},
errors::Error as SymphoniaError,
formats::{FormatOptions, FormatReader, SeekMode, SeekTo, Track},
io::MediaSourceStream,
meta::{MetadataOptions, MetadataRevision},
probe::Hint,
units::{Time, TimeBase},
},
default::get_probe,
};
use tracing::warn;
use crate::player_engine::PlayerEngineCommand;
// Decoder errors are not considered fatal.
// The correct action is to just get a new packet and try again.
// But a decode error in more than 3 consecutive packets is fatal.
const MAX_DECODE_ERRORS: usize = 3;
#[derive(Clone)]
pub struct MediaInfo {
pub duration: Option<Duration>,
pub metadata: Option<MetadataRevision>,
pub track: Track,
}
pub struct SymphoniaDecoder {
decoder: Box<dyn Decoder>,
current_frame_offset: usize,
format: Box<dyn FormatReader>,
buffer: SampleBuffer<i16>,
spec: SignalSpec,
time_base: Option<TimeBase>,
duration: u64,
elapsed: u64,
metadata: Option<MetadataRevision>,
track: Track,
tx: Sender<PlayerEngineCommand>,
}
impl SymphoniaDecoder {
pub fn new(
mss: MediaSourceStream,
hint: Hint,
tx: Sender<PlayerEngineCommand>,
) -> Result<Self, DecoderError> {
match SymphoniaDecoder::init(mss, hint, tx) {
Err(e) => match e {
SymphoniaError::IoError(e) => Err(DecoderError::IoError(e.to_string())),
SymphoniaError::DecodeError(e) => Err(DecoderError::DecodeError(e)),
SymphoniaError::SeekError(_) => {
unreachable!("Seek errors should not occur during initialization")
}
SymphoniaError::Unsupported(_) => Err(DecoderError::UnrecognizedFormat),
SymphoniaError::LimitError(e) => Err(DecoderError::LimitError(e)),
SymphoniaError::ResetRequired => Err(DecoderError::ResetRequired),
},
Ok(Some(decoder)) => Ok(decoder),
Ok(None) => Err(DecoderError::NoStreams),
}
}
fn init(
mss: MediaSourceStream,
hint: Hint,
tx: Sender<PlayerEngineCommand>,
) -> symphonia::core::errors::Result<Option<SymphoniaDecoder>> {
let format_opts: FormatOptions = FormatOptions {
enable_gapless: true,
..Default::default()
};
let metadata_opts: MetadataOptions = Default::default();
let mut probed = get_probe().format(&hint, mss, &format_opts, &metadata_opts)?;
let track = match probed.format.default_track() {
Some(stream) => stream,
None => return Ok(None),
}
.clone();
let time_base = track.codec_params.time_base;
let duration = track
.codec_params
.n_frames
.map(|frames| track.codec_params.start_ts + frames)
.unwrap_or_default();
let mut _elapsed = 0;
let mut decoder = symphonia::default::get_codecs()
.make(&track.codec_params, &DecoderOptions { verify: true })?;
let mut decode_errors: usize = 0;
let decoded = loop {
let current_frame = probed.format.next_packet()?;
_elapsed = current_frame.ts();
match decoder.decode(&current_frame) {
Ok(decoded) => break decoded,
Err(e) => match e {
SymphoniaError::DecodeError(_) => {
decode_errors += 1;
if decode_errors > MAX_DECODE_ERRORS {
return Err(e);
} else {
continue;
}
}
_ => return Err(e),
},
}
};
let spec = decoded.spec().to_owned();
let buffer = SymphoniaDecoder::get_buffer(decoded, &spec);
// Prefer metadata that's provided in the container format, over other tags found during the
// probe operation.
let metadata = probed.format.metadata().current().cloned().or_else(|| {
probed
.metadata
.get()
.as_ref()
.and_then(|m| m.current().cloned())
});
Ok(Some(SymphoniaDecoder {
decoder,
current_frame_offset: 0,
format: probed.format,
buffer,
spec,
time_base,
duration,
elapsed: _elapsed,
metadata,
track,
tx,
}))
}
#[inline]
pub fn media_info(&self) -> MediaInfo {
MediaInfo {
duration: self.total_duration(),
metadata: self.metadata.clone(),
track: self.track.clone(),
}
}
#[inline]
pub fn elapsed(&self) -> Duration {
if let Some(tb) = self.time_base {
let time = tb.calc_time(self.elapsed);
return Duration::from_secs_f64(time.seconds as f64 + time.frac);
};
Duration::default()
}
#[inline]
pub fn seek(&mut self, time: Duration) -> Option<Duration> {
let nanos_per_sec = 1_000_000_000.0;
match self.format.seek(
SeekMode::Coarse,
SeekTo::Time {
time: Time::new(
time.as_secs(),
f64::from(time.subsec_nanos()) / nanos_per_sec,
),
track_id: None,
},
) {
Ok(seeked_to) => {
let base = TimeBase::new(1, self.sample_rate());
let time = base.calc_time(seeked_to.actual_ts);
Some(Duration::from_millis(
time.seconds * 1000 + ((time.frac * 60. * 1000.).round() as u64),
))
}
Err(_) => None,
}
}
#[inline]
fn get_buffer(decoded: AudioBufferRef, spec: &SignalSpec) -> SampleBuffer<i16> {
let duration = decoded.capacity() as u64;
let mut buffer = SampleBuffer::<i16>::new(duration, *spec);
buffer.copy_interleaved_ref(decoded);
buffer
}
}
impl Source for SymphoniaDecoder {
#[inline]
fn current_frame_len(&self) -> Option<usize> {
Some(self.buffer.samples().len())
}
#[inline]
fn channels(&self) -> u16 {
self.spec.channels.count() as u16
}
#[inline]
fn sample_rate(&self) -> u32 {
self.spec.rate
}
#[inline]
fn total_duration(&self) -> Option<Duration> {
match self.time_base {
Some(tb) => {
let time = tb.calc_time(self.duration);
Some(Duration::from_secs_f64(time.seconds as f64 + time.frac))
}
None => None,
}
}
}
impl Iterator for SymphoniaDecoder {
type Item = i16;
#[inline]
fn next(&mut self) -> Option<i16> {
if self.current_frame_offset == self.buffer.len() {
let mut decode_errors: usize = 0;
let decoded = loop {
match self.format.next_packet() {
Ok(packet) => {
self.elapsed = packet.ts();
match self.decoder.decode(&packet) {
Ok(decoded) => break decoded,
Err(e) => match e {
SymphoniaError::DecodeError(_) => {
decode_errors += 1;
if decode_errors > MAX_DECODE_ERRORS {
return None;
} else {
continue;
}
}
_ => return None,
},
}
}
Err(SymphoniaError::IoError(err)) => {
if err.kind() == std::io::ErrorKind::UnexpectedEof
&& err.to_string() == "end of stream"
{
self.tx
.send(PlayerEngineCommand::Eos)
.unwrap_or_else(|e| warn!("Send error {}", e));
return None;
}
}
Err(_) => return None,
}
};
self.spec = decoded.spec().to_owned();
self.buffer = SymphoniaDecoder::get_buffer(decoded, &self.spec);
self.current_frame_offset = 0;
}
let sample = *self.buffer.samples().get(self.current_frame_offset)?;
self.current_frame_offset += 1;
Some(sample)
}
}
/// Error that can happen when creating a decoder.
#[derive(Debug, Clone)]
pub enum DecoderError {
/// The format of the data has not been recognized.
UnrecognizedFormat,
/// An IO error occurred while reading, writing, or seeking the stream.
IoError(String),
/// The stream contained malformed data and could not be decoded or demuxed.
DecodeError(&'static str),
/// A default or user-defined limit was reached while decoding or demuxing the stream. Limits
/// are used to prevent denial-of-service attacks from malicious streams.
LimitError(&'static str),
/// The demuxer or decoder needs to be reset before continuing.
ResetRequired,
/// No streams were found by the decoder
NoStreams,
}
impl fmt::Display for DecoderError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let text = match self {
DecoderError::UnrecognizedFormat => "Unrecognized format",
DecoderError::IoError(msg) => &msg[..],
DecoderError::DecodeError(msg) => msg,
DecoderError::LimitError(msg) => msg,
DecoderError::ResetRequired => "Reset required",
DecoderError::NoStreams => "No streams",
};
write!(f, "{}", text)
}
}
impl Error for DecoderError {}

4
audio-player/src/lib.rs
View File

@ -1,7 +1,5 @@
mod decoder;
mod player;
mod player_engine;
pub use decoder::MediaInfo;
pub use player::{Player, PlayerError};
pub use player_engine::PlayerMessage;
pub use player_engine::{MediaInfo, PlayerMessage};

130
audio-player/src/player.rs
View File

@ -3,13 +3,9 @@ use std::time::Duration;
use anyhow::Result;
use flume::{Receiver, Sender};
use tracing::{error, warn};
use tracing::error;
use crate::decoder::MediaInfo;
use crate::player_engine::{PlayerEngine, PlayerEngineCommand, PlayerMessage};
// TODO:
// * Emit buffering
use crate::player_engine::{MediaInfo, PlayerEngine, PlayerEngineCommand, PlayerMessage};
pub enum PlayerError {}
@ -20,82 +16,24 @@ pub struct Player {
impl Default for Player {
fn default() -> Self {
let (tx_engine, rx_engine) = flume::bounded(10);
let (tx_engine, rx_engine) = flume::bounded(16);
let (tx_player, messages): (Sender<PlayerMessage>, Receiver<PlayerMessage>) =
flume::bounded(10);
flume::bounded(16);
let tx_decoder = tx_engine.clone();
let tx_callbacks = tx_engine.clone();
// Capture the runtime handle here: the engine thread itself is not a
// tokio context but needs one to create http streams.
let runtime = tokio::runtime::Handle::try_current().ok();
thread::spawn(move || {
let mut player = match PlayerEngine::init(tx_decoder, tx_player) {
let engine = match PlayerEngine::init(tx_callbacks, tx_player, runtime) {
Err(e) => {
error!("Could not initialize player: {}", e);
return;
}
Ok(engine) => engine,
};
loop {
match rx_engine.recv() {
Ok(PlayerEngineCommand::Play(source_str, tx)) => {
tx.send(player.play(&source_str))
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::Pause(tx)) => {
tx.send(player.pause())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::Unpause(tx)) => {
tx.send(player.unpause())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::Stop(tx)) => {
tx.send(player.stop())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::TogglePlay(tx)) => {
tx.send(player.toggle_play())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::Restart(tx)) => {
tx.send(player.restart())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::GetDuration(tx)) => {
tx.send(player.duration())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::GetElapsed(tx)) => {
tx.send(player.elapsed())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::SeekTo(time, tx)) => {
tx.send(player.seek_to(time))
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::GetVolume(tx)) => {
tx.send(player.volume())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::GetPaused(tx)) => {
tx.send(player.is_paused())
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::SetVolume(volume, tx)) => {
tx.send(player.set_volume(volume))
.unwrap_or_else(|e| warn!("Send error {}", e));
}
Ok(PlayerEngineCommand::SetElapsed(elapsed)) => {
player.handle_elapsed(elapsed);
}
Ok(PlayerEngineCommand::Eos) => {
player.handle_eos();
}
Err(e) => {
warn!("Recv error {}", e);
}
}
}
engine.run(rx_engine);
});
Self {
@ -109,74 +47,96 @@ impl Player {
pub async fn play(&self, source_str: &str) -> Result<MediaInfo> {
let (tx, rx) = flume::bounded(1);
self.tx_engine
.send(PlayerEngineCommand::Play(source_str.to_string(), tx))?;
.send_async(PlayerEngineCommand::Play(source_str.to_string(), tx))
.await?;
rx.recv_async().await?
}
pub async fn restart(&self) -> Result<MediaInfo> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::Restart(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::Restart(tx))
.await?;
rx.recv_async().await?
}
pub async fn elpased(&self) -> Result<Duration> {
pub async fn elapsed(&self) -> Result<Duration> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::GetElapsed(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::GetElapsed(tx))
.await?;
rx.recv_async().await?
}
pub async fn duration(&self) -> Result<Duration> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::GetDuration(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::GetDuration(tx))
.await?;
rx.recv_async().await?
}
pub async fn seek_to(&self, time: Duration) -> Result<Duration> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::SeekTo(time, tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::SeekTo(time, tx))
.await?;
rx.recv_async().await?
}
pub async fn volume(&self) -> Result<f32> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::GetVolume(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::GetVolume(tx))
.await?;
Ok(rx.recv_async().await?)
}
pub async fn is_paused(&self) -> Result<bool> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::GetPaused(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::GetPaused(tx))
.await?;
rx.recv_async().await?
}
pub async fn set_volume(&self, volume: f32) -> Result<f32> {
let (tx, rx) = flume::bounded(1);
self.tx_engine
.send(PlayerEngineCommand::SetVolume(volume, tx))?;
.send_async(PlayerEngineCommand::SetVolume(volume, tx))
.await?;
Ok(rx.recv_async().await?)
}
pub async fn pause(&self) -> Result<()> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::Pause(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::Pause(tx))
.await?;
rx.recv_async().await?
}
pub async fn unpause(&self) -> Result<()> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::Unpause(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::Unpause(tx))
.await?;
rx.recv_async().await?
}
pub async fn toggle_play(&self) -> Result<bool> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::TogglePlay(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::TogglePlay(tx))
.await?;
rx.recv_async().await?
}
pub async fn stop(&self) -> Result<()> {
let (tx, rx) = flume::bounded(1);
self.tx_engine.send(PlayerEngineCommand::Stop(tx))?;
self.tx_engine
.send_async(PlayerEngineCommand::Stop(tx))
.await?;
rx.recv_async().await?
}
}

336
audio-player/src/player_engine.rs
View File

@ -1,19 +1,23 @@
use flume::Sender;
use std::fs::File;
use std::io::BufReader;
use std::path::Path;
use std::sync::atomic::AtomicU64;
use std::thread;
use std::time::Duration;
use std::{fs::File, sync::atomic::Ordering};
use symphonia::core::probe::Hint;
use tracing::{debug, warn};
use anyhow::{anyhow, Context, Result};
use flume::{Receiver, RecvTimeoutError, Sender};
use rodio::source::EmptyCallback;
use rodio::stream::{DeviceSinkBuilder, MixerDeviceSink};
use rodio::{Decoder, Source};
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};
use thiserror::Error;
use tracing::{debug, info, instrument, trace, warn};
use url::Url;
use crate::decoder::{MediaInfo, SymphoniaDecoder};
use anyhow::{anyhow, Result};
use rodio::{OutputStream, OutputStreamHandle, Sink, Source};
use stream_download::StreamDownload;
use symphonia::core::io::{MediaSource, MediaSourceStream, MediaSourceStreamOptions};
use thiserror::Error;
/// How long we wait for the initial prefetch of a network stream.
const STREAM_OPEN_TIMEOUT: Duration = Duration::from_secs(30);
/// Interval between elapsed-position updates while playing.
const TICK_INTERVAL: Duration = Duration::from_millis(250);
pub enum PlayerEngineCommand {
Play(String, Sender<Result<MediaInfo>>),
@ -28,8 +32,10 @@ pub enum PlayerEngineCommand {
SeekTo(Duration, Sender<Result<Duration>>),
GetVolume(Sender<f32>),
GetPaused(Sender<Result<bool>>),
Eos,
SetElapsed(Duration),
/// End of stream for the source started by the given generation.
/// Stale generations are ignored so an old track finishing can never
/// interfere with a newly started one.
Eos(u64),
}
pub enum PlayerMessage {
@ -46,8 +52,10 @@ pub enum PlayerMessage {
EndOfStream,
}
// TODO:
// * Emit buffering
#[derive(Clone, Debug)]
pub struct MediaInfo {
pub duration: Option<Duration>,
}
#[derive(Debug, Error)]
pub enum PlayerEngineError {
@ -55,85 +63,195 @@ pub enum PlayerEngineError {
NotPlaying,
}
// Used for seeking in the stream
static SEEK_TO: AtomicU64 = AtomicU64::new(0);
pub struct PlayerEngine {
elapsed: Duration,
current_source: Option<String>,
media_info: Option<MediaInfo>,
sink: Sink,
// We need to keep the stream around as it will stop playing when it's dropped
_stream: OutputStream,
_handle: OutputStreamHandle,
/// Monotonically increasing id for the currently playing source. Used to
/// discard end-of-stream callbacks from sources that were replaced.
generation: u64,
sink: rodio::Player,
// We need to keep the device sink around; audio stops when it's dropped.
_stream: MixerDeviceSink,
tx_engine: Sender<PlayerEngineCommand>,
tx_player: Sender<PlayerMessage>,
runtime: tokio::runtime::Handle,
// Present when the engine had to bring its own runtime because the
// creating thread was not inside a tokio context.
_owned_runtime: Option<tokio::runtime::Runtime>,
}
impl PlayerEngine {
pub fn init(
tx_engine: Sender<PlayerEngineCommand>,
tx_player: Sender<PlayerMessage>,
runtime: Option<tokio::runtime::Handle>,
) -> Result<Self> {
let (_stream, handle) = OutputStream::try_default()?;
let sink = Sink::try_new(&handle)?;
let stream =
DeviceSinkBuilder::open_default_sink().context("failed to open audio output device")?;
let sink = rodio::Player::connect_new(stream.mixer());
let (runtime, owned_runtime) = match runtime {
Some(handle) => (handle, None),
None => {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.enable_all()
.build()
.context("failed to create tokio runtime for the player engine")?;
(rt.handle().clone(), Some(rt))
}
};
info!("audio output device opened");
Ok(Self {
current_source: None,
media_info: None,
elapsed: Duration::default(),
generation: 0,
sink,
_stream,
_handle: handle,
_stream: stream,
tx_engine,
tx_player,
runtime,
_owned_runtime: owned_runtime,
})
}
pub fn play(&mut self, source_str: &str) -> Result<MediaInfo> {
let tx_player = self.tx_player.clone();
let tx_engine = self.tx_engine.clone();
/// Drives the engine until all command senders are dropped.
pub fn run(mut self, rx_engine: Receiver<PlayerEngineCommand>) {
loop {
match rx_engine.recv_timeout(TICK_INTERVAL) {
Ok(command) => self.handle_command(command),
Err(RecvTimeoutError::Timeout) => self.tick(),
Err(RecvTimeoutError::Disconnected) => {
debug!("player engine channel closed, shutting down");
break;
}
}
}
}
fn handle_command(&mut self, command: PlayerEngineCommand) {
match command {
PlayerEngineCommand::Play(source_str, tx) => {
send_reply(tx, self.play(&source_str));
}
PlayerEngineCommand::Pause(tx) => send_reply(tx, self.pause()),
PlayerEngineCommand::Unpause(tx) => send_reply(tx, self.unpause()),
PlayerEngineCommand::Stop(tx) => send_reply(tx, self.stop()),
PlayerEngineCommand::TogglePlay(tx) => send_reply(tx, self.toggle_play()),
PlayerEngineCommand::Restart(tx) => send_reply(tx, self.restart()),
PlayerEngineCommand::GetDuration(tx) => send_reply(tx, self.duration()),
PlayerEngineCommand::GetElapsed(tx) => send_reply(tx, self.elapsed()),
PlayerEngineCommand::SeekTo(time, tx) => send_reply(tx, self.seek_to(time)),
PlayerEngineCommand::SetVolume(volume, tx) => {
send_reply(tx, self.set_volume(volume));
}
PlayerEngineCommand::GetVolume(tx) => send_reply(tx, self.volume()),
PlayerEngineCommand::GetPaused(tx) => send_reply(tx, self.is_paused()),
PlayerEngineCommand::Eos(generation) => self.handle_eos(generation),
}
}
/// Emits an elapsed-position update while a source is playing.
fn tick(&self) {
if self.sink.empty() || self.sink.is_paused() {
return;
}
let duration = self
.media_info
.as_ref()
.and_then(|m| m.duration)
.unwrap_or_default();
// Dropping a tick when the channel is full is harmless.
let _ = self.tx_player.try_send(PlayerMessage::Elapsed {
duration,
elapsed: self.sink.get_pos(),
});
}
#[instrument(skip(self))]
pub fn play(&mut self, source_str: &str) -> Result<MediaInfo> {
self.reset();
let (source, hint) = self.get_source(source_str)?;
let mss = MediaSourceStream::new(source, MediaSourceStreamOptions::default());
let decoder = SymphoniaDecoder::new(mss, hint, self.tx_engine.clone())?;
let duration = self.start_source(source_str)?;
let media_info = MediaInfo { duration };
let media_info = decoder.media_info();
let media_info_copy = media_info.clone();
let duration = media_info.duration.unwrap_or_default();
self.media_info = Some(media_info);
self.media_info = Some(media_info.clone());
self.current_source = Some(source_str.to_string());
tx_player
.send(PlayerMessage::Duration { duration })
.unwrap_or_else(|e| warn!("Send error {}", e));
// FIXME: regularly update metadata revision
let decoder = decoder.periodic_access(Duration::from_millis(250), move |src| {
let seek = SEEK_TO.load(Ordering::SeqCst);
if seek > 0 {
src.seek(Duration::from_secs(seek));
SEEK_TO.store(0, Ordering::SeqCst);
}
let elapsed = src.elapsed();
tx_engine
.send(PlayerEngineCommand::SetElapsed(elapsed))
.unwrap_or_else(|e| warn!("Send error {}", e));
tx_player
.send(PlayerMessage::Elapsed { elapsed, duration })
.unwrap_or_else(|e| warn!("Send error {}", e));
self.notify(PlayerMessage::Duration {
duration: duration.unwrap_or_default(),
});
self.sink.append(decoder);
self.sink.play();
self.notify(PlayerMessage::Playing);
debug!(duration = ?duration, "started playback");
self.tx_player
.send(PlayerMessage::Playing)
.unwrap_or_else(|e| warn!("Send error {}", e));
Ok(media_info)
}
Ok(media_info_copy)
/// Decodes the source and appends it (plus an end-of-stream callback) to
/// the sink. Returns the total duration if known.
fn start_source(&mut self, source_str: &str) -> Result<Option<Duration>> {
self.generation += 1;
let duration = match Url::parse(source_str) {
Ok(url) if matches!(url.scheme(), "http" | "https") => {
trace!(%url, "opening network stream");
let reader = self.runtime.block_on(async {
tokio::time::timeout(
STREAM_OPEN_TIMEOUT,
StreamDownload::new_http(
url.clone(),
TempStorageProvider::new(),
Settings::default(),
),
)
.await
.map_err(|_| anyhow!("timed out opening stream after {STREAM_OPEN_TIMEOUT:?}"))?
.context("failed to open http stream")
})?;
let mut builder = Decoder::builder().with_data(reader).with_seekable(true);
if let Some(extension) = Path::new(url.path()).extension().and_then(|e| e.to_str())
{
builder = builder.with_hint(extension);
}
let decoder = builder.build().context("failed to decode http stream")?;
let duration = decoder.total_duration();
self.sink.append(decoder);
duration
}
Ok(url) => return Err(anyhow!("Not a valid URL scheme: {}", url.scheme())),
Err(_) => {
trace!(path = source_str, "opening local file");
let file = File::open(source_str)
.with_context(|| format!("failed to open file {source_str}"))?;
let byte_len = file.metadata().ok().map(|m| m.len());
let mut builder = Decoder::builder()
.with_data(BufReader::new(file))
.with_seekable(true);
if let Some(len) = byte_len {
builder = builder.with_byte_len(len);
}
if let Some(extension) = Path::new(source_str).extension().and_then(|e| e.to_str())
{
builder = builder.with_hint(extension);
}
let decoder = builder.build().context("failed to decode file")?;
let duration = decoder.total_duration();
self.sink.append(decoder);
duration
}
};
// Fires only when the decoder ahead of it finished naturally; a
// stop/replace clears the queue before this source is ever played.
let tx_engine = self.tx_engine.clone();
let generation = self.generation;
self.sink.append(EmptyCallback::new(Box::new(move || {
if let Err(err) = tx_engine.try_send(PlayerEngineCommand::Eos(generation)) {
warn!("failed to send end-of-stream signal: {err}");
}
})));
Ok(duration)
}
pub fn restart(&mut self) -> Result<MediaInfo> {
@ -148,9 +266,7 @@ impl PlayerEngine {
return Err(PlayerEngineError::NotPlaying.into());
}
self.sink.pause();
self.tx_player
.send(PlayerMessage::Paused)
.unwrap_or_else(|e| warn!("Send error {}", e));
self.notify(PlayerMessage::Paused);
Ok(())
}
@ -159,9 +275,7 @@ impl PlayerEngine {
return Err(PlayerEngineError::NotPlaying.into());
}
self.sink.play();
self.tx_player
.send(PlayerMessage::Playing)
.unwrap_or_else(|e| warn!("Send error {}", e));
self.notify(PlayerMessage::Playing);
Ok(())
}
@ -171,9 +285,11 @@ impl PlayerEngine {
}
if self.sink.is_paused() {
self.sink.play();
self.notify(PlayerMessage::Playing);
Ok(true)
} else {
self.sink.pause();
self.notify(PlayerMessage::Paused);
Ok(false)
}
}
@ -183,9 +299,7 @@ impl PlayerEngine {
return Err(PlayerEngineError::NotPlaying.into());
}
self.reset();
self.tx_player
.send(PlayerMessage::Stopped)
.unwrap_or_else(|e| warn!("Send error {}", e));
self.notify(PlayerMessage::Stopped);
Ok(())
}
@ -212,18 +326,16 @@ impl PlayerEngine {
if self.is_stopped() {
return Err(PlayerEngineError::NotPlaying.into());
}
Ok(self.elapsed)
Ok(self.sink.get_pos())
}
pub fn seek_to(&self, time: Duration) -> Result<Duration> {
// We can seek between 1 second and the total duration of the track
let duration = self.duration().unwrap_or(self.elapsed);
let duration = self.duration().unwrap_or_else(|_| self.sink.get_pos());
let time = time.clamp(Duration::from_secs(1), duration);
SEEK_TO.store(time.as_secs(), Ordering::SeqCst);
// FIXME: ideally we would like to return once the seeking is successful
// then return the current elapsed time
// Cond-var might be needed to sleep this (seeking takes time)
Ok(time)
self.sink
.try_seek(time)
.map_err(|err| anyhow!("seek failed: {err}"))?;
Ok(self.sink.get_pos())
}
pub fn volume(&self) -> f32 {
@ -235,54 +347,36 @@ impl PlayerEngine {
self.sink.volume()
}
pub fn handle_eos(&mut self) {
fn handle_eos(&mut self, generation: u64) {
if generation != self.generation {
debug!(
stale = generation,
current = self.generation,
"ignoring end-of-stream from replaced source"
);
return;
}
debug!("end of stream");
self.reset();
self.tx_player
.send(PlayerMessage::EndOfStream)
.unwrap_or_else(|e| warn!("Send error {}", e));
}
pub fn handle_elapsed(&mut self, elapsed: Duration) {
self.elapsed = elapsed;
self.notify(PlayerMessage::EndOfStream);
}
fn reset(&mut self) {
self.elapsed = Duration::default();
self.current_source = None;
self.sink.pause();
self.media_info = None;
self.generation += 1;
self.sink.stop();
}
fn get_source(&self, source_str: &str) -> Result<(Box<dyn MediaSource>, Hint)> {
match Url::parse(source_str) {
Ok(url) => {
if let "http" | "https" = url.scheme() {
let reader = StreamDownload::new_http(source_str.parse().unwrap());
let path = Path::new(url.path());
let hint = self.get_hint(path);
Ok((Box::new(reader), hint))
} else {
Err(anyhow!("Not a valid URL scheme: {}", url.scheme()))
}
}
Err(_) => {
let path = Path::new(source_str);
let hint = self.get_hint(path);
Ok((Box::new(File::open(path)?), hint))
}
}
}
fn get_hint(&self, path: &Path) -> Hint {
// Create a hint to help the format registry guess what format reader is appropriate.
let mut hint = Hint::new();
// Provide the file extension as a hint.
if let Some(extension) = path.extension() {
if let Some(extension_str) = extension.to_str() {
hint.with_extension(extension_str);
}
}
hint
fn notify(&self, message: PlayerMessage) {
self.tx_player
.send(message)
.unwrap_or_else(|e| warn!("Send error {}", e));
}
}
fn send_reply<T>(tx: Sender<T>, value: T) {
if tx.send(value).is_err() {
warn!("player engine reply receiver dropped");
}
}

24
cbd-tui/Cargo.toml
View File

@ -1,17 +1,15 @@
[package]
name = "cbd-tui"
version = "0.1.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
version.workspace = true
edition.workspace = true
[dependencies]
crossterm = "0.26.1"
crabidy-core = { path = "../crabidy-core" }
flume = "0.10.14"
ratatui = "0.20.1"
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"
tonic = "0.9"
notify-rust = "4.8.0"
serde = "1.0.164"
crabidy-core.workspace = true
crossterm.workspace = true
flume.workspace = true
notify-rust.workspace = true
ratatui.workspace = true
serde.workspace = true
tokio = { workspace = true, features = ["full"] }
tokio-stream.workspace = true
tonic.workspace = true

3
cbd-tui/src/app/library.rs
View File

@ -2,7 +2,6 @@ use std::collections::HashMap;
use flume::Sender;
use ratatui::{
backend::Backend,
layout::Rect,
style::{Modifier, Style},
text::Span,
@ -160,7 +159,7 @@ impl Library {
self.update_selection();
}
pub fn render<B: Backend>(&mut self, f: &mut Frame<B>, area: Rect, focused: bool) {
pub fn render(&mut self, f: &mut Frame, area: Rect, focused: bool) {
let library_items: Vec<ListItem> = self
.list
.iter()

7
cbd-tui/src/app/mod.rs
View File

@ -5,7 +5,6 @@ mod queue;
use flume::Sender;
use ratatui::{
backend::Backend,
layout::{Constraint, Direction, Layout},
style::Color,
Frame,
@ -105,8 +104,8 @@ impl App {
};
}
pub fn render<B: Backend>(&mut self, f: &mut Frame<B>) {
let full_screen = f.size();
pub fn render(&mut self, f: &mut Frame) {
let full_screen = f.area();
let library_focused = matches!(self.focus, UiFocus::Library);
let queue_focused = matches!(self.focus, UiFocus::Queue);
@ -114,7 +113,7 @@ impl App {
let main = Layout::default()
.direction(Direction::Horizontal)
.constraints([Constraint::Percentage(50), Constraint::Percentage(50)].as_ref())
.split(f.size());
.split(f.area());
self.library.render(f, main[0], library_focused);

21
cbd-tui/src/app/now_playing.rs
View File

@ -5,10 +5,9 @@ use notify_rust::Notification;
use crabidy_core::proto::crabidy::{PlayState, QueueModifiers, Track, TrackPosition};
use ratatui::{
backend::Backend,
layout::{Alignment, Constraint, Direction, Layout, Rect},
style::{Color, Modifier, Style},
text::{Span, Spans},
text::{Line, Span},
widgets::{Block, BorderType, Borders, LineGauge, Paragraph, Wrap},
Frame,
};
@ -69,7 +68,7 @@ impl NowPlaying {
self.modifiers = mods.clone();
}
pub fn render<B: Backend>(&self, f: &mut Frame<B>, area: Rect) {
pub fn render(&self, f: &mut Frame, area: Rect) {
let now_playing_layout = Layout::default()
.direction(Direction::Vertical)
.constraints([Constraint::Max(8), Constraint::Max(1)])
@ -91,9 +90,9 @@ impl NowPlaying {
self.modifiers.shuffle, self.modifiers.repeat
);
vec![
Spans::from(Span::raw(mods)),
Spans::from(Span::raw(play_text)),
Spans::from(vec![
Line::from(Span::raw(mods)),
Line::from(Span::raw(play_text)),
Line::from(vec![
Span::styled(
track.title.to_string(),
Style::default().add_modifier(Modifier::BOLD),
@ -104,13 +103,13 @@ impl NowPlaying {
Style::default().add_modifier(Modifier::BOLD),
),
]),
Spans::from(Span::raw(album_text)),
Line::from(Span::raw(album_text)),
]
} else {
vec![
Spans::from(Span::raw("")),
Spans::from(Span::raw("")),
Spans::from(Span::raw("No track playing")),
Line::from(Span::raw("")),
Line::from(Span::raw("")),
Line::from(Span::raw("No track playing")),
]
};
@ -173,7 +172,7 @@ impl NowPlaying {
};
let time_text = Span::raw(completion_text);
let time_p = Paragraph::new(Spans::from(time_text));
let time_p = Paragraph::new(Line::from(time_text));
f.render_widget(time_p, elapsed_layout[1]);
}
}

3
cbd-tui/src/app/queue.rs
View File

@ -1,6 +1,5 @@
use flume::Sender;
use ratatui::{
backend::Backend,
layout::Rect,
style::{Modifier, Style},
text::Span,
@ -71,7 +70,7 @@ impl Queue {
self.update_selection();
}
pub fn render<B: Backend>(&mut self, f: &mut Frame<B>, area: Rect, focused: bool) {
pub fn render(&mut self, f: &mut Frame, area: Rect, focused: bool) {
let queue_items: Vec<ListItem> = self
.list
.iter()

4
cbd-tui/src/main.rs
View File

@ -173,7 +173,7 @@ fn run_ui(tx: Sender<MessageFromUi>, rx: Receiver<MessageToUi>) {
app.now_playing.update_track(track.track);
app.queue.update_position(track.queue_position as usize);
}
if let Some(ps) = PlayState::from_i32(init_data.play_state) {
if let Some(ps) = PlayState::try_from(init_data.play_state).ok() {
app.now_playing.update_play_state(ps);
}
if let Some(mods) = init_data.mods {
@ -190,7 +190,7 @@ fn run_ui(tx: Sender<MessageFromUi>, rx: Receiver<MessageToUi>) {
}
StreamUpdate::Position(pos) => app.now_playing.update_position(pos),
StreamUpdate::PlayState(play_state) => {
if let Some(ps) = PlayState::from_i32(play_state) {
if let Some(ps) = PlayState::try_from(play_state).ok() {
app.now_playing.update_play_state(ps);
}
}

27
crabidy-core/Cargo.toml
View File

@ -1,21 +1,18 @@
[package]
name = "crabidy-core"
version = "0.1.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
version.workspace = true
edition.workspace = true
[dependencies]
async-trait = "0.1.68"
clap = "4.3.3"
clap-serde-derive = "0.2.0"
dirs = "5.0.1"
prost = "0.11"
serde = "1.0.163"
toml = "0.7.4"
tonic = "0.9"
async-trait.workspace = true
clap.workspace = true
clap-serde-derive.workspace = true
dirs.workspace = true
prost.workspace = true
serde.workspace = true
toml.workspace = true
tonic.workspace = true
tonic-prost.workspace = true
[build-dependencies]
async-trait = "0.1.68"
serde = "1.0.163"
tonic-build = "0.9"
tonic-prost-build.workspace = true

2
crabidy-core/build.rs
View File

@ -1,4 +1,4 @@
fn main() -> Result<(), Box<dyn std::error::Error>> {
tonic_build::compile_protos("crabidy/v1/crabidy.proto")?;
tonic_prost_build::compile_protos("crabidy/v1/crabidy.proto")?;
Ok(())
}

43
crabidy-server/Cargo.toml
View File

@ -1,32 +1,27 @@
[package]
name = "crabidy-server"
version = "0.1.0"
edition = "2021"
version.workspace = true
edition.workspace = true
[[bin]]
name = "crabidy-server"
path = "src/main.rs"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
anyhow = "1.0.71"
tokio = { version = "1.28.0", features = ["full"] }
tidaldy = { path = "../tidaldy" }
crabidy-core = { path = "../crabidy-core" }
audio-player = { path = "../audio-player" }
once_cell = "1.17.1"
serde_json = "1.0.96"
serde = "1.0.163"
flume = "0.10.14"
tonic = "0.9.2"
async-trait = "0.1.68"
futures = "0.3.28"
tokio-stream = { version = "0.1.14", features = ["sync"] }
dirs = "5.0.1"
tracing = "0.1.37"
tracing-subscriber = "0.3.17"
tracing-appender = "0.2.2"
tracing-log = "0.1.3"
log = "0.4.18"
rand = "0.8.5"
anyhow.workspace = true
async-trait.workspace = true
audio-player.workspace = true
crabidy-core.workspace = true
dirs.workspace = true
flume.workspace = true
futures.workspace = true
log.workspace = true
rand.workspace = true
tidaldy.workspace = true
tokio = { workspace = true, features = ["full"] }
tokio-stream = { workspace = true, features = ["sync"] }
tonic.workspace = true
tracing.workspace = true
tracing-appender.workspace = true
tracing-log.workspace = true
tracing-subscriber.workspace = true

32
crabidy-server/src/lib.rs
View File

@ -1,5 +1,5 @@
use crabidy_core::proto::crabidy::{Queue, Track};
use rand::{seq::SliceRandom, thread_rng};
use rand::{rng, seq::SliceRandom};
use std::time::SystemTime;
use tracing::{debug, error};
@ -64,15 +64,15 @@ impl QueueManager {
}
pub fn shuffle_all(&mut self) {
self.play_order.shuffle(&mut thread_rng());
self.play_order.shuffle(&mut rng());
}
pub fn shuffle_before(&mut self, pos: usize) {
self.play_order[..pos].shuffle(&mut thread_rng());
self.play_order[..pos].shuffle(&mut rng());
}
pub fn shuffle_behind(&mut self, pos: usize) {
self.play_order[pos + 1..].shuffle(&mut thread_rng());
self.play_order[pos + 1..].shuffle(&mut rng());
}
pub fn current_track(&self) -> Option<Track> {
@ -128,11 +128,11 @@ impl QueueManager {
.play_order
.iter()
.position(|&i| i == current_position as usize)
else {
error!("invalid current position");
error!("queue: {:#?}", self);
return false
};
else {
error!("invalid current position");
error!("queue: {:#?}", self);
return false;
};
if self.shuffle {
self.play_order.swap(0, current_offset);
self.current_offset = 0;
@ -184,15 +184,11 @@ impl QueueManager {
if *pos == self.current_position() as u32 {
play_next = true;
}
let Some(offset) = self
.play_order
.iter()
.position(|&i| i == *pos as usize)
else {
error!("invalid current position");
error!("queue: {:#?}", self);
return None
};
let Some(offset) = self.play_order.iter().position(|&i| i == *pos as usize) else {
error!("invalid current position");
error!("queue: {:#?}", self);
return None;
};
if offset < self.current_offset {
self.current_offset -= 1;
}

39
crabidy-server/src/playback.rs
View File

@ -216,9 +216,9 @@ impl Playback {
debug!("queue lock released");
let state = {
let Ok(state) = self.state.lock() else {
error!("failed to get play state lock");
continue;
};
error!("failed to get play state lock");
continue;
};
*state
};
if state == PlayState::Playing {
@ -376,9 +376,9 @@ impl Playback {
{
let state = {
let Ok(state) = self.state.lock() else {
debug!("got state lock");
continue;
};
debug!("got state lock");
continue;
};
*state
};
debug!("got state lock");
@ -523,19 +523,20 @@ impl Playback {
let tx = self.provider_tx.clone();
let (result_tx, result_rx) = flume::bounded(1);
let span = debug_span!("prov-chan");
let Ok(_) = tx.send_async(ProviderMessage::FlattenNode {
uuid: uuid.to_string(),
result_tx,
span,
}).in_current_span().await else {
let Ok(_) = tx
.send_async(ProviderMessage::FlattenNode {
uuid: uuid.to_string(),
result_tx,
span,
})
.in_current_span()
.await
else {
return Vec::new();
};
let Ok(tracks) = result_rx.recv_async().in_current_span().await else {
return Vec::new();
};
let Ok(tracks) = result_rx
.recv_async()
.in_current_span()
.await else {
return Vec::new();
};
tracks
}
@ -608,7 +609,7 @@ impl Playback {
{
let Ok(queue) = self.queue.lock() else {
error!("poisend queue lock");
return
return;
};
let queue_update_tx = self.update_tx.clone();
let track = queue.current_track();
@ -655,7 +656,7 @@ impl Playback {
{
let Ok(queue) = self.queue.lock() else {
error!("poisend queue lock");
return
return;
};
let queue_update_tx = self.update_tx.clone();
let track = queue.current_track();

4
crabidy-server/src/provider.rs
View File

@ -76,8 +76,8 @@ impl ProviderOrchestrator {
nodes_to_go.push(node_uuid.to_string());
while let Some(node_uuid) = nodes_to_go.pop() {
let Ok(node) = self.get_lib_node(&node_uuid).in_current_span().await else {
continue
};
continue;
};
if node.is_queable {
tracks.extend(node.tracks);
nodes_to_go.extend(node.children.into_iter().map(|c| c.uuid))

1
devenv.nix
View File

@ -12,6 +12,7 @@ let
extraPackages = with pkgs; [
pkg-config
protobuf
];
in
{

10
stream-download/.gitignore vendored
View File

@ -1,10 +0,0 @@
# Generated by Cargo
# will have compiled files and executables
/target/
# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html
Cargo.lock
# These are backup files generated by rustfmt
**/*.rs.bk

32
stream-download/Cargo.toml
View File

@ -1,32 +0,0 @@
[package]
edition = "2021"
name = "stream-download"
version = "0.1.0"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
async-trait = "0.1"
bytes = "1"
futures = "0.3"
futures-util = "0.3"
parking_lot = "0.12"
rangemap = "1"
reqwest = { version = "0.11", features = [
"rustls-tls",
"trust-dns",
"stream",
], default-features = false, optional = true }
symphonia = "0.5"
tempfile = "3"
tokio = { version = "1", features = ["sync", "macros"] }
tracing = "0.1"
[features]
default = ["http"]
http = ["reqwest"]
[dev-dependencies]
rodio = "0.17.1"
tracing-subscriber = "0.3.16"
tokio = { version = "1", features = ["sync", "macros", "rt-multi-thread"] }

21
stream-download/LICENSE
View File

@ -1,21 +0,0 @@
MIT License
Copyright (c) 2023 Austin Schey
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

1
stream-download/README.md
View File

@ -1 +0,0 @@
# stream-download-rs

18
stream-download/examples/audio.rs
View File

@ -1,18 +0,0 @@
use stream_download::StreamDownload;
#[tokio::main]
async fn main() {
tracing_subscriber::fmt().init();
let (_stream, handle) = rodio::OutputStream::try_default().unwrap();
let sink = rodio::Sink::try_new(&handle).unwrap();
let reader = StreamDownload::new_http(
"https://dl.espressif.com/dl/audio/ff-16b-2c-44100hz.flac"
.parse()
.unwrap(),
);
sink.append(rodio::Decoder::new(reader).unwrap());
sink.sleep_until_end();
}

17
stream-download/examples/no_tokio_runtime.rs
View File

@ -1,17 +0,0 @@
use stream_download::StreamDownload;
fn main() {
tracing_subscriber::fmt().init();
let (_stream, handle) = rodio::OutputStream::try_default().unwrap();
let sink = rodio::Sink::try_new(&handle).unwrap();
let reader = StreamDownload::new_http(
"https://dl.espressif.com/dl/audio/ff-16b-2c-44100hz.flac"
.parse()
.unwrap(),
);
sink.append(rodio::Decoder::new(reader).unwrap());
sink.sleep_until_end();
}

18
stream-download/examples/stream.rs
View File

@ -1,18 +0,0 @@
use stream_download::StreamDownload;
#[tokio::main]
async fn main() {
tracing_subscriber::fmt().init();
let (_stream, handle) = rodio::OutputStream::try_default().unwrap();
let sink = rodio::Sink::try_new(&handle).unwrap();
let reader = StreamDownload::new_http(
"https://uk1.internet-radio.com/proxy/pinknoise?mp=/stream"
.parse()
.unwrap(),
);
sink.append(rodio::Decoder::new(reader).unwrap());
sink.sleep_until_end();
}

81
stream-download/src/http.rs
View File

@ -1,81 +0,0 @@
use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use reqwest::Client;
use std::{
pin::Pin,
str::FromStr,
task::{self, Poll},
};
use tracing::{info, warn};
use crate::source::SourceStream;
pub struct HttpStream {
stream: Box<dyn Stream<Item = Result<Bytes, reqwest::Error>> + Unpin + Send + Sync>,
client: Client,
content_length: Option<u64>,
url: reqwest::Url,
}
impl Stream for HttpStream {
type Item = Result<Bytes, reqwest::Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
Pin::new(&mut self.stream).poll_next(cx)
}
}
#[async_trait]
impl SourceStream for HttpStream {
type Url = reqwest::Url;
type Error = reqwest::Error;
async fn create(url: Self::Url) -> Self {
let client = Client::new();
info!("Requesting content length");
let response = client.get(url.as_str()).send().await.unwrap();
let mut content_length = None;
if let Some(length) = response.headers().get(reqwest::header::CONTENT_LENGTH) {
let length = u64::from_str(length.to_str().unwrap()).unwrap();
info!("Got content length {length}");
content_length = Some(length);
} else {
warn!("Content length header missing");
}
let stream = response.bytes_stream();
Self {
stream: Box::new(stream),
client,
content_length,
url,
}
}
async fn content_length(&self) -> Option<u64> {
self.content_length
}
async fn seek(&mut self, pos: u64) {
info!("Seeking");
self.stream = Box::new(
self.client
.get(self.url.as_str())
.header(
"Range",
format!(
"bytes={pos}-{}",
self.content_length
.map(|l| l.to_string())
.unwrap_or_default()
),
)
.send()
.await
.unwrap()
.bytes_stream(),
);
info!("Done seeking");
}
}

166
stream-download/src/lib.rs
View File

@ -1,166 +0,0 @@
use source::{Source, SourceHandle, SourceStream};
use std::{
io::{self, BufReader, Read, Seek, SeekFrom},
thread,
};
use symphonia::core::io::MediaSource;
use tempfile::NamedTempFile;
use tracing::debug;
#[cfg(feature = "http")]
pub mod http;
pub mod source;
#[derive(Debug)]
pub struct StreamDownload {
output_reader: BufReader<NamedTempFile>,
handle: SourceHandle,
read_position: u64,
}
impl StreamDownload {
#[cfg(feature = "http")]
pub fn new_http(url: reqwest::Url) -> Self {
Self::new::<http::HttpStream>(url)
}
pub fn new<S: SourceStream>(url: S::Url) -> Self {
let tempfile = tempfile::Builder::new().tempfile().unwrap();
let source = Source::new(tempfile.reopen().unwrap());
let handle = source.source_handle();
if let Ok(handle) = tokio::runtime::Handle::try_current() {
handle.spawn(async move {
let stream = S::create(url).await;
source.download(stream).await;
});
} else {
thread::spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(async move {
let stream = S::create(url).await;
source.download(stream).await;
});
});
};
Self {
output_reader: BufReader::new(tempfile),
read_position: 0,
handle,
}
}
pub fn from_stream<S: SourceStream>(stream: S) -> Self {
let tempfile = tempfile::Builder::new().tempfile().unwrap();
let source = Source::new(tempfile.reopen().unwrap());
let handle = source.source_handle();
if let Ok(handle) = tokio::runtime::Handle::try_current() {
handle.spawn(async move {
source.download(stream).await;
});
} else {
thread::spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(async move {
source.download(stream).await;
});
});
};
Self {
output_reader: BufReader::new(tempfile),
handle,
read_position: 0,
}
}
}
impl Read for StreamDownload {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
debug!("Read request buf len: {}", buf.len());
let requested_position = self.read_position + buf.len() as u64;
debug!(
"read: current position: {} requested position: {requested_position}",
self.read_position
);
if let Some(closest_set) = self.handle.downloaded().get(&self.read_position) {
debug!("Already downloaded {closest_set:?}");
if closest_set.end >= requested_position {
let read_len = self.output_reader.read(buf);
if let Ok(read_len) = read_len {
self.read_position += read_len as u64;
}
return read_len;
}
}
self.handle.request_position(requested_position);
debug!("waiting for position");
self.handle.wait_for_requested_position();
debug!("reached requested position {requested_position}");
self.output_reader.read(buf)
}
}
impl Seek for StreamDownload {
fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
let seek_pos = match pos {
SeekFrom::Start(pos) => pos,
SeekFrom::End(pos) => {
if let Some(length) = self.handle.content_length() {
(length as i64 + pos) as u64
} else {
return Err(io::Error::new(
io::ErrorKind::Unsupported,
"Cannot seek from end when content length is unknown",
));
}
}
SeekFrom::Current(pos) => (self.read_position as i64 + pos) as u64,
};
if let Some(closest_set) = self.handle.downloaded().get(&seek_pos) {
if closest_set.end >= seek_pos {
let new_pos = self.output_reader.seek(pos);
if let Ok(new_pos) = new_pos {
self.read_position = new_pos;
}
}
}
self.handle.request_position(seek_pos);
debug!(
"seek: current position {seek_pos} requested position {:?}. waiting",
seek_pos
);
self.handle.seek(seek_pos);
self.handle.wait_for_requested_position();
debug!("reached seek position");
self.output_reader.seek(pos)
}
}
impl MediaSource for StreamDownload {
fn is_seekable(&self) -> bool {
true
}
// FIXME: Can this be implemented?
fn byte_len(&self) -> Option<u64> {
None
}
}

228
stream-download/src/source.rs
View File

@ -1,228 +0,0 @@
use async_trait::async_trait;
use bytes::Bytes;
use futures::{Stream, StreamExt};
use parking_lot::{Condvar, Mutex, RwLock, RwLockReadGuard};
use rangemap::RangeSet;
use std::{
error::Error,
fs::File,
io::{BufWriter, Seek, SeekFrom, Write},
sync::{
atomic::{AtomicI64, Ordering},
Arc,
},
};
use tokio::sync::mpsc;
use tracing::{debug, info, trace};
#[async_trait]
pub trait SourceStream:
Stream<Item = Result<Bytes, Self::Error>> + Unpin + Send + Sync + 'static
{
type Url: Send;
type Error: Error + Send;
async fn create(url: Self::Url) -> Self;
async fn content_length(&self) -> Option<u64>;
async fn seek(&mut self, position: u64);
}
#[derive(Debug, Clone)]
pub struct SourceHandle {
downloaded: Arc<RwLock<RangeSet<u64>>>,
requested_position: Arc<AtomicI64>,
position_reached: Arc<(Mutex<Waiter>, Condvar)>,
content_length_retrieved: Arc<(Mutex<bool>, Condvar)>,
content_length: Arc<AtomicI64>,
seek_tx: mpsc::Sender<u64>,
}
impl SourceHandle {
pub fn downloaded(&self) -> RwLockReadGuard<rangemap::RangeSet<u64>> {
self.downloaded.read()
}
pub fn request_position(&self, position: u64) {
self.requested_position
.store(position as i64, Ordering::SeqCst);
}
pub fn wait_for_requested_position(&self) {
let (mutex, cvar) = &*self.position_reached;
let mut waiter = mutex.lock();
if !waiter.stream_done {
debug!("Waiting for requested position");
cvar.wait_while(&mut waiter, |waiter| {
!waiter.stream_done && !waiter.position_reached
});
if !waiter.stream_done {
waiter.position_reached = false;
}
debug!("Position reached");
}
}
pub fn seek(&self, position: u64) {
self.seek_tx.try_send(position).ok();
}
pub fn content_length(&self) -> Option<u64> {
let (mutex, cvar) = &*self.content_length_retrieved;
let mut done = mutex.lock();
if !*done {
cvar.wait_while(&mut done, |done| !*done);
}
let length = self.content_length.load(Ordering::SeqCst);
if length > -1 {
Some(length as u64)
} else {
None
}
}
}
#[derive(Default, Debug)]
struct Waiter {
position_reached: bool,
stream_done: bool,
}
pub struct Source {
writer: BufWriter<File>,
downloaded: Arc<RwLock<RangeSet<u64>>>,
position: u64,
requested_position: Arc<AtomicI64>,
position_reached: Arc<(Mutex<Waiter>, Condvar)>,
content_length_retrieved: Arc<(Mutex<bool>, Condvar)>,
content_length: Arc<AtomicI64>,
seek_tx: mpsc::Sender<u64>,
seek_rx: mpsc::Receiver<u64>,
}
const PREFETCH_BYTES: u64 = 1024 * 256;
impl Source {
pub fn new(tempfile: File) -> Self {
let (seek_tx, seek_rx) = mpsc::channel(32);
Self {
writer: BufWriter::new(tempfile),
downloaded: Default::default(),
position: Default::default(),
requested_position: Arc::new(AtomicI64::new(-1)),
position_reached: Default::default(),
content_length_retrieved: Default::default(),
seek_tx,
seek_rx,
content_length: Default::default(),
}
}
pub async fn download<S: SourceStream>(mut self, mut stream: S) {
info!("Starting file download");
let content_length = stream.content_length().await;
if let Some(content_length) = content_length {
self.content_length
.swap(content_length as i64, Ordering::SeqCst);
} else {
self.content_length.swap(-1, Ordering::SeqCst);
}
{
let (mutex, cvar) = &*self.content_length_retrieved;
*mutex.lock() = true;
cvar.notify_all();
}
let mut initial_buffer = 0;
loop {
if let Some(bytes) = stream.next().await {
let bytes = bytes.unwrap();
self.writer.write_all(&bytes).unwrap();
initial_buffer += bytes.len() as u64;
trace!("Prefetch: {}/{} bytes", initial_buffer, PREFETCH_BYTES);
if initial_buffer >= PREFETCH_BYTES {
self.position += initial_buffer;
self.downloaded.write().insert(0..initial_buffer);
break;
}
} else {
info!("File shorter than prefetch length");
self.writer.flush().unwrap();
self.position += initial_buffer;
self.downloaded.write().insert(0..initial_buffer);
let (mutex, cvar) = &*self.position_reached;
(mutex.lock()).stream_done = true;
cvar.notify_all();
return;
}
}
info!("Prefetch complete");
loop {
tokio::select! {
bytes = stream.next() => {
if let Some(bytes) = bytes {
let bytes = bytes.unwrap();
let chunk_len = bytes.len() as u64;
self.writer.write_all(&bytes).unwrap();
let new_position = self.position + chunk_len;
trace!("Received response chunk. position={}", new_position);
self.downloaded.write().insert(self.position..new_position);
let requested = self.requested_position.load(Ordering::SeqCst);
if requested > -1 {
debug!("downloader: requested {requested} current {}", new_position);
}
if requested > -1 && new_position as i64 >= requested {
info!("Notifying");
self.requested_position.store(-1, Ordering::SeqCst);
let (mutex, cvar) = &*self.position_reached;
(mutex.lock()).position_reached = true;
cvar.notify_all();
}
self.position = new_position;
} else {
info!("Stream finished downloading");
self.writer.flush().unwrap();
let (mutex, cvar) = &*self.position_reached;
(mutex.lock()).stream_done = true;
cvar.notify_all();
return;
}
},
pos = self.seek_rx.recv() => {
if let Some(pos) = pos {
debug!("Received seek position {pos}");
let do_seek = {
let downloaded = self.downloaded.read();
if let Some(range) = downloaded.get(&pos) {
!range.contains(&self.position)
} else {
true
}
};
if do_seek {
stream.seek(pos).await;
self.writer.seek(SeekFrom::Start(pos)).unwrap();
self.position = pos;
}
}
}
}
}
}
pub fn source_handle(&self) -> SourceHandle {
SourceHandle {
downloaded: self.downloaded.clone(),
requested_position: self.requested_position.clone(),
position_reached: self.position_reached.clone(),
seek_tx: self.seek_tx.clone(),
content_length_retrieved: self.content_length_retrieved.clone(),
content_length: self.content_length.clone(),
}
}
}

34
tidaldy/Cargo.toml
View File

@ -1,25 +1,21 @@
[package]
name = "tidaldy"
version = "0.0.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
version.workspace = true
edition.workspace = true
[dependencies]
async-trait = "0.1.68"
base64 = "0.21.0"
chrono = "0.4.24"
confique = "0.2.3"
crabidy-core = { path = "../crabidy-core" }
reqwest = { version = "0.11.17", features = ["json", "rustls-tls", "trust-dns"], default-features = false }
secrecy = { version = "0.8.0", features = ["serde"] }
serde = { version = "1.0.162", features = ["derive"] }
serde_json = "1.0.96"
serde_urlencoded = "0.7.1"
thiserror = "1.0.40"
tokio = { version = "1.28.1", features = ["full", "time"] }
toml = "0.7.4"
tracing = "0.1.37"
async-trait.workspace = true
base64.workspace = true
chrono.workspace = true
crabidy-core.workspace = true
reqwest.workspace = true
serde.workspace = true
serde_json.workspace = true
serde_urlencoded.workspace = true
thiserror.workspace = true
tokio = { workspace = true, features = ["time", "sync", "macros"] }
toml.workspace = true
tracing.workspace = true
[dev-dependencies]
tokio = { version = "1.28.1", features = ["full"] }
tokio = { workspace = true, features = ["full"] }

52
tidaldy/src/lib.rs
View File

@ -1,5 +1,3 @@
use std::fmt::format;
/// Lots of stuff and especially the auth handling is shamelessly copied from
/// https://github.com/MinisculeGirraffe/tdl
use reqwest::Client as HttpClient;
@ -53,12 +51,12 @@ impl crabidy_core::ProviderClient for Client {
debug!("get_urls_for_track {}", track_uuid);
let (_, track_uuid, _) = split_uuid(track_uuid);
let Ok(playback) = self.get_track_playback(&track_uuid).await else {
return Err(crabidy_core::ProviderError::FetchError)
};
return Err(crabidy_core::ProviderError::FetchError);
};
debug!("playback {:?}", playback);
let Ok(manifest) = playback.get_manifest() else {
return Err(crabidy_core::ProviderError::FetchError)
};
return Err(crabidy_core::ProviderError::FetchError);
};
debug!("manifest {:?}", manifest);
Ok(manifest.urls)
}
@ -70,8 +68,8 @@ impl crabidy_core::ProviderClient for Client {
) -> Result<crabidy_core::proto::crabidy::Track, crabidy_core::ProviderError> {
debug!("get_metadata_for_track {}", track_uuid);
let Ok(track) = self.get_track(track_uuid).await else {
return Err(crabidy_core::ProviderError::FetchError)
};
return Err(crabidy_core::ProviderError::FetchError);
};
Ok(track.into())
}
@ -107,8 +105,8 @@ impl crabidy_core::ProviderClient for Client {
uuid: &str,
) -> Result<crabidy_core::proto::crabidy::LibraryNode, crabidy_core::ProviderError> {
let Some(user_id) = self.settings.login.user_id.clone() else {
return Err(crabidy_core::ProviderError::UnknownUser)
};
return Err(crabidy_core::ProviderError::UnknownUser);
};
debug!("get_lib_node in tidaldy{}", uuid);
let (_kind, module, uuid) = split_uuid(uuid);
error!("module:{},uuid: {}", module, uuid);
@ -236,14 +234,10 @@ impl Client {
) -> Result<T, ClientError> {
debug!("make_request {}", uri);
let Some(ref access_token) = self.settings.login.access_token.clone() else {
return Err(ClientError::AuthError(
"No access token found".to_string(),
))
return Err(ClientError::AuthError("No access token found".to_string()));
};
let Some(country_code) = self.settings.login.country_code.clone() else {
return Err(ClientError::AuthError(
"No country code found".to_string(),
))
return Err(ClientError::AuthError("No country code found".to_string()));
};
let country_param = ("countryCode", country_code);
let mut params: Vec<&(&str, String)> = vec![&country_param];
@ -279,14 +273,10 @@ impl Client {
) -> Result<Vec<T>, ClientError> {
debug!("make_paginated_request {}", uri);
let Some(ref access_token) = self.settings.login.access_token.clone() else {
return Err(ClientError::AuthError(
"No access token found".to_string(),
))
return Err(ClientError::AuthError("No access token found".to_string()));
};
let Some(country_code) = self.settings.login.country_code.clone() else {
return Err(ClientError::AuthError(
"No country code found".to_string(),
))
return Err(ClientError::AuthError("No country code found".to_string()));
};
let country_param = ("countryCode", country_code);
let limit = 50;
@ -353,14 +343,10 @@ impl Client {
query: Option<&[(&str, String)]>,
) -> Result<(), ClientError> {
let Some(ref access_token) = self.settings.login.access_token.clone() else {
return Err(ClientError::AuthError(
"No access token found".to_string(),
))
return Err(ClientError::AuthError("No access token found".to_string()));
};
let Some(country_code) = self.settings.login.country_code.clone() else {
return Err(ClientError::AuthError(
"No country code found".to_string(),
))
return Err(ClientError::AuthError("No country code found".to_string()));
};
let country_param = ("countryCode", country_code);
let mut params: Vec<&(&str, String)> = vec![&country_param];
@ -543,9 +529,7 @@ impl Client {
#[instrument(skip(self))]
pub async fn login_config(&mut self) -> Result<(), ClientError> {
let Some(access_token) = self.settings.login.access_token.clone() else {
return Err(ClientError::AuthError(
"No access token found".to_string(),
))
return Err(ClientError::AuthError("No access token found".to_string()));
};
//return if our session is still valid
if self
@ -576,10 +560,8 @@ impl Client {
#[instrument(skip(self))]
pub async fn refresh_access_token(&self) -> Result<RefreshResponse, ClientError> {
let Some(refresh_token) = self.settings.login.refresh_token.clone() else {
return Err(ClientError::AuthError(
"No refresh token found".to_string(),
))
};
return Err(ClientError::AuthError("No refresh token found".to_string()));
};
let data = DeviceAuthRequest {
client_id: self.settings.oauth.client_id.clone(),
client_secret: Some(self.settings.oauth.client_secret.clone()),

3
tidaldy/src/models.rs
View File

@ -1,5 +1,6 @@
use std::{str::FromStr, string::FromUtf8Error};
use base64::Engine as _;
use crabidy_core::proto::crabidy::{LibraryNode, LibraryNodeChild};
use serde::{Deserialize, Serialize};
use serde_json::Value;
@ -432,7 +433,7 @@ pub struct PlaybackManifest {
impl FromStr for PlaybackManifest {
type Err = ClientError;
fn from_str(input: &str) -> Result<PlaybackManifest, Self::Err> {
let decode = base64::decode(input)?;
let decode = base64::engine::general_purpose::STANDARD.decode(input)?;
let json = String::from_utf8(decode)?;
let parsed: PlaybackManifest = serde_json::from_str(&json)?;
Ok(parsed)