WIP: centralize mpv broker

.gitignore

@@ -1 +1,2 @@
 /target
+result

Cargo.toml

@@ -2,6 +2,9 @@
 name = "greg-ng"
 version = "0.1.0"
 edition = "2021"
+license = "MIT"
+authors = ["oysteikt@pvv.ntnu.no"]
+readme = "README.md"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -17,3 +20,8 @@ serde_json = "1.0.105"
 tokio = { version = "1.32.0", features = ["full"] }
 tower = { version = "0.4.13", features = ["full"] }
 tower-http = { version = "0.4.3", features = ["full"] }
+
+[profile.release]
+strip = true
+lto = true
+codegen-units = 1

README.md

@@ -1,3 +1,21 @@
 # Greg-ng
 
 New implementation of https://github.com/Programvareverkstedet/grzegorz
+
+## Test it out
+
+```sh
+# NixOS
+nix run "git+https://git.pvv.ntnu.no/Projects/greg-ng#" -- --mpv-socket-path /tmp/mpv.sock
+
+# Other (after git clone and rust toolchain has been set up)
+cargo run -- --mpv-socket-path /tmp/mpv.sock
+```
+
+See also https://git.pvv.ntnu.no/Projects/grzegorz-clients for frontend alternatives
+
+## Debugging
+
+```sh
+RUST_LOG=greg_ng=trace,mpvipc=trace cargo run -- --mpv-socket-path /tmp/mpv.sock
+```

default.nix

@@ -0,0 +1,34 @@
+{
+  lib
+, fetchFromGitHub
+, rustPlatform
+, makeWrapper
+, mpv
+}:
+
+rustPlatform.buildRustPackage rec {
+  pname = "greg-ng";
+  version = "0.1.0";
+  src = lib.cleanSource ./.;
+
+  nativeBuildInputs = [ makeWrapper ];
+
+  cargoLock = {
+    lockFile = ./Cargo.lock;
+    outputHashes = {
+      "mpvipc-async-0.1.0" = "sha256-2TQ2d4q9/DTxTZe9kOAoKBhsmegRZw32x3G2hbluS6U=";
+    };
+  };
+
+  postInstall = ''
+    wrapProgram $out/bin/greg-ng \
+      --prefix PATH : '${lib.makeBinPath [ mpv ]}'
+  '';
+
+  meta = with lib; {
+    license = licenses.mit;
+    maintainers = with maintainers; [ h7x4 ];
+    platforms = platforms.linux ++ platforms.darwin;
+    mainProgram = "greg-ng";
+  };
+}

flake.nix

@@ -8,6 +8,8 @@
 
   outputs = { self, nixpkgs, rust-overlay }:
   let
+    inherit (nixpkgs) lib;
+
     systems = [
       "x86_64-linux"
       "aarch64-linux"
@@ -15,7 +17,8 @@
       "aarch64-darwin"
       "armv7l-linux"
     ];
-    forAllSystems = f: nixpkgs.lib.genAttrs systems (system: let
+
+    forAllSystems = f: lib.genAttrs systems (system: let
       pkgs = import nixpkgs {
         inherit system;
         overlays = [
@@ -27,6 +30,16 @@
       toolchain = rust-bin.stable.latest.default;
     in f system pkgs toolchain);
   in {
+    apps = forAllSystems (system: pkgs: _: {
+      default = self.apps.${system}.greg-ng;
+      greg-ng = let
+        package = self.packages.${system}.greg-ng;
+      in {
+        type = "app";
+        program = lib.getExe package;
+      };
+    });
+
     devShells = forAllSystems (system: pkgs: toolchain: {
       default = pkgs.mkShell {
         nativeBuildInputs = [
@@ -37,5 +50,10 @@
         RUST_SRC_PATH = "${toolchain}/lib/rustlib/src/rust/library";
       };
     });
+
+    packages = forAllSystems (system: pkgs: _: {
+      default = self.packages.${system}.greg-ng;
+      greg-ng = pkgs.callPackage ./default.nix { };
+    });
   };
-}
+}

src/mpv_broker.rs

@@ -0,0 +1,154 @@
+use std::{fs::create_dir_all, path::Path};
+
+use anyhow::Context;
+use mpvipc_async::{Mpv, MpvCommand, Event as MpvEvent};
+use tokio::{process::{Child, Command}, sync::{broadcast::{Receiver as BroadcastReceiver, Sender as BroadcastSender}, mpsc::{Receiver as MpscReceiver, Sender as MpscSender}}};
+
+#[derive(Debug)]
+pub struct MpvBroker {
+    mpv: Mpv,
+    command_channel: MpscReceiver<MpvCommand>,
+    event_listeners: BroadcastSender<MpvEvent>,
+}
+
+impl MpvBroker {
+    pub fn new(
+        mpv: Mpv,
+        command_channel: MpscReceiver<MpvCommand>,
+        event_listeners: BroadcastSender<MpvEvent>,
+    ) -> Self {
+        Self {
+            mpv,
+            command_channel,
+            event_listeners,
+        }
+    }
+
+    pub async fn run(&mut self) -> anyhow::Result<()> {
+        loop {
+            tokio::select! {
+                Some(command) = self.command_channel.recv() => {
+                    self.mpv.run_command(command)?;
+                }
+                Ok(event) = async { self.mpv.event_listen() } => {
+                    self.event_listeners.send(event)?;
+                }
+            }
+        }
+    }
+}
+
+pub struct MpvConnectionArgs {
+    pub socket_path: String,
+    pub executable_path: Option<String>,
+    pub auto_start: bool,
+    pub force_auto_start: bool,
+}
+
+pub async fn connect_to_mpv(args: &MpvConnectionArgs) -> anyhow::Result<(Mpv, Option<Child>)> {
+    log::debug!("Connecting to mpv");
+
+    debug_assert!(
+        !args.force_auto_start || args.auto_start,
+        "force_auto_start requires auto_start"
+    );
+
+    let socket_path = Path::new(&args.socket_path);
+
+    if !socket_path.exists() {
+        log::debug!("Mpv socket not found at {}", &args.socket_path);
+        if !args.auto_start {
+            panic!("Mpv socket not found at {}", &args.socket_path);
+        }
+
+        log::debug!("Ensuring parent dir of mpv socket exists");
+        let parent_dir = Path::new(&args.socket_path)
+            .parent()
+            .context("Failed to get parent dir of mpv socket")?;
+
+        if !parent_dir.is_dir() {
+            create_dir_all(parent_dir).context("Failed to create parent dir of mpv socket")?;
+        }
+    } else {
+        log::debug!("Existing mpv socket found at {}", &args.socket_path);
+        if args.force_auto_start {
+            log::debug!("Removing mpv socket");
+            std::fs::remove_file(&args.socket_path)?;
+        }
+    }
+
+    let process_handle = if args.auto_start {
+        log::info!("Starting mpv with socket at {}", &args.socket_path);
+
+        // TODO: try to fetch mpv from PATH
+        Some(
+            Command::new(args.executable_path.as_deref().unwrap_or("mpv"))
+                .arg(format!("--input-ipc-server={}", &args.socket_path))
+                .arg("--idle")
+                .arg("--force-window")
+                // .arg("--fullscreen")
+                // .arg("--no-terminal")
+                // .arg("--load-unsafe-playlists")
+                .arg("--keep-open") // Keep last frame of video on end of video
+                .arg("--really-quiet")
+                .spawn()
+                .context("Failed to start mpv")?,
+        )
+    } else {
+        None
+    };
+
+    // Wait for mpv to create the socket
+    if tokio::time::timeout(tokio::time::Duration::from_millis(500), async {
+        while !&socket_path.exists() {
+            log::debug!("Waiting for mpv socket at {}", &args.socket_path);
+            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .is_err()
+    {
+        return Err(anyhow::anyhow!(
+            "Failed to connect to mpv socket: {}",
+            &args.socket_path
+        ));
+    }
+
+    Ok((
+        Mpv::connect(&args.socket_path).context(format!(
+            "Failed to connect to mpv socket: {}",
+            &args.socket_path
+        ))?,
+        process_handle,
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use mpvipc_async::MpvCommand;
+    use tokio::sync::{broadcast, mpsc};
+
+    #[tokio::test]
+    async fn test_run() -> anyhow::Result<()> {
+        let (command_tx, command_rx) = mpsc::channel(1);
+        let (event_tx, _) = broadcast::channel(1);
+
+        let (mpv, _) = connect_to_mpv(&MpvConnectionArgs {
+            socket_path: "/tmp/mpv-test.sock".to_string(),
+            executable_path: None,
+            auto_start: true,
+            force_auto_start: true,
+        }).await?;
+
+        let mut broker = MpvBroker::new(mpv, command_rx, event_tx);
+        let broker_handle = tokio::spawn(async move {
+          broker.run().await.unwrap();
+        });
+
+        let _ = command_tx.send(MpvCommand::PlaylistClear).await;
+        let _ = broker_handle.await.unwrap();
+
+        Ok(())
+    }
+}