feat: add basic chunked audio send in /playback endpoint

- add `Library::get_track_by_id()`
- add futures as a dependency

Cargo.toml

@@ -7,6 +7,7 @@ edition = "2024"
 diesel = { version = "2.3.14", features = ["sqlite", "returning_clauses_for_sqlite_3_35"] }
 diesel_migrations = { version = "2.3.2", features = ["sqlite"] }
 dotenvy = "0.15.7"
+futures = "0.3.34"
 lofty = "0.25.4"
 ntex = { version = "4.4.0", features = ["tokio"] }
 serde = { version = "1.0.229", features = ["derive"] }

src/library.rs

@@ -5,7 +5,7 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-use diesel::{prelude::*, sqlite::SqliteConnection};
+use diesel::{prelude::*, result::Error::NotFound, sqlite::SqliteConnection};
 use lofty::{prelude::*, read_from};
 
 use crate::{
@@ -33,6 +33,16 @@ impl Library {
             .load::<Track>(&mut self.conn)
     }
 
+    pub fn get_track_by_id(&mut self, id: u32) -> QueryResult<Track> {
+        let Ok(id) = i32::try_from(id) else {
+            return Err(NotFound);
+        };
+
+        schema::tracks::table
+            .find(i32::from(id))
+            .first(&mut self.conn)
+    }
+
     pub fn print(&mut self) -> QueryResult<()> {
         let artists = schema::artists::table
             .order(schema::artists::id.asc())

src/main.rs

@@ -12,6 +12,7 @@ mod library;
 mod models;
 mod net;
 mod schema;
+mod stream;
 mod types;
 
 use library::{Handle, Importer, Library};

src/net.rs

@@ -1,14 +1,22 @@
 pub mod http {
     use std::{
-        env, io,
+        env, fs,
+        io::{self, Read},
         sync::{self, atomic},
     };
 
     use dotenvy::dotenv;
-    use ntex::{SharedCfg, server, web};
+    use futures::{future, stream, task};
+    use ntex::{
+        SharedCfg, server,
+        util::{self, Buf},
+        web,
+    };
+    use serde::Deserialize;
 
     use crate::library;
     use crate::net::ws;
+    use crate::stream::AudioFileStream;
 
     // This is per-worker state
     #[derive(Clone)]
@@ -37,6 +45,11 @@ pub mod http {
         }
     }
 
+    #[derive(Deserialize)]
+    struct PlaybackQuery {
+        track_id: u32,
+    }
+
     impl server::ServerAppConfig for AppBuilder {
         type State = AppState;
 
@@ -85,6 +98,37 @@ pub mod http {
         web::HttpResponse::Ok().body("websocket connected")
     }
 
+    async fn playback(
+        state: &AppState,
+        _: (),
+        req: web::HttpRequest,
+        query: web::types::Query<PlaybackQuery>,
+    ) -> web::HttpResponse {
+        println!("hit /playback, requesting {}", query.track_id);
+
+        let Ok(track) = state
+            .library_handle
+            .clone()
+            .with_library(|library| library.get_track_by_id(query.track_id))
+        else {
+            return web::HttpResponse::NotFound().build();
+        };
+
+        println!("opening {}", track.file_path);
+
+        let file_handle =
+            fs::File::open(&track.file_path).expect("whoops so the file isn't there i guess");
+
+        println!("  opened {}", &track.file_path);
+
+        let file_buffer = io::BufReader::new(file_handle);
+        let file_stream = AudioFileStream::from_file_buffer(file_buffer);
+
+        web::HttpResponse::Ok()
+            .content_type("application/ogg")
+            .streaming(file_stream)
+    }
+
     #[ntex::main]
     pub async fn start(library_handle: library::Handle) -> std::io::Result<()> {
         dotenv().ok();
@@ -107,6 +151,7 @@ pub mod http {
             web::App::new()
                 .service(web::resource("/").to(index))
                 .service(web::resource("/ws").to_with_state(websocket))
+                .service(web::resource("/playback").to_with_state(playback))
         })
         .bind((address, port), ntex::SharedCfg::new("STATE"))?
         .workers(2)

src/stream.rs

@@ -0,0 +1,52 @@
+use std::{fs, io, pin, task};
+
+use futures::stream;
+use ntex::util::{self, Buf};
+
+pub struct AudioFileStream {
+    file_buffer: io::BufReader<fs::File>,
+    poll_count: u64,
+}
+
+impl AudioFileStream {
+    pub fn from_file_buffer(buffer: io::BufReader<fs::File>) -> Self {
+        println!(
+            "creating AudioFileStream from buffer, {}",
+            buffer.buffer().len()
+        );
+
+        Self {
+            file_buffer: buffer,
+            poll_count: 0,
+        }
+    }
+}
+
+impl stream::Stream for AudioFileStream {
+    type Item = Result<util::Bytes, io::Error>;
+
+    fn poll_next(
+        mut self: pin::Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+    ) -> task::Poll<Option<Self::Item>> {
+        self.poll_count += 1;
+
+        let Ok(view) = io::BufRead::fill_buf(&mut self.file_buffer) else {
+            eprintln!("failed to continue filling buffer! poll_count {}", self.poll_count);
+            return task::Poll::Ready(None);
+        };
+
+        let out = util::Bytes::copy_from_slice(view);
+
+        if out.len() == 0 {
+            return task::Poll::Ready(None);
+        }
+
+        if let Err(e) = self.file_buffer.seek_relative(8192) {
+            eprintln!("failed to seek buffer! poll_count {}", self.poll_count);
+            return task::Poll::Pending;
+        }
+
+        task::Poll::Ready(Some(Ok(out)))
+    }
+}