Create server side
This commit is contained in:
Alexander
Generated
+342
@@ -107,6 +107,12 @@ dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "anyhow"
|
||||
version = "1.0.103"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3"
|
||||
|
||||
[[package]]
|
||||
name = "arrayvec"
|
||||
version = "0.7.7"
|
||||
@@ -314,12 +320,61 @@ dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "atomic-waker"
|
||||
version = "1.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
|
||||
|
||||
[[package]]
|
||||
name = "autocfg"
|
||||
version = "1.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
|
||||
|
||||
[[package]]
|
||||
name = "axum"
|
||||
version = "0.8.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
|
||||
dependencies = [
|
||||
"axum-core",
|
||||
"bytes",
|
||||
"futures-util",
|
||||
"http",
|
||||
"http-body",
|
||||
"http-body-util",
|
||||
"itoa",
|
||||
"matchit",
|
||||
"memchr",
|
||||
"mime",
|
||||
"percent-encoding",
|
||||
"pin-project-lite",
|
||||
"serde_core",
|
||||
"sync_wrapper",
|
||||
"tower",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "axum-core"
|
||||
version = "0.5.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"http",
|
||||
"http-body",
|
||||
"http-body-util",
|
||||
"mime",
|
||||
"pin-project-lite",
|
||||
"sync_wrapper",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "base64"
|
||||
version = "0.22.1"
|
||||
@@ -1028,6 +1083,25 @@ dependencies = [
|
||||
"rand_core 0.10.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "h2"
|
||||
version = "0.4.15"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155"
|
||||
dependencies = [
|
||||
"atomic-waker",
|
||||
"bytes",
|
||||
"fnv",
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
"http",
|
||||
"indexmap",
|
||||
"slab",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "half"
|
||||
version = "2.7.1"
|
||||
@@ -1111,6 +1185,51 @@ dependencies = [
|
||||
"digest 0.11.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "http"
|
||||
version = "1.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"itoa",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "http-body"
|
||||
version = "1.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"http",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "http-body-util"
|
||||
version = "0.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"http",
|
||||
"http-body",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "httparse"
|
||||
version = "1.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
|
||||
|
||||
[[package]]
|
||||
name = "httpdate"
|
||||
version = "1.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
|
||||
|
||||
[[package]]
|
||||
name = "hybrid-array"
|
||||
version = "0.4.12"
|
||||
@@ -1120,6 +1239,61 @@ dependencies = [
|
||||
"typenum",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hyper"
|
||||
version = "1.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498"
|
||||
dependencies = [
|
||||
"atomic-waker",
|
||||
"bytes",
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"h2",
|
||||
"http",
|
||||
"http-body",
|
||||
"httparse",
|
||||
"httpdate",
|
||||
"itoa",
|
||||
"pin-project-lite",
|
||||
"smallvec",
|
||||
"tokio",
|
||||
"want",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hyper-timeout"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
|
||||
dependencies = [
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hyper-util"
|
||||
version = "0.1.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-channel",
|
||||
"futures-util",
|
||||
"http",
|
||||
"http-body",
|
||||
"hyper",
|
||||
"libc",
|
||||
"pin-project-lite",
|
||||
"socket2",
|
||||
"tokio",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "iana-time-zone"
|
||||
version = "0.1.65"
|
||||
@@ -1458,6 +1632,12 @@ dependencies = [
|
||||
"winapi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "matchit"
|
||||
version = "0.8.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
|
||||
|
||||
[[package]]
|
||||
name = "md-5"
|
||||
version = "0.11.0"
|
||||
@@ -1483,6 +1663,12 @@ dependencies = [
|
||||
"autocfg",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mime"
|
||||
version = "0.3.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
|
||||
|
||||
[[package]]
|
||||
name = "mio"
|
||||
version = "1.2.1"
|
||||
@@ -1499,15 +1685,23 @@ dependencies = [
|
||||
name = "musicfs"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"clap",
|
||||
"fuser",
|
||||
"http",
|
||||
"libc",
|
||||
"notify",
|
||||
"prost",
|
||||
"sea-orm",
|
||||
"symphonia",
|
||||
"tempfile",
|
||||
"time",
|
||||
"tokio",
|
||||
"tokio-stream",
|
||||
"tonic",
|
||||
"tonic-prost",
|
||||
"twox-hash",
|
||||
]
|
||||
|
||||
@@ -1729,6 +1923,26 @@ dependencies = [
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pin-project"
|
||||
version = "1.1.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924"
|
||||
dependencies = [
|
||||
"pin-project-internal",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pin-project-internal"
|
||||
version = "1.1.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.118",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pin-project-lite"
|
||||
version = "0.2.17"
|
||||
@@ -1828,6 +2042,29 @@ dependencies = [
|
||||
"yansi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost"
|
||||
version = "0.14.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"prost-derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost-derive"
|
||||
version = "0.14.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"itertools",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.118",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ptr_meta"
|
||||
version = "0.1.4"
|
||||
@@ -2772,6 +3009,12 @@ dependencies = [
|
||||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "sync_wrapper"
|
||||
version = "1.0.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263"
|
||||
|
||||
[[package]]
|
||||
name = "synstructure"
|
||||
version = "0.13.2"
|
||||
@@ -2924,6 +3167,19 @@ dependencies = [
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-util"
|
||||
version = "0.7.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"futures-sink",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "toml_datetime"
|
||||
version = "1.1.1+spec-1.1.0"
|
||||
@@ -2954,6 +3210,77 @@ dependencies = [
|
||||
"winnow",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tonic"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"axum",
|
||||
"base64",
|
||||
"bytes",
|
||||
"h2",
|
||||
"http",
|
||||
"http-body",
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
"hyper-timeout",
|
||||
"hyper-util",
|
||||
"percent-encoding",
|
||||
"pin-project",
|
||||
"socket2",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tokio-stream",
|
||||
"tower",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tonic-prost"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "50849f68853be452acf590cde0b146665b8d507b3b8af17261df47e02c209ea0"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"prost",
|
||||
"tonic",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tower"
|
||||
version = "0.5.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-util",
|
||||
"indexmap",
|
||||
"pin-project-lite",
|
||||
"slab",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tower-layer"
|
||||
version = "0.3.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e"
|
||||
|
||||
[[package]]
|
||||
name = "tower-service"
|
||||
version = "0.3.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3"
|
||||
|
||||
[[package]]
|
||||
name = "tracing"
|
||||
version = "0.1.44"
|
||||
@@ -2986,6 +3313,12 @@ dependencies = [
|
||||
"once_cell",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "try-lock"
|
||||
version = "0.2.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
|
||||
|
||||
[[package]]
|
||||
name = "twox-hash"
|
||||
version = "2.1.2"
|
||||
@@ -3091,6 +3424,15 @@ dependencies = [
|
||||
"winapi-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "want"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e"
|
||||
dependencies = [
|
||||
"try-lock",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "wasi"
|
||||
version = "0.11.1+wasi-snapshot-preview1"
|
||||
|
||||
+22
-1
@@ -3,6 +3,18 @@ name = "musicfs"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
|
||||
[lib]
|
||||
name = "musicfs"
|
||||
path = "src/lib.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "musicfs"
|
||||
path = "src/main.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "musicfs-server"
|
||||
path = "src/bin/musicfs-server.rs"
|
||||
|
||||
[dependencies]
|
||||
clap = { version = "4.6.1", features = ["derive"] }
|
||||
fuser = "0.17.0"
|
||||
@@ -15,7 +27,16 @@ symphonia = { version = "0.5", default-features = false, features = [
|
||||
twox-hash = "2.1.2"
|
||||
|
||||
sea-orm = { version = "2.0.0-rc", features = [ "sqlx-postgres", "runtime-tokio", "macros" ] }
|
||||
tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal"] }
|
||||
tokio = { version = "1", features = ["macros", "rt-multi-thread", "signal", "fs", "io-util", "sync"] }
|
||||
|
||||
anyhow = "1"
|
||||
async-trait = "0.1"
|
||||
bytes = "1"
|
||||
tonic = "0.14"
|
||||
tonic-prost = "0.14"
|
||||
prost = "0.14"
|
||||
tokio-stream = "0.1"
|
||||
http = "1"
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = "3"
|
||||
|
||||
@@ -29,6 +29,9 @@
|
||||
just
|
||||
flac
|
||||
ffmpeg
|
||||
protobuf
|
||||
buf
|
||||
grpcurl
|
||||
|
||||
opencode
|
||||
inputs.pinix.packages.${pkgs.system}.default
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
version: v2
|
||||
plugins:
|
||||
- remote: buf.build/community/neoeinstein-prost
|
||||
out: ../src/proto/generated
|
||||
- remote: buf.build/community/neoeinstein-tonic
|
||||
out: ../src/proto/generated
|
||||
@@ -0,0 +1,9 @@
|
||||
version: v2
|
||||
modules:
|
||||
- path: .
|
||||
lint:
|
||||
use:
|
||||
- STANDARD
|
||||
breaking:
|
||||
use:
|
||||
- FILE
|
||||
@@ -0,0 +1,75 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package musicfs;
|
||||
|
||||
// MusicFs service exposes a server-side music library to a remote musicfs
|
||||
// FUSE mount. Three RPCs, all server-streaming so the client can pull
|
||||
// incrementally and the server can keep large bodies off the heap.
|
||||
service MusicFs {
|
||||
// Stream every file entry in the library. The client diffs each entry's
|
||||
// hash against its DB to decide what changed.
|
||||
rpc GetManifest(GetManifestRequest) returns (stream ManifestEntry);
|
||||
|
||||
// Stream the bytes of one file. If `start` and `length` are both set on
|
||||
// the request, only that byte range is sent; otherwise the whole file.
|
||||
// The first chunk carries `total_size` and `offset`; subsequent chunks
|
||||
// continue from there.
|
||||
rpc GetFile(GetFileRequest) returns (stream FileChunk);
|
||||
|
||||
// Long-lived stream of filesystem change events. The client uses these as
|
||||
// wake-ups to refetch the manifest; correctness always rests on the hash
|
||||
// diff, never on the event payload.
|
||||
rpc SubscribeEvents(SubscribeEventsRequest) returns (stream ChangeEvent);
|
||||
}
|
||||
|
||||
message GetManifestRequest {}
|
||||
|
||||
message GetFileRequest {
|
||||
uint64 id = 1;
|
||||
// When both are zero the server streams the entire file.
|
||||
uint64 start = 2;
|
||||
uint64 length = 3;
|
||||
}
|
||||
|
||||
message SubscribeEventsRequest {}
|
||||
|
||||
message ManifestEntry {
|
||||
uint64 id = 1;
|
||||
string rel_path = 2;
|
||||
uint64 size = 3;
|
||||
uint64 mtime = 4;
|
||||
uint64 ctime = 5;
|
||||
uint64 crtime = 6;
|
||||
MusicMetadata music_metadata = 7;
|
||||
}
|
||||
|
||||
message MusicMetadata {
|
||||
repeated string artist = 1;
|
||||
optional string album_artist = 2;
|
||||
string album = 3;
|
||||
int32 track_number = 4;
|
||||
string track_title = 5;
|
||||
repeated string other_tags = 6;
|
||||
bytes header = 7;
|
||||
repeated bytes picture_block_headers = 8;
|
||||
repeated PictureDataRange picture_data_ranges = 9;
|
||||
uint64 real_audio_start = 10;
|
||||
uint64 vorbis_comment_offset = 11;
|
||||
uint64 vorbis_comment_length = 12;
|
||||
}
|
||||
|
||||
message PictureDataRange {
|
||||
uint64 offset = 1;
|
||||
uint64 length = 2;
|
||||
}
|
||||
|
||||
message FileChunk {
|
||||
bytes data = 1;
|
||||
uint64 offset = 2;
|
||||
uint64 total_size = 3;
|
||||
}
|
||||
|
||||
message ChangeEvent {
|
||||
// "create", "modify", or "remove".
|
||||
string kind = 1;
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
use std::{net::SocketAddr, path::PathBuf};
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use clap::Parser;
|
||||
use musicfs::server::{
|
||||
state::ServerState,
|
||||
transport::{self, TransportArgs},
|
||||
watcher::ServerWatcher,
|
||||
};
|
||||
use tokio::signal::unix::{SignalKind, signal};
|
||||
|
||||
#[derive(Parser, Debug)]
|
||||
#[command(version, about, long_about = None)]
|
||||
struct Args {
|
||||
/// Directory containing the music library to serve.
|
||||
#[arg(short, long, required = true)]
|
||||
source: PathBuf,
|
||||
|
||||
/// Address:port the transport should listen on (e.g. 0.0.0.0:50051).
|
||||
#[arg(short, long, default_value = "0.0.0.0:50051")]
|
||||
listen: SocketAddr,
|
||||
|
||||
/// Name of the transport implementation to use. Today only "grpc".
|
||||
#[arg(short, long, default_value = "grpc")]
|
||||
transport: String,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<()> {
|
||||
let args = Args::parse();
|
||||
|
||||
let source = args.source.clone();
|
||||
if !source.is_dir() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"source is not a readable directory: {}",
|
||||
source.display()
|
||||
));
|
||||
}
|
||||
|
||||
let state = ServerState::new();
|
||||
state
|
||||
.replace_all(&source)
|
||||
.with_context(|| format!("initial scan of {}", source.display()))?;
|
||||
|
||||
let (_watcher, events_rx) = ServerWatcher::spawn(source.clone(), state.clone());
|
||||
|
||||
let transport = transport::build(
|
||||
&args.transport,
|
||||
TransportArgs {
|
||||
listen: args.listen,
|
||||
},
|
||||
)
|
||||
.with_context(|| format!("building transport {:?}", args.transport))?;
|
||||
|
||||
println!(
|
||||
"musicfs-server: transport={}, listen={}, initial scan complete",
|
||||
args.transport, args.listen
|
||||
);
|
||||
|
||||
let server_task = tokio::spawn(async move {
|
||||
if let Err(e) = transport.run(state, events_rx).await {
|
||||
eprintln!("transport ended with error: {e}");
|
||||
}
|
||||
});
|
||||
|
||||
let mut sigint = signal(SignalKind::interrupt()).expect("register SIGINT");
|
||||
let mut sigterm = signal(SignalKind::terminate()).expect("register SIGTERM");
|
||||
|
||||
tokio::select! {
|
||||
_ = sigint.recv() => println!("received SIGINT, shutting down"),
|
||||
_ = sigterm.recv() => println!("received SIGTERM, shutting down"),
|
||||
_ = server_task => println!("transport task exited"),
|
||||
}
|
||||
|
||||
return Ok(());
|
||||
}
|
||||
@@ -2,4 +2,6 @@ pub mod db;
|
||||
pub mod item;
|
||||
pub mod music;
|
||||
pub mod origins;
|
||||
pub mod proto;
|
||||
pub mod server;
|
||||
pub mod virtual_dirs;
|
||||
|
||||
@@ -3,4 +3,5 @@ pub mod encoder;
|
||||
pub mod flac;
|
||||
pub mod metadata;
|
||||
pub mod mp3;
|
||||
pub mod parse;
|
||||
pub mod parser;
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
use std::path::Path;
|
||||
|
||||
use crate::music::encoder::MusicMetadataEncoderFactory;
|
||||
use crate::music::metadata::MusicMetadata;
|
||||
use crate::music::parser::MusicMetadataParserFactory;
|
||||
|
||||
/// Parse a single media file into a fully-encoded [`MusicMetadata`].
|
||||
///
|
||||
/// `None` is returned for files musicfs does not treat as music (unknown
|
||||
/// extension or unparseable container). When metadata is produced, the
|
||||
/// matching [`MusicMetadataEncoder`] has already baked the tag fields into
|
||||
/// the in-memory `header`, so callers can serve virtualized bytes directly.
|
||||
///
|
||||
/// This helper is the single source of truth for "parse + encode" — used by
|
||||
/// the local FUSE origin's snapshot builder, the future network origin's
|
||||
/// manifest builder, and the HTTP/3 server's manifest endpoint.
|
||||
pub fn parse_music_metadata_for_path(path: &Path) -> Option<MusicMetadata> {
|
||||
let mut metadata = MusicMetadataParserFactory::for_path(path)?.parse(path)?;
|
||||
if let Some(encoder) = MusicMetadataEncoderFactory::for_path(path) {
|
||||
encoder.encode(&mut metadata);
|
||||
}
|
||||
return Some(metadata);
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::path::Path;
|
||||
|
||||
#[test]
|
||||
fn none_for_non_music_extension() {
|
||||
let path = Path::new("notes.txt");
|
||||
assert!(parse_music_metadata_for_path(path).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn none_for_missing_file_with_music_extension() {
|
||||
let path = Path::new("/nonexistent/track.flac");
|
||||
assert!(parse_music_metadata_for_path(path).is_none());
|
||||
}
|
||||
}
|
||||
@@ -9,8 +9,7 @@ use fuser::INodeNo;
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
|
||||
use crate::item::{FileType, Item};
|
||||
use crate::music::encoder::MusicMetadataEncoderFactory;
|
||||
use crate::music::parser::MusicMetadataParserFactory;
|
||||
use crate::music::parse::parse_music_metadata_for_path;
|
||||
use crate::origins::attrs::FileAttrs;
|
||||
use crate::virtual_dirs::ensure_virtual_dirs;
|
||||
|
||||
@@ -72,14 +71,7 @@ pub fn read_into_map(
|
||||
|
||||
let name = entry.file_name().to_string_lossy().into_owned();
|
||||
let metadata = entry.metadata()?;
|
||||
let music_metadata = MusicMetadataParserFactory::for_path(&item_path)
|
||||
.and_then(|parser| parser.parse(&item_path))
|
||||
.map(|mut mm| {
|
||||
if let Some(encoder) = MusicMetadataEncoderFactory::for_path(&item_path) {
|
||||
encoder.encode(&mut mm);
|
||||
}
|
||||
mm
|
||||
});
|
||||
let music_metadata = parse_music_metadata_for_path(&item_path);
|
||||
|
||||
let mut local_path = PathBuf::new();
|
||||
if let Some(ref mm) = music_metadata {
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
// @generated
|
||||
// This file is @generated by prost-build.
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
|
||||
pub struct GetManifestRequest {}
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
|
||||
pub struct GetFileRequest {
|
||||
#[prost(uint64, tag = "1")]
|
||||
pub id: u64,
|
||||
/// When both are zero the server streams the entire file.
|
||||
#[prost(uint64, tag = "2")]
|
||||
pub start: u64,
|
||||
#[prost(uint64, tag = "3")]
|
||||
pub length: u64,
|
||||
}
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
|
||||
pub struct SubscribeEventsRequest {}
|
||||
#[derive(Clone, PartialEq, ::prost::Message)]
|
||||
pub struct ManifestEntry {
|
||||
#[prost(uint64, tag = "1")]
|
||||
pub id: u64,
|
||||
#[prost(string, tag = "2")]
|
||||
pub rel_path: ::prost::alloc::string::String,
|
||||
#[prost(uint64, tag = "3")]
|
||||
pub size: u64,
|
||||
#[prost(uint64, tag = "4")]
|
||||
pub mtime: u64,
|
||||
#[prost(uint64, tag = "5")]
|
||||
pub ctime: u64,
|
||||
#[prost(uint64, tag = "6")]
|
||||
pub crtime: u64,
|
||||
#[prost(message, optional, tag = "7")]
|
||||
pub music_metadata: ::core::option::Option<MusicMetadata>,
|
||||
}
|
||||
#[derive(Clone, PartialEq, ::prost::Message)]
|
||||
pub struct MusicMetadata {
|
||||
#[prost(string, repeated, tag = "1")]
|
||||
pub artist: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
|
||||
#[prost(string, optional, tag = "2")]
|
||||
pub album_artist: ::core::option::Option<::prost::alloc::string::String>,
|
||||
#[prost(string, tag = "3")]
|
||||
pub album: ::prost::alloc::string::String,
|
||||
#[prost(int32, tag = "4")]
|
||||
pub track_number: i32,
|
||||
#[prost(string, tag = "5")]
|
||||
pub track_title: ::prost::alloc::string::String,
|
||||
#[prost(string, repeated, tag = "6")]
|
||||
pub other_tags: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
|
||||
#[prost(bytes = "vec", tag = "7")]
|
||||
pub header: ::prost::alloc::vec::Vec<u8>,
|
||||
#[prost(bytes = "vec", repeated, tag = "8")]
|
||||
pub picture_block_headers: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec<u8>>,
|
||||
#[prost(message, repeated, tag = "9")]
|
||||
pub picture_data_ranges: ::prost::alloc::vec::Vec<PictureDataRange>,
|
||||
#[prost(uint64, tag = "10")]
|
||||
pub real_audio_start: u64,
|
||||
#[prost(uint64, tag = "11")]
|
||||
pub vorbis_comment_offset: u64,
|
||||
#[prost(uint64, tag = "12")]
|
||||
pub vorbis_comment_length: u64,
|
||||
}
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
|
||||
pub struct PictureDataRange {
|
||||
#[prost(uint64, tag = "1")]
|
||||
pub offset: u64,
|
||||
#[prost(uint64, tag = "2")]
|
||||
pub length: u64,
|
||||
}
|
||||
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
|
||||
pub struct FileChunk {
|
||||
#[prost(bytes = "vec", tag = "1")]
|
||||
pub data: ::prost::alloc::vec::Vec<u8>,
|
||||
#[prost(uint64, tag = "2")]
|
||||
pub offset: u64,
|
||||
#[prost(uint64, tag = "3")]
|
||||
pub total_size: u64,
|
||||
}
|
||||
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
|
||||
pub struct ChangeEvent {
|
||||
/// "create", "modify", or "remove".
|
||||
#[prost(string, tag = "1")]
|
||||
pub kind: ::prost::alloc::string::String,
|
||||
}
|
||||
include!("musicfs.tonic.rs");
|
||||
// @@protoc_insertion_point(module)
|
||||
@@ -0,0 +1,419 @@
|
||||
// @generated
|
||||
/// Generated client implementations.
|
||||
pub mod music_fs_client {
|
||||
#![allow(
|
||||
unused_variables,
|
||||
dead_code,
|
||||
missing_docs,
|
||||
clippy::wildcard_imports,
|
||||
clippy::let_unit_value
|
||||
)]
|
||||
use tonic::codegen::http::Uri;
|
||||
use tonic::codegen::*;
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct MusicFsClient<T> {
|
||||
inner: tonic::client::Grpc<T>,
|
||||
}
|
||||
impl MusicFsClient<tonic::transport::Channel> {
|
||||
/// Attempt to create a new client by connecting to a given endpoint.
|
||||
pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
|
||||
where
|
||||
D: TryInto<tonic::transport::Endpoint>,
|
||||
D::Error: Into<StdError>,
|
||||
{
|
||||
let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
|
||||
Ok(Self::new(conn))
|
||||
}
|
||||
}
|
||||
impl<T> MusicFsClient<T>
|
||||
where
|
||||
T: tonic::client::GrpcService<tonic::body::Body>,
|
||||
T::Error: Into<StdError>,
|
||||
T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
|
||||
<T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
|
||||
{
|
||||
pub fn new(inner: T) -> Self {
|
||||
let inner = tonic::client::Grpc::new(inner);
|
||||
Self { inner }
|
||||
}
|
||||
pub fn with_origin(inner: T, origin: Uri) -> Self {
|
||||
let inner = tonic::client::Grpc::with_origin(inner, origin);
|
||||
Self { inner }
|
||||
}
|
||||
pub fn with_interceptor<F>(
|
||||
inner: T,
|
||||
interceptor: F,
|
||||
) -> MusicFsClient<InterceptedService<T, F>>
|
||||
where
|
||||
F: tonic::service::Interceptor,
|
||||
T::ResponseBody: Default,
|
||||
T: tonic::codegen::Service<
|
||||
http::Request<tonic::body::Body>,
|
||||
Response = http::Response<
|
||||
<T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
|
||||
>,
|
||||
>,
|
||||
<T as tonic::codegen::Service<http::Request<tonic::body::Body>>>::Error:
|
||||
Into<StdError> + std::marker::Send + std::marker::Sync,
|
||||
{
|
||||
MusicFsClient::new(InterceptedService::new(inner, interceptor))
|
||||
}
|
||||
/// Compress requests with the given encoding.
|
||||
///
|
||||
/// This requires the server to support it otherwise it might respond with an
|
||||
/// error.
|
||||
#[must_use]
|
||||
pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
|
||||
self.inner = self.inner.send_compressed(encoding);
|
||||
self
|
||||
}
|
||||
/// Enable decompressing responses.
|
||||
#[must_use]
|
||||
pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
|
||||
self.inner = self.inner.accept_compressed(encoding);
|
||||
self
|
||||
}
|
||||
/// Limits the maximum size of a decoded message.
|
||||
///
|
||||
/// Default: `4MB`
|
||||
#[must_use]
|
||||
pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
|
||||
self.inner = self.inner.max_decoding_message_size(limit);
|
||||
self
|
||||
}
|
||||
/// Limits the maximum size of an encoded message.
|
||||
///
|
||||
/// Default: `usize::MAX`
|
||||
#[must_use]
|
||||
pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
|
||||
self.inner = self.inner.max_encoding_message_size(limit);
|
||||
self
|
||||
}
|
||||
pub async fn get_manifest(
|
||||
&mut self,
|
||||
request: impl tonic::IntoRequest<super::GetManifestRequest>,
|
||||
) -> std::result::Result<
|
||||
tonic::Response<tonic::codec::Streaming<super::ManifestEntry>>,
|
||||
tonic::Status,
|
||||
> {
|
||||
self.inner.ready().await.map_err(|e| {
|
||||
tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
|
||||
})?;
|
||||
let codec = tonic_prost::ProstCodec::default();
|
||||
let path = http::uri::PathAndQuery::from_static("/musicfs.MusicFs/GetManifest");
|
||||
let mut req = request.into_request();
|
||||
req.extensions_mut()
|
||||
.insert(GrpcMethod::new("musicfs.MusicFs", "GetManifest"));
|
||||
self.inner.server_streaming(req, path, codec).await
|
||||
}
|
||||
pub async fn get_file(
|
||||
&mut self,
|
||||
request: impl tonic::IntoRequest<super::GetFileRequest>,
|
||||
) -> std::result::Result<
|
||||
tonic::Response<tonic::codec::Streaming<super::FileChunk>>,
|
||||
tonic::Status,
|
||||
> {
|
||||
self.inner.ready().await.map_err(|e| {
|
||||
tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
|
||||
})?;
|
||||
let codec = tonic_prost::ProstCodec::default();
|
||||
let path = http::uri::PathAndQuery::from_static("/musicfs.MusicFs/GetFile");
|
||||
let mut req = request.into_request();
|
||||
req.extensions_mut()
|
||||
.insert(GrpcMethod::new("musicfs.MusicFs", "GetFile"));
|
||||
self.inner.server_streaming(req, path, codec).await
|
||||
}
|
||||
pub async fn subscribe_events(
|
||||
&mut self,
|
||||
request: impl tonic::IntoRequest<super::SubscribeEventsRequest>,
|
||||
) -> std::result::Result<
|
||||
tonic::Response<tonic::codec::Streaming<super::ChangeEvent>>,
|
||||
tonic::Status,
|
||||
> {
|
||||
self.inner.ready().await.map_err(|e| {
|
||||
tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
|
||||
})?;
|
||||
let codec = tonic_prost::ProstCodec::default();
|
||||
let path = http::uri::PathAndQuery::from_static("/musicfs.MusicFs/SubscribeEvents");
|
||||
let mut req = request.into_request();
|
||||
req.extensions_mut()
|
||||
.insert(GrpcMethod::new("musicfs.MusicFs", "SubscribeEvents"));
|
||||
self.inner.server_streaming(req, path, codec).await
|
||||
}
|
||||
}
|
||||
}
|
||||
/// Generated server implementations.
|
||||
pub mod music_fs_server {
|
||||
#![allow(
|
||||
unused_variables,
|
||||
dead_code,
|
||||
missing_docs,
|
||||
clippy::wildcard_imports,
|
||||
clippy::let_unit_value
|
||||
)]
|
||||
use tonic::codegen::*;
|
||||
/// Generated trait containing gRPC methods that should be implemented for use with MusicFsServer.
|
||||
#[async_trait]
|
||||
pub trait MusicFs: std::marker::Send + std::marker::Sync + 'static {
|
||||
/// Server streaming response type for the GetManifest method.
|
||||
type GetManifestStream: tonic::codegen::tokio_stream::Stream<
|
||||
Item = std::result::Result<super::ManifestEntry, tonic::Status>,
|
||||
> + std::marker::Send
|
||||
+ 'static;
|
||||
async fn get_manifest(
|
||||
&self,
|
||||
request: tonic::Request<super::GetManifestRequest>,
|
||||
) -> std::result::Result<tonic::Response<Self::GetManifestStream>, tonic::Status>;
|
||||
/// Server streaming response type for the GetFile method.
|
||||
type GetFileStream: tonic::codegen::tokio_stream::Stream<
|
||||
Item = std::result::Result<super::FileChunk, tonic::Status>,
|
||||
> + std::marker::Send
|
||||
+ 'static;
|
||||
async fn get_file(
|
||||
&self,
|
||||
request: tonic::Request<super::GetFileRequest>,
|
||||
) -> std::result::Result<tonic::Response<Self::GetFileStream>, tonic::Status>;
|
||||
/// Server streaming response type for the SubscribeEvents method.
|
||||
type SubscribeEventsStream: tonic::codegen::tokio_stream::Stream<
|
||||
Item = std::result::Result<super::ChangeEvent, tonic::Status>,
|
||||
> + std::marker::Send
|
||||
+ 'static;
|
||||
async fn subscribe_events(
|
||||
&self,
|
||||
request: tonic::Request<super::SubscribeEventsRequest>,
|
||||
) -> std::result::Result<tonic::Response<Self::SubscribeEventsStream>, tonic::Status>;
|
||||
}
|
||||
#[derive(Debug)]
|
||||
pub struct MusicFsServer<T> {
|
||||
inner: Arc<T>,
|
||||
accept_compression_encodings: EnabledCompressionEncodings,
|
||||
send_compression_encodings: EnabledCompressionEncodings,
|
||||
max_decoding_message_size: Option<usize>,
|
||||
max_encoding_message_size: Option<usize>,
|
||||
}
|
||||
impl<T> MusicFsServer<T> {
|
||||
pub fn new(inner: T) -> Self {
|
||||
Self::from_arc(Arc::new(inner))
|
||||
}
|
||||
pub fn from_arc(inner: Arc<T>) -> Self {
|
||||
Self {
|
||||
inner,
|
||||
accept_compression_encodings: Default::default(),
|
||||
send_compression_encodings: Default::default(),
|
||||
max_decoding_message_size: None,
|
||||
max_encoding_message_size: None,
|
||||
}
|
||||
}
|
||||
pub fn with_interceptor<F>(inner: T, interceptor: F) -> InterceptedService<Self, F>
|
||||
where
|
||||
F: tonic::service::Interceptor,
|
||||
{
|
||||
InterceptedService::new(Self::new(inner), interceptor)
|
||||
}
|
||||
/// Enable decompressing requests with the given encoding.
|
||||
#[must_use]
|
||||
pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
|
||||
self.accept_compression_encodings.enable(encoding);
|
||||
self
|
||||
}
|
||||
/// Compress responses with the given encoding, if the client supports it.
|
||||
#[must_use]
|
||||
pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
|
||||
self.send_compression_encodings.enable(encoding);
|
||||
self
|
||||
}
|
||||
/// Limits the maximum size of a decoded message.
|
||||
///
|
||||
/// Default: `4MB`
|
||||
#[must_use]
|
||||
pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
|
||||
self.max_decoding_message_size = Some(limit);
|
||||
self
|
||||
}
|
||||
/// Limits the maximum size of an encoded message.
|
||||
///
|
||||
/// Default: `usize::MAX`
|
||||
#[must_use]
|
||||
pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
|
||||
self.max_encoding_message_size = Some(limit);
|
||||
self
|
||||
}
|
||||
}
|
||||
impl<T, B> tonic::codegen::Service<http::Request<B>> for MusicFsServer<T>
|
||||
where
|
||||
T: MusicFs,
|
||||
B: Body + std::marker::Send + 'static,
|
||||
B::Error: Into<StdError> + std::marker::Send + 'static,
|
||||
{
|
||||
type Response = http::Response<tonic::body::Body>;
|
||||
type Error = std::convert::Infallible;
|
||||
type Future = BoxFuture<Self::Response, Self::Error>;
|
||||
fn poll_ready(
|
||||
&mut self,
|
||||
_cx: &mut Context<'_>,
|
||||
) -> Poll<std::result::Result<(), Self::Error>> {
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
fn call(&mut self, req: http::Request<B>) -> Self::Future {
|
||||
match req.uri().path() {
|
||||
"/musicfs.MusicFs/GetManifest" => {
|
||||
#[allow(non_camel_case_types)]
|
||||
struct GetManifestSvc<T: MusicFs>(pub Arc<T>);
|
||||
impl<T: MusicFs>
|
||||
tonic::server::ServerStreamingService<super::GetManifestRequest>
|
||||
for GetManifestSvc<T>
|
||||
{
|
||||
type Response = super::ManifestEntry;
|
||||
type ResponseStream = T::GetManifestStream;
|
||||
type Future =
|
||||
BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
|
||||
fn call(
|
||||
&mut self,
|
||||
request: tonic::Request<super::GetManifestRequest>,
|
||||
) -> Self::Future {
|
||||
let inner = Arc::clone(&self.0);
|
||||
let fut =
|
||||
async move { <T as MusicFs>::get_manifest(&inner, request).await };
|
||||
Box::pin(fut)
|
||||
}
|
||||
}
|
||||
let accept_compression_encodings = self.accept_compression_encodings;
|
||||
let send_compression_encodings = self.send_compression_encodings;
|
||||
let max_decoding_message_size = self.max_decoding_message_size;
|
||||
let max_encoding_message_size = self.max_encoding_message_size;
|
||||
let inner = self.inner.clone();
|
||||
let fut = async move {
|
||||
let method = GetManifestSvc(inner);
|
||||
let codec = tonic_prost::ProstCodec::default();
|
||||
let mut grpc = tonic::server::Grpc::new(codec)
|
||||
.apply_compression_config(
|
||||
accept_compression_encodings,
|
||||
send_compression_encodings,
|
||||
)
|
||||
.apply_max_message_size_config(
|
||||
max_decoding_message_size,
|
||||
max_encoding_message_size,
|
||||
);
|
||||
let res = grpc.server_streaming(method, req).await;
|
||||
Ok(res)
|
||||
};
|
||||
Box::pin(fut)
|
||||
}
|
||||
"/musicfs.MusicFs/GetFile" => {
|
||||
#[allow(non_camel_case_types)]
|
||||
struct GetFileSvc<T: MusicFs>(pub Arc<T>);
|
||||
impl<T: MusicFs> tonic::server::ServerStreamingService<super::GetFileRequest> for GetFileSvc<T> {
|
||||
type Response = super::FileChunk;
|
||||
type ResponseStream = T::GetFileStream;
|
||||
type Future =
|
||||
BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
|
||||
fn call(
|
||||
&mut self,
|
||||
request: tonic::Request<super::GetFileRequest>,
|
||||
) -> Self::Future {
|
||||
let inner = Arc::clone(&self.0);
|
||||
let fut =
|
||||
async move { <T as MusicFs>::get_file(&inner, request).await };
|
||||
Box::pin(fut)
|
||||
}
|
||||
}
|
||||
let accept_compression_encodings = self.accept_compression_encodings;
|
||||
let send_compression_encodings = self.send_compression_encodings;
|
||||
let max_decoding_message_size = self.max_decoding_message_size;
|
||||
let max_encoding_message_size = self.max_encoding_message_size;
|
||||
let inner = self.inner.clone();
|
||||
let fut = async move {
|
||||
let method = GetFileSvc(inner);
|
||||
let codec = tonic_prost::ProstCodec::default();
|
||||
let mut grpc = tonic::server::Grpc::new(codec)
|
||||
.apply_compression_config(
|
||||
accept_compression_encodings,
|
||||
send_compression_encodings,
|
||||
)
|
||||
.apply_max_message_size_config(
|
||||
max_decoding_message_size,
|
||||
max_encoding_message_size,
|
||||
);
|
||||
let res = grpc.server_streaming(method, req).await;
|
||||
Ok(res)
|
||||
};
|
||||
Box::pin(fut)
|
||||
}
|
||||
"/musicfs.MusicFs/SubscribeEvents" => {
|
||||
#[allow(non_camel_case_types)]
|
||||
struct SubscribeEventsSvc<T: MusicFs>(pub Arc<T>);
|
||||
impl<T: MusicFs>
|
||||
tonic::server::ServerStreamingService<super::SubscribeEventsRequest>
|
||||
for SubscribeEventsSvc<T>
|
||||
{
|
||||
type Response = super::ChangeEvent;
|
||||
type ResponseStream = T::SubscribeEventsStream;
|
||||
type Future =
|
||||
BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
|
||||
fn call(
|
||||
&mut self,
|
||||
request: tonic::Request<super::SubscribeEventsRequest>,
|
||||
) -> Self::Future {
|
||||
let inner = Arc::clone(&self.0);
|
||||
let fut = async move {
|
||||
<T as MusicFs>::subscribe_events(&inner, request).await
|
||||
};
|
||||
Box::pin(fut)
|
||||
}
|
||||
}
|
||||
let accept_compression_encodings = self.accept_compression_encodings;
|
||||
let send_compression_encodings = self.send_compression_encodings;
|
||||
let max_decoding_message_size = self.max_decoding_message_size;
|
||||
let max_encoding_message_size = self.max_encoding_message_size;
|
||||
let inner = self.inner.clone();
|
||||
let fut = async move {
|
||||
let method = SubscribeEventsSvc(inner);
|
||||
let codec = tonic_prost::ProstCodec::default();
|
||||
let mut grpc = tonic::server::Grpc::new(codec)
|
||||
.apply_compression_config(
|
||||
accept_compression_encodings,
|
||||
send_compression_encodings,
|
||||
)
|
||||
.apply_max_message_size_config(
|
||||
max_decoding_message_size,
|
||||
max_encoding_message_size,
|
||||
);
|
||||
let res = grpc.server_streaming(method, req).await;
|
||||
Ok(res)
|
||||
};
|
||||
Box::pin(fut)
|
||||
}
|
||||
_ => Box::pin(async move {
|
||||
let mut response = http::Response::new(tonic::body::Body::default());
|
||||
let headers = response.headers_mut();
|
||||
headers.insert(
|
||||
tonic::Status::GRPC_STATUS,
|
||||
(tonic::Code::Unimplemented as i32).into(),
|
||||
);
|
||||
headers.insert(
|
||||
http::header::CONTENT_TYPE,
|
||||
tonic::metadata::GRPC_CONTENT_TYPE,
|
||||
);
|
||||
Ok(response)
|
||||
}),
|
||||
}
|
||||
}
|
||||
}
|
||||
impl<T> Clone for MusicFsServer<T> {
|
||||
fn clone(&self) -> Self {
|
||||
let inner = self.inner.clone();
|
||||
Self {
|
||||
inner,
|
||||
accept_compression_encodings: self.accept_compression_encodings,
|
||||
send_compression_encodings: self.send_compression_encodings,
|
||||
max_decoding_message_size: self.max_decoding_message_size,
|
||||
max_encoding_message_size: self.max_encoding_message_size,
|
||||
}
|
||||
}
|
||||
}
|
||||
/// Generated gRPC service name
|
||||
pub const SERVICE_NAME: &str = "musicfs.MusicFs";
|
||||
impl<T> tonic::server::NamedService for MusicFsServer<T> {
|
||||
const NAME: &'static str = SERVICE_NAME;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
//! Generated protobuf + tonic bindings for the musicfs service.
|
||||
//!
|
||||
//! `buf generate` writes prost messages to `generated/musicfs/musicfs.rs` and
|
||||
//! the tonic service to `generated/musicfs/musicfs.tonic.rs`. The prost plugin
|
||||
//! appends `include!("musicfs.tonic.rs")` to the end of `musicfs.rs`, so
|
||||
//! including that one file here also pulls in the service. The only glue buf
|
||||
//! does not generate is the module named after the proto package (`musicfs`),
|
||||
//! which we provide below before re-exporting the types callers use.
|
||||
pub mod musicfs {
|
||||
include!("generated/musicfs/musicfs.rs");
|
||||
}
|
||||
|
||||
pub use musicfs::{
|
||||
ChangeEvent, FileChunk, GetFileRequest, GetManifestRequest, ManifestEntry, MusicMetadata,
|
||||
PictureDataRange, SubscribeEventsRequest,
|
||||
music_fs_server::{MusicFs, MusicFsServer},
|
||||
};
|
||||
@@ -0,0 +1,80 @@
|
||||
use crate::music::metadata::MusicMetadata;
|
||||
|
||||
/// One row of the in-memory manifest: every field the client needs to
|
||||
/// reconstruct an `Item` whose hash matches the server's hash.
|
||||
///
|
||||
/// Field semantics:
|
||||
/// - `id` is the server filesystem inode. The client uses it as the FUSE
|
||||
/// inode, which keeps `compute_hash` inputs identical on both sides.
|
||||
/// - `rel_path` is the path relative to the server's `--source` root, using
|
||||
/// `/` as separator. The client stores it as `Item.original_path` and uses
|
||||
/// it as the byte-source locator.
|
||||
/// - `mtime` / `ctime` / `crtime` are seconds since the Unix epoch. These are
|
||||
/// the exact three time fields `compute_hash` consumes.
|
||||
/// - `size` is the real on-disk file size in bytes (the client's virtual size
|
||||
/// is derived from `music_metadata.virtual_size(size)` when present).
|
||||
/// - `music_metadata` is the fully-encoded metadata produced server-side via
|
||||
/// `parse_music_metadata_for_path`. The client never parses audio.
|
||||
///
|
||||
/// Wire format conversion lives in `transport/grpc.rs` (`From<ManifestEntry>`
|
||||
/// for the generated proto type).
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ManifestEntry {
|
||||
pub id: u64,
|
||||
pub rel_path: String,
|
||||
pub size: u64,
|
||||
pub mtime: u64,
|
||||
pub ctime: u64,
|
||||
pub crtime: u64,
|
||||
pub music_metadata: Option<MusicMetadata>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn manifest_entry_clones_fields_into_copy() {
|
||||
let entry = ManifestEntry {
|
||||
id: 42,
|
||||
rel_path: "Artist/Album/track.flac".to_string(),
|
||||
size: 12345,
|
||||
mtime: 1_700_000_000,
|
||||
ctime: 1_699_999_000,
|
||||
crtime: 1_699_990_000,
|
||||
music_metadata: None,
|
||||
};
|
||||
|
||||
let cloned = entry.clone();
|
||||
assert_eq!(cloned.id, entry.id);
|
||||
assert_eq!(cloned.rel_path, entry.rel_path);
|
||||
assert_eq!(cloned.size, entry.size);
|
||||
assert_eq!(cloned.mtime, entry.mtime);
|
||||
assert_eq!(cloned.ctime, entry.ctime);
|
||||
assert_eq!(cloned.crtime, entry.crtime);
|
||||
assert!(cloned.music_metadata.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn manifest_entry_carries_music_metadata_when_present() {
|
||||
let mm = MusicMetadata {
|
||||
artist: vec!["Test Artist".to_string()],
|
||||
album: "Test Album".to_string(),
|
||||
track_title: "Test Title".to_string(),
|
||||
track_number: 3,
|
||||
..MusicMetadata::default()
|
||||
};
|
||||
let entry = ManifestEntry {
|
||||
id: 1,
|
||||
rel_path: "a/b.flac".to_string(),
|
||||
size: 100,
|
||||
mtime: 0,
|
||||
ctime: 0,
|
||||
crtime: 0,
|
||||
music_metadata: Some(mm.clone()),
|
||||
};
|
||||
|
||||
assert_eq!(entry.music_metadata.as_ref().unwrap().album, "Test Album");
|
||||
assert_eq!(entry.music_metadata.as_ref().unwrap().track_number, 3);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
pub mod manifest;
|
||||
pub mod range;
|
||||
pub mod state;
|
||||
pub mod transport;
|
||||
pub mod watcher;
|
||||
@@ -0,0 +1,175 @@
|
||||
/// Parsed `Range: bytes=...` request header.
|
||||
///
|
||||
/// Supported forms (RFC 7233):
|
||||
/// - `bytes=a-b` → `StartEnd(a, b)` (inclusive)
|
||||
/// - `bytes=a-` → `Start(a)`
|
||||
/// - `bytes=-N` → `Suffix(N)` (last N bytes)
|
||||
///
|
||||
/// Returns `None` if the header is missing, uses a unit other than `bytes`,
|
||||
/// or fails to parse. Multiple ranges (`bytes=a-b,c-d`) are not supported —
|
||||
/// the first is used and the rest ignored.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum ByteRange {
|
||||
StartEnd(u64, u64),
|
||||
Start(u64),
|
||||
Suffix(u64),
|
||||
}
|
||||
|
||||
pub fn parse_range_header(header: &str) -> Option<ByteRange> {
|
||||
let header = header.trim();
|
||||
let rest = header.strip_prefix("bytes=")?;
|
||||
let first = rest.split(',').next()?.trim();
|
||||
let (left, right) = first.split_once('-')?;
|
||||
let left = left.trim();
|
||||
let right = right.trim();
|
||||
|
||||
return match (left.is_empty(), right.is_empty()) {
|
||||
(false, false) => Some(ByteRange::StartEnd(left.parse().ok()?, right.parse().ok()?)),
|
||||
(false, true) => Some(ByteRange::Start(left.parse().ok()?)),
|
||||
(true, false) => Some(ByteRange::Suffix(right.parse().ok()?)),
|
||||
(true, true) => None,
|
||||
};
|
||||
}
|
||||
|
||||
/// Resolve a parsed range against a real file size, yielding an absolute
|
||||
/// `(start, length)` pair suitable for `seek + read`.
|
||||
///
|
||||
/// Returns `None` if the resolved range is unsatisfiable (e.g. start past
|
||||
/// end of file). The returned `start` is clamped to `[0, size]` and `length`
|
||||
/// is clamped to not exceed `size - start`.
|
||||
pub fn resolve_range(range: ByteRange, size: u64) -> Option<(u64, u64)> {
|
||||
let (start, end_inclusive) = match range {
|
||||
ByteRange::StartEnd(a, b) => {
|
||||
if a >= size || a > b {
|
||||
return None;
|
||||
}
|
||||
(a, b.min(size - 1))
|
||||
}
|
||||
ByteRange::Start(a) => {
|
||||
if a >= size {
|
||||
return None;
|
||||
}
|
||||
(a, size - 1)
|
||||
}
|
||||
ByteRange::Suffix(n) => {
|
||||
if n == 0 || size == 0 {
|
||||
return None;
|
||||
}
|
||||
// A suffix larger than the file clamps to the whole file.
|
||||
let start = size.saturating_sub(n);
|
||||
(start, size - 1)
|
||||
}
|
||||
};
|
||||
return Some((start, end_inclusive - start + 1));
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parse_start_end() {
|
||||
assert_eq!(
|
||||
parse_range_header("bytes=0-499"),
|
||||
Some(ByteRange::StartEnd(0, 499))
|
||||
);
|
||||
assert_eq!(
|
||||
parse_range_header("bytes=500-999"),
|
||||
Some(ByteRange::StartEnd(500, 999))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_start_only() {
|
||||
assert_eq!(
|
||||
parse_range_header("bytes=9500-"),
|
||||
Some(ByteRange::Start(9500))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_suffix() {
|
||||
assert_eq!(
|
||||
parse_range_header("bytes=-500"),
|
||||
Some(ByteRange::Suffix(500))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_ignores_additional_ranges_after_first() {
|
||||
assert_eq!(
|
||||
parse_range_header("bytes=0-499,1000-1499"),
|
||||
Some(ByteRange::StartEnd(0, 499))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_returns_none_for_non_bytes_unit() {
|
||||
assert_eq!(parse_range_header("items=0-4"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_returns_none_for_malformed() {
|
||||
assert_eq!(parse_range_header("not a range"), None);
|
||||
assert_eq!(parse_range_header("bytes="), None);
|
||||
assert_eq!(parse_range_header("bytes=-"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_start_end_in_bounds() {
|
||||
assert_eq!(
|
||||
resolve_range(ByteRange::StartEnd(0, 499), 1000),
|
||||
Some((0, 500))
|
||||
);
|
||||
assert_eq!(
|
||||
resolve_range(ByteRange::StartEnd(100, 199), 1000),
|
||||
Some((100, 100))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_start_end_clamps_end_to_size() {
|
||||
assert_eq!(
|
||||
resolve_range(ByteRange::StartEnd(900, 2000), 1000),
|
||||
Some((900, 100))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_start_end_unsatisfiable_when_start_past_size() {
|
||||
assert_eq!(resolve_range(ByteRange::StartEnd(1500, 2000), 1000), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_start_only() {
|
||||
assert_eq!(resolve_range(ByteRange::Start(900), 1000), Some((900, 100)));
|
||||
assert_eq!(resolve_range(ByteRange::Start(0), 1000), Some((0, 1000)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_start_unsatisfiable_when_past_size() {
|
||||
assert_eq!(resolve_range(ByteRange::Start(1500), 1000), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_suffix() {
|
||||
assert_eq!(
|
||||
resolve_range(ByteRange::Suffix(500), 1000),
|
||||
Some((500, 500))
|
||||
);
|
||||
assert_eq!(resolve_range(ByteRange::Suffix(1), 1000), Some((999, 1)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_suffix_larger_than_size_clamps_to_full_file() {
|
||||
assert_eq!(
|
||||
resolve_range(ByteRange::Suffix(2000), 1000),
|
||||
Some((0, 1000))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resolve_suffix_zero_unsatisfiable() {
|
||||
assert_eq!(resolve_range(ByteRange::Suffix(0), 1000), None);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
fs, io,
|
||||
path::{Path, PathBuf},
|
||||
sync::{Arc, Mutex},
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
|
||||
use crate::music::parse::parse_music_metadata_for_path;
|
||||
use crate::origins::attrs::FileAttrs;
|
||||
use crate::server::manifest::ManifestEntry;
|
||||
|
||||
/// Server-side entry for one file. Built once on startup from a directory
|
||||
/// scan and refreshed by the watcher on inotify events.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct FileEntry {
|
||||
pub abs_path: PathBuf,
|
||||
pub rel_path: String,
|
||||
pub attrs: FileAttrs,
|
||||
pub music_metadata: Option<crate::music::metadata::MusicMetadata>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct ServerState {
|
||||
inner: Arc<Mutex<HashMap<u64, FileEntry>>>,
|
||||
}
|
||||
|
||||
impl ServerState {
|
||||
pub fn new() -> Self {
|
||||
return ServerState {
|
||||
inner: Arc::new(Mutex::new(HashMap::new())),
|
||||
};
|
||||
}
|
||||
|
||||
/// Replace the entire map with a fresh scan of `source`. Used on startup
|
||||
/// and on watcher-driven reconciliations.
|
||||
pub fn replace_all(&self, source: &Path) -> io::Result<()> {
|
||||
let entries = scan_directory(source)?;
|
||||
let mut map = self.inner.lock().unwrap();
|
||||
map.clear();
|
||||
for (id, entry) in entries {
|
||||
map.insert(id, entry);
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
/// Snapshot the current state into a manifest. Order is by inode ascending
|
||||
/// so two scans of an unchanged library serialize identically.
|
||||
pub fn manifest(&self) -> Vec<ManifestEntry> {
|
||||
let map = self.inner.lock().unwrap();
|
||||
let mut entries: Vec<ManifestEntry> = map
|
||||
.iter()
|
||||
.map(|(id, file)| ManifestEntry {
|
||||
id: *id,
|
||||
rel_path: file.rel_path.clone(),
|
||||
size: file.attrs.size,
|
||||
mtime: secs(file.attrs.mtime),
|
||||
ctime: secs(file.attrs.ctime),
|
||||
crtime: secs(file.attrs.crtime),
|
||||
music_metadata: file.music_metadata.clone(),
|
||||
})
|
||||
.collect();
|
||||
entries.sort_by_key(|e| e.id);
|
||||
return entries;
|
||||
}
|
||||
|
||||
pub fn lookup(&self, id: u64) -> Option<FileEntry> {
|
||||
return self.inner.lock().unwrap().get(&id).cloned();
|
||||
}
|
||||
}
|
||||
|
||||
fn secs(time: SystemTime) -> u64 {
|
||||
return time
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0);
|
||||
}
|
||||
|
||||
fn scan_directory(source: &Path) -> io::Result<Vec<(u64, FileEntry)>> {
|
||||
let mut out = Vec::new();
|
||||
walk(source, source, &mut out)?;
|
||||
return Ok(out);
|
||||
}
|
||||
|
||||
fn walk(source: &Path, dir: &Path, out: &mut Vec<(u64, FileEntry)>) -> io::Result<()> {
|
||||
for entry in fs::read_dir(dir)? {
|
||||
let entry = entry?;
|
||||
let path = entry.path();
|
||||
let file_type = entry.file_type()?;
|
||||
if file_type.is_dir() {
|
||||
walk(source, &path, out)?;
|
||||
continue;
|
||||
}
|
||||
if !file_type.is_file() {
|
||||
continue;
|
||||
}
|
||||
|
||||
let metadata = entry.metadata()?;
|
||||
let rel_path = path
|
||||
.strip_prefix(source)
|
||||
.map(|p| p.to_path_buf())
|
||||
.unwrap_or_else(|_| path.clone())
|
||||
.to_string_lossy()
|
||||
.replace('\\', "/");
|
||||
|
||||
let inode = metadata.ino();
|
||||
let music_metadata = parse_music_metadata_for_path(&path);
|
||||
let file_entry = FileEntry {
|
||||
abs_path: path,
|
||||
rel_path,
|
||||
attrs: FileAttrs::from(&metadata),
|
||||
music_metadata,
|
||||
};
|
||||
out.push((inode, file_entry));
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io::Write;
|
||||
|
||||
#[test]
|
||||
fn replace_all_loads_files_into_state() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let source = tmp.path();
|
||||
|
||||
let mut f = fs::File::create(source.join("a.txt")).unwrap();
|
||||
f.write_all(b"hello").unwrap();
|
||||
drop(f);
|
||||
let mut f = fs::File::create(source.join("b.txt")).unwrap();
|
||||
f.write_all(b"world!").unwrap();
|
||||
drop(f);
|
||||
|
||||
let state = ServerState::new();
|
||||
state.replace_all(source).unwrap();
|
||||
|
||||
let manifest = state.manifest();
|
||||
assert_eq!(manifest.len(), 2);
|
||||
assert!(manifest.iter().any(|e| e.rel_path == "a.txt"));
|
||||
assert!(manifest.iter().any(|e| e.rel_path == "b.txt"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn manifest_entries_sorted_by_id_ascending() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let source = tmp.path();
|
||||
// Create files in any order; inode order is determined by the FS.
|
||||
for name in ["z.txt", "a.txt", "m.txt"] {
|
||||
fs::write(source.join(name), b"x").unwrap();
|
||||
}
|
||||
|
||||
let state = ServerState::new();
|
||||
state.replace_all(source).unwrap();
|
||||
|
||||
let ids: Vec<u64> = state.manifest().iter().map(|e| e.id).collect();
|
||||
let mut sorted = ids.clone();
|
||||
sorted.sort();
|
||||
assert_eq!(ids, sorted);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn replace_all_clears_existing_entries() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let source = tmp.path();
|
||||
fs::write(source.join("a.txt"), b"x").unwrap();
|
||||
|
||||
let state = ServerState::new();
|
||||
state.replace_all(source).unwrap();
|
||||
assert_eq!(state.manifest().len(), 1);
|
||||
|
||||
fs::remove_file(source.join("a.txt")).unwrap();
|
||||
state.replace_all(source).unwrap();
|
||||
assert_eq!(state.manifest().len(), 0);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,214 @@
|
||||
use std::{
|
||||
fs,
|
||||
io::{Read, Seek, SeekFrom},
|
||||
path::PathBuf,
|
||||
pin::Pin,
|
||||
};
|
||||
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
use tokio::sync::broadcast::Receiver;
|
||||
use tonic::{Request, Response, Status, transport::Server};
|
||||
|
||||
use crate::music::metadata::MusicMetadata;
|
||||
use crate::proto as proto_types;
|
||||
use crate::proto::MusicFs as MusicFsTrait;
|
||||
use crate::proto::{
|
||||
ChangeEvent as ProtoChangeEvent, FileChunk, GetFileRequest, GetManifestRequest, ManifestEntry,
|
||||
MusicFsServer, MusicMetadata as ProtoMusicMetadata, PictureDataRange, SubscribeEventsRequest,
|
||||
};
|
||||
use crate::server::manifest::ManifestEntry as DomainManifestEntry;
|
||||
use crate::server::state::ServerState;
|
||||
use crate::server::transport::{MusicTransport, TransportArgs};
|
||||
use crate::server::watcher::{ChangeEvent, ChangeKind};
|
||||
|
||||
pub struct GrpcTransport {
|
||||
args: TransportArgs,
|
||||
}
|
||||
|
||||
impl GrpcTransport {
|
||||
pub fn new(args: TransportArgs) -> Self {
|
||||
return GrpcTransport { args };
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl MusicTransport for GrpcTransport {
|
||||
async fn run(self: Box<Self>, state: ServerState, events: Receiver<ChangeEvent>) -> Result<()> {
|
||||
let service = MusicFsService {
|
||||
state,
|
||||
events: tokio::sync::Mutex::new(events),
|
||||
};
|
||||
let listen = self.args.listen;
|
||||
println!("musicfs-server gRPC listening on {listen}");
|
||||
Server::builder()
|
||||
.add_service(MusicFsServer::new(service))
|
||||
.serve(listen)
|
||||
.await?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
struct MusicFsService {
|
||||
state: ServerState,
|
||||
events: tokio::sync::Mutex<Receiver<ChangeEvent>>,
|
||||
}
|
||||
|
||||
type BoxStream<T> = Pin<Box<dyn tokio_stream::Stream<Item = Result<T, Status>> + Send>>;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl MusicFsTrait for MusicFsService {
|
||||
type GetManifestStream = BoxStream<ManifestEntry>;
|
||||
type GetFileStream = BoxStream<FileChunk>;
|
||||
type SubscribeEventsStream = BoxStream<ProtoChangeEvent>;
|
||||
|
||||
async fn get_manifest(
|
||||
&self,
|
||||
_request: Request<GetManifestRequest>,
|
||||
) -> Result<Response<Self::GetManifestStream>, Status> {
|
||||
let entries: Vec<DomainManifestEntry> = self.state.manifest();
|
||||
let stream = tokio_stream::iter(
|
||||
entries
|
||||
.into_iter()
|
||||
.map(|e| Ok::<ManifestEntry, Status>(e.into())),
|
||||
);
|
||||
return Ok(Response::new(Box::pin(stream)));
|
||||
}
|
||||
|
||||
async fn get_file(
|
||||
&self,
|
||||
request: Request<GetFileRequest>,
|
||||
) -> Result<Response<Self::GetFileStream>, Status> {
|
||||
let req = request.into_inner();
|
||||
let id = req.id;
|
||||
let entry = self
|
||||
.state
|
||||
.lookup(id)
|
||||
.ok_or_else(|| Status::not_found(format!("no file with id {id}")))?;
|
||||
let abs_path: PathBuf = entry.abs_path.clone();
|
||||
let total_size = entry.attrs.size;
|
||||
let start = if req.start == 0 && req.length == 0 {
|
||||
0u64
|
||||
} else {
|
||||
req.start
|
||||
};
|
||||
let length = if req.start == 0 && req.length == 0 {
|
||||
total_size
|
||||
} else if req.length == 0 {
|
||||
total_size.saturating_sub(start)
|
||||
} else {
|
||||
req.length
|
||||
};
|
||||
|
||||
let chunk =
|
||||
tokio::task::spawn_blocking(move || read_range_blocking(&abs_path, start, length))
|
||||
.await
|
||||
.map_err(|e| Status::internal(format!("join blocking read: {e}")))?
|
||||
.map_err(|e| Status::internal(format!("read file range: {e}")))?;
|
||||
|
||||
let chunk_stream = tokio_stream::iter(vec![Ok::<FileChunk, Status>(FileChunk {
|
||||
data: chunk,
|
||||
offset: start,
|
||||
total_size,
|
||||
})]);
|
||||
return Ok(Response::new(Box::pin(chunk_stream)));
|
||||
}
|
||||
|
||||
async fn subscribe_events(
|
||||
&self,
|
||||
_request: Request<SubscribeEventsRequest>,
|
||||
) -> Result<Response<Self::SubscribeEventsStream>, Status> {
|
||||
// Take an owned receiver (the guard is dropped at the end of this
|
||||
// statement) so it can move into the 'static spawned task, and so each
|
||||
// subscriber gets its own receiver rather than contending on one.
|
||||
let mut rx = self.events.lock().await.resubscribe();
|
||||
let (tx, rx_stream) =
|
||||
tokio::sync::mpsc::unbounded_channel::<Result<ProtoChangeEvent, Status>>();
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
match rx.recv().await {
|
||||
Ok(event) => {
|
||||
let proto = ProtoChangeEvent {
|
||||
kind: kind_to_string(event.kind),
|
||||
};
|
||||
if tx.send(Ok(proto)).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
|
||||
}
|
||||
}
|
||||
});
|
||||
let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx_stream);
|
||||
return Ok(Response::new(Box::pin(stream)));
|
||||
}
|
||||
}
|
||||
|
||||
fn read_range_blocking(
|
||||
path: &std::path::Path,
|
||||
start: u64,
|
||||
length: u64,
|
||||
) -> std::io::Result<Vec<u8>> {
|
||||
let mut file = fs::File::open(path)?;
|
||||
file.seek(SeekFrom::Start(start))?;
|
||||
let mut buf = vec![0u8; length as usize];
|
||||
let mut filled = 0usize;
|
||||
while filled < buf.len() {
|
||||
let n = file.read(&mut buf[filled..])?;
|
||||
if n == 0 {
|
||||
buf.truncate(filled);
|
||||
break;
|
||||
}
|
||||
filled += n;
|
||||
}
|
||||
return Ok(buf);
|
||||
}
|
||||
|
||||
fn kind_to_string(kind: ChangeKind) -> String {
|
||||
return match kind {
|
||||
ChangeKind::Create => "create".to_string(),
|
||||
ChangeKind::Modify => "modify".to_string(),
|
||||
ChangeKind::Remove => "remove".to_string(),
|
||||
};
|
||||
}
|
||||
|
||||
impl From<DomainManifestEntry> for ManifestEntry {
|
||||
fn from(entry: DomainManifestEntry) -> Self {
|
||||
return ManifestEntry {
|
||||
id: entry.id,
|
||||
rel_path: entry.rel_path,
|
||||
size: entry.size,
|
||||
mtime: entry.mtime,
|
||||
ctime: entry.ctime,
|
||||
crtime: entry.crtime,
|
||||
music_metadata: entry.music_metadata.map(Into::into),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
impl From<MusicMetadata> for ProtoMusicMetadata {
|
||||
fn from(mm: MusicMetadata) -> Self {
|
||||
return ProtoMusicMetadata {
|
||||
artist: mm.artist,
|
||||
album_artist: mm.album_artist,
|
||||
album: mm.album,
|
||||
track_number: mm.track_number,
|
||||
track_title: mm.track_title,
|
||||
other_tags: mm.other_tags,
|
||||
header: mm.header,
|
||||
picture_block_headers: mm.picture_block_headers,
|
||||
picture_data_ranges: mm
|
||||
.picture_data_ranges
|
||||
.into_iter()
|
||||
.map(|(offset, length)| PictureDataRange { offset, length })
|
||||
.collect(),
|
||||
real_audio_start: mm.real_audio_start,
|
||||
vorbis_comment_offset: mm.vorbis_comment_offset,
|
||||
vorbis_comment_length: mm.vorbis_comment_length,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(unused_imports)]
|
||||
use proto_types as _proto_types_anchor;
|
||||
@@ -0,0 +1,39 @@
|
||||
pub mod grpc;
|
||||
|
||||
use std::net::SocketAddr;
|
||||
|
||||
use anyhow::{Result, anyhow};
|
||||
use async_trait::async_trait;
|
||||
use tokio::sync::broadcast::Receiver;
|
||||
|
||||
use crate::server::state::ServerState;
|
||||
use crate::server::watcher::ChangeEvent;
|
||||
|
||||
/// Transport-agnostic musicfs server. A single implementation is wired in
|
||||
/// today (`grpc`); the factory in [`build`] is the seam where additional
|
||||
/// transports (HTTP/3, raw QUIC, in-process for tests, ...) get plugged in
|
||||
/// without touching the binary.
|
||||
///
|
||||
/// `run` consumes `self` — a transport serves exactly once. It receives the
|
||||
/// shared [`ServerState`] and a fresh broadcast receiver for change events.
|
||||
#[async_trait]
|
||||
pub trait MusicTransport: Send + Sync + 'static {
|
||||
async fn run(self: Box<Self>, state: ServerState, events: Receiver<ChangeEvent>) -> Result<()>;
|
||||
}
|
||||
|
||||
/// Arguments every transport understands. Transports may extend this with
|
||||
/// their own configuration via their constructors.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct TransportArgs {
|
||||
pub listen: SocketAddr,
|
||||
}
|
||||
|
||||
/// Construct the named transport. The list of accepted names is intentionally
|
||||
/// discoverable (a single match arm) so adding a transport means adding an
|
||||
/// arm here plus a module under `transport/`.
|
||||
pub fn build(name: &str, args: TransportArgs) -> Result<Box<dyn MusicTransport>> {
|
||||
return match name {
|
||||
"grpc" => Ok(Box::new(grpc::GrpcTransport::new(args))),
|
||||
other => Err(anyhow!("unknown transport: {other}")),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
use std::{
|
||||
path::PathBuf,
|
||||
sync::{Arc, Mutex},
|
||||
thread,
|
||||
};
|
||||
|
||||
use notify::{EventKind, RecursiveMode, Watcher};
|
||||
use tokio::sync::broadcast;
|
||||
|
||||
use crate::server::state::ServerState;
|
||||
|
||||
/// A change observed by the watcher. Pushed onto the broadcast channel for
|
||||
/// `/events` subscribers. The client treats these as wake-ups: correctness
|
||||
/// always rests on the subsequent `/manifest` hash diff.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ChangeEvent {
|
||||
pub kind: ChangeKind,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub enum ChangeKind {
|
||||
Create,
|
||||
Modify,
|
||||
Remove,
|
||||
}
|
||||
|
||||
/// Server-side file watcher. Owns a background std thread driving `notify`
|
||||
/// (inotify on Linux). On any filesystem event under `source` it:
|
||||
/// 1. Rebuilds the shared [`ServerState`] from a fresh scan.
|
||||
/// 2. Broadcasts a [`ChangeEvent`] so connected `/events` clients wake up.
|
||||
///
|
||||
/// The state rebuild is full-scan rather than incremental. This matches the
|
||||
/// existing `LocalOriginFileWatcher` pattern, keeps the watcher simple, and
|
||||
/// is cheap on a LAN-scale library. Hash-diff reconciliation on the client
|
||||
/// absorbs any over-reporting.
|
||||
pub struct ServerWatcher {
|
||||
_events_tx: broadcast::Sender<ChangeEvent>,
|
||||
_worker: Arc<Mutex<Option<thread::JoinHandle<()>>>>,
|
||||
}
|
||||
|
||||
impl ServerWatcher {
|
||||
/// Spawn the watcher. Returns the broadcast receiver that `/events`
|
||||
/// handlers subscribe to.
|
||||
pub fn spawn(source: PathBuf, state: ServerState) -> (Self, broadcast::Receiver<ChangeEvent>) {
|
||||
let (events_tx, events_rx) = broadcast::channel(64);
|
||||
let worker_tx = events_tx.clone();
|
||||
let handle = thread::spawn(move || {
|
||||
run_watcher_loop(source, state, worker_tx);
|
||||
});
|
||||
let watcher = ServerWatcher {
|
||||
_events_tx: events_tx,
|
||||
_worker: Arc::new(Mutex::new(Some(handle))),
|
||||
};
|
||||
return (watcher, events_rx);
|
||||
}
|
||||
}
|
||||
|
||||
fn run_watcher_loop(
|
||||
source: PathBuf,
|
||||
state: ServerState,
|
||||
events_tx: broadcast::Sender<ChangeEvent>,
|
||||
) {
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
let mut watcher = match notify::recommended_watcher(tx) {
|
||||
Ok(w) => w,
|
||||
Err(e) => {
|
||||
eprintln!("server watcher: failed to create inotify watcher: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(e) = watcher.watch(&source, RecursiveMode::Recursive) {
|
||||
eprintln!("server watcher: failed to watch {}: {e}", source.display());
|
||||
return;
|
||||
}
|
||||
|
||||
println!("server watcher: scanning {source:?} for changes");
|
||||
for res in rx {
|
||||
match res {
|
||||
Ok(event) => match event.kind {
|
||||
EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => {
|
||||
let kind = match event.kind {
|
||||
EventKind::Create(_) => ChangeKind::Create,
|
||||
EventKind::Modify(_) => ChangeKind::Modify,
|
||||
EventKind::Remove(_) => ChangeKind::Remove,
|
||||
_ => continue,
|
||||
};
|
||||
if let Err(e) = state.replace_all(&source) {
|
||||
eprintln!("server watcher: state refresh failed: {e}");
|
||||
continue;
|
||||
}
|
||||
let _ = events_tx.send(ChangeEvent { kind });
|
||||
}
|
||||
_ => {}
|
||||
},
|
||||
Err(e) => eprintln!("server watcher: inotify error: {e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user