Add MPD control socket integration with command interface

Displays the currently playing song as the bot's comment in Mumble.

Allows for controlling playback using text chat commands.

Cargo.lock

@@ -18,6 +18,12 @@ version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693"
 
+[[package]]
+name = "bufstream"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "40e38929add23cdf8a366df9b0e088953150724bcbe5fc330b0d8eb3b328eec8"
+
 [[package]]
 name = "byteorder"
 version = "1.3.4"
@@ -198,7 +204,7 @@ checksum = "fc587bc0ec293155d5bfa6b9891ec18a1e330c234f896ea47fbada4cadbe47e6"
 dependencies = [
  "cfg-if",
  "libc",
- "wasi",
+ "wasi 0.9.0+wasi-snapshot-preview1",
 ]
 
 [[package]]
@@ -320,6 +326,17 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "mpd"
+version = "0.0.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57a20784da57fa01bf7910a5da686d9f39ff37feaa774856b71f050e4331bf82"
+dependencies = [
+ "bufstream",
+ "rustc-serialize",
+ "time",
+]
+
 [[package]]
 name = "mumble-bot-rs"
 version = "0.1.0"
@@ -327,6 +344,7 @@ dependencies = [
  "argparse",
  "bytes",
  "futures",
+ "mpd",
  "mumble-protocol",
  "native-tls",
  "opus",
@@ -602,6 +620,12 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "rustc-serialize"
+version = "0.3.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dcf128d1287d2ea9d80910b5f1120d0b8eede3fbf1abe91c40d39ea7d51e6fda"
+
 [[package]]
 name = "schannel"
 version = "0.1.19"
@@ -687,6 +711,17 @@ dependencies = [
  "winapi 0.3.9",
 ]
 
+[[package]]
+name = "time"
+version = "0.1.44"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6db9e6914ab8b1ae1c260a4ae7a49b6c5611b40328a735b21862567685e73255"
+dependencies = [
+ "libc",
+ "wasi 0.10.0+wasi-snapshot-preview1",
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "tokio"
 version = "0.2.22"
@@ -764,6 +799,12 @@ version = "0.9.0+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cccddf32554fecc6acb585f82a32a72e28b48f8c4c1883ddfeeeaa96f7d8e519"
 
+[[package]]
+name = "wasi"
+version = "0.10.0+wasi-snapshot-preview1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1a143597ca7c7793eff794def352d41792a93c481eb1042423ff7ff72ba2c31f"
+
 [[package]]
 name = "winapi"
 version = "0.2.8"

Cargo.toml

@@ -17,3 +17,4 @@ tokio-util = { version = "0.3", features = ["codec", "udp"] }
 tokio-tls = "0.3"
 bytes = "0.5"
 
+mpd = "*"

src/commands.rs

@@ -0,0 +1,126 @@
+use crate::util::display_song;
+use mpd::search::Term;
+use mpd::search::Window;
+use mpd::Client;
+use mpd::Query;
+use std::collections::BTreeMap;
+
+fn paginate<K: Ord, V: Clone>(
+    pages: &mut BTreeMap<K, Vec<V>>,
+    page_size: usize,
+    key: K,
+    remaining: Vec<V>,
+) -> Vec<V> {
+    if remaining.len() > page_size {
+        let (result, remaining) = remaining.split_at(page_size);
+        pages.insert(key, remaining.to_vec());
+        result.to_vec()
+    } else {
+        pages.remove(&key);
+        remaining
+    }
+}
+
+pub fn parse_command(
+    conn: &mut Client,
+    pages: &mut BTreeMap<u32, Vec<String>>,
+    actor: u32,
+    msg: &str,
+) -> Option<String> {
+    let reply = run_command(conn, pages, actor, msg);
+    if reply.len() > 50 {
+        Some(format!(
+            "{} <br> Output has been paginated, use !more to view the rest.",
+            paginate(pages, 50, actor, reply).join("<br>")
+        ))
+    } else {
+        pages.remove(&actor);
+        Some(reply.join("<br>"))
+    }
+}
+
+fn run_command(
+    conn: &mut Client,
+    pages: &mut BTreeMap<u32, Vec<String>>,
+    actor: u32,
+    msg: &str,
+) -> Vec<String> {
+    let mut command = msg.split_whitespace();
+    match command.nth(0).unwrap() {
+        "!more" => {
+            if let Some(lines) = pages.get(&actor) {
+                lines.clone()
+            } else {
+                vec!["No more pages :(".to_string()]
+            }
+        }
+        "!next" | "!skip" => {
+            conn.next().unwrap();
+            vec!["Skipping".to_string()]
+        }
+        "!pause" => {
+            conn.pause(true).unwrap();
+            vec!["Pausing".to_string()]
+        }
+        "!play" => {
+            conn.play().unwrap();
+            vec!["Playing".to_string()]
+        }
+        "!toggle" => {
+            conn.toggle_pause().unwrap();
+            vec!["Toggling".to_string()]
+        }
+        "!search" | "!se" => {
+            let mut query = Query::new();
+            let query = command.fold(&mut query, |a, b| a.and(Term::Any, b));
+
+            if let Ok(result) = conn.search(&query, None) {
+                let display = result.iter().filter_map(display_song);
+                vec![format!("{} results:", result.len())]
+                    .into_iter()
+                    .chain(display)
+                    .collect()
+            } else {
+                vec!["Search failed".to_string()]
+            }
+        }
+        "!searchplay" | "!sp" => {
+            let mut query = Query::new();
+            let query = command.fold(&mut query, |a, b| a.and(Term::Any, b));
+
+            if let Ok(result) = conn.search(&query, Window::from((0, 1))) {
+                let target_song = result.first().unwrap();
+                let position = conn
+                    .queue()
+                    .unwrap()
+                    .iter()
+                    .enumerate()
+                    .find_map(|(i, song)| {
+                        if song.file == target_song.file {
+                            Some(i)
+                        } else {
+                            None
+                        }
+                    });
+                if let Some(pos) = position {
+                    conn.switch(pos as u32).unwrap();
+                    vec![format!("Playing {}", display_song(&target_song).unwrap())]
+                } else {
+                    vec![format!(
+                        "Song not found in queue: {}",
+                        display_song(&target_song).unwrap()
+                    )]
+                }
+            } else {
+                vec!["Search failed".to_string()]
+            }
+        }
+        "!queue" => conn
+            .queue()
+            .unwrap()
+            .iter()
+            .filter_map(display_song)
+            .collect(),
+        _ => vec![],
+    }
+}

src/main.rs

@@ -1,18 +1,24 @@
 extern crate argparse;
 extern crate bytes;
 extern crate futures;
+extern crate mpd;
 extern crate mumble_protocol;
 extern crate opus;
 
+mod commands;
+mod util;
+
 use argparse::ArgumentParser;
 use argparse::Store;
 use argparse::StoreTrue;
 use bytes::Bytes;
+use commands::parse_command;
 use futures::channel::oneshot;
 use futures::join;
 use futures::stream::SplitSink;
 use futures::SinkExt;
 use futures::StreamExt;
+use mpd::Client;
 use mumble_protocol::control::msgs;
 use mumble_protocol::control::ClientControlCodec;
 use mumble_protocol::control::ControlPacket;
@@ -23,6 +29,7 @@ use mumble_protocol::voice::VoicePacketPayload;
 use mumble_protocol::Clientbound;
 use mumble_protocol::Serverbound;
 use opus::Encoder;
+use std::collections::BTreeMap;
 use std::convert::Into;
 use std::convert::TryInto;
 use std::fs::File;
@@ -38,6 +45,7 @@ use tokio::sync::mpsc;
 use tokio_tls::TlsConnector;
 use tokio_util::codec::Decoder;
 use tokio_util::udp::UdpFramed;
+use util::display_song;
 
 struct AudioSettings {
     fifo_path: String,
@@ -49,6 +57,7 @@ async fn connect(
     server_host: String,
     user_name: String,
     accept_invalid_cert: bool,
+    mpd_addr: String,
     crypt_state_sender: oneshot::Sender<ClientCryptState>,
 ) {
     // Wrap crypt_state_sender in Option, so we can call it only once
@@ -82,6 +91,9 @@ async fn connect(
     msg.set_opus(true);
     sink.send(msg.into()).await.unwrap();
 
+    // Connect to MPD
+    let mut conn = Client::connect(mpd_addr).unwrap();
+
     println!("Logging in..");
     let mut crypt_state = None;
 
@@ -105,6 +117,8 @@ async fn connect(
         }
     });
 
+    let mut paged_results = BTreeMap::new();
+
     while let Some(i) = rx.recv().await {
         match i {
             Some(packet) => match packet.unwrap() {
@@ -115,10 +129,17 @@ async fn connect(
                         msg.get_message()
                     );
                     // Send reply back to server
-                    let mut response = msgs::TextMessage::new();
-                    response.mut_session().push(msg.get_actor());
-                    response.set_message(msg.take_message());
-                    sink.send(response.into()).await.unwrap();
+                    if let Some(reply) = parse_command(
+                        &mut conn,
+                        &mut paged_results,
+                        msg.get_actor(),
+                        &msg.take_message(),
+                    ) {
+                        let mut response = msgs::TextMessage::new();
+                        response.mut_session().push(msg.get_actor());
+                        response.set_message(reply);
+                        sink.send(response.into()).await.unwrap();
+                    }
                 }
                 ControlPacket::CryptSetup(msg) => {
                     // Wait until we're fully connected before initiating UDP voice
@@ -150,6 +171,17 @@ async fn connect(
                 _ => {}
             },
             None => {
+                // Update status
+                let status = conn
+                    .currentsong()
+                    .ok()
+                    .flatten()
+                    .map(|s| display_song(&s))
+                    .flatten()
+                    .unwrap_or("".to_string());
+                let mut response = msgs::UserState::new();
+                response.set_comment(status);
+                sink.send(response.into()).await.unwrap();
                 sink.send(msgs::Ping::new().into()).await.unwrap();
             }
         }
@@ -291,6 +323,7 @@ async fn main() {
     let mut accept_invalid_cert = false;
     let mut bitrate = 96_000i32;
     let mut fifo_path = "".to_string();
+    let mut mpd_addr = "127.0.0.1:6600".to_string();
     {
         let mut ap = ArgumentParser::new();
         ap.set_description("Mumble bot for streaming music from MPD");
@@ -315,6 +348,11 @@ async fn main() {
             .required();
         ap.refer(&mut bitrate)
             .add_option(&["--bitrate"], Store, "Opus encoder bitrate (bits/s)");
+        ap.refer(&mut mpd_addr).add_option(
+            &["--mpd-addr"],
+            Store,
+            "MPD TCP control socket address",
+        );
         ap.parse_args_or_exit();
     }
     let server_addr = (server_host.as_ref(), server_port)
@@ -334,6 +372,7 @@ async fn main() {
             server_host,
             user_name,
             accept_invalid_cert,
+            mpd_addr,
             crypt_state_sender,
         ),
         handle_udp(

src/util.rs

@@ -0,0 +1,11 @@
+use mpd::Song;
+
+pub fn display_song(song: &Song) -> Option<String> {
+    song.title.to_owned().map(|title| {
+        if let Some(artist) = song.tags.get("Artist") {
+            format!("<i>{}</i> - <b>{}</b>", artist, title)
+        } else {
+            title
+        }
+    })
+}