Add playlist feature and default channel flag

3 files changed, 311 insertions, 111 deletions

diff --git a/src/main.rs b/src/main.rs
index 7cb0aff..6d6c00e 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -7,17 +7,18 @@ use futures::{
     future::{FutureExt, TryFutureExt},
 };
 
-use futures01::{future::Future, stream::Stream, sync::mpsc};
+use futures01::future::Future;
 use structopt::clap::AppSettings;
 use structopt::StructOpt;
 
 use tsclientlib::{
-    events::Event, ChannelId, ConnectOptions, Connection, ConnectionLock, DisconnectOptions,
-    Event::ConEvents, Identity, MessageTarget,
+    events::Event, ConnectOptions, Connection, ConnectionLock, Event::ConEvents,
+    Identity, MessageTarget,
 };
 
 use log::error;
 
+mod playlist;
 mod state;
 use state::State;
 
@@ -39,6 +40,12 @@ struct Args {
     )]
     id_path: Option<PathBuf>,
     #[structopt(
+        short = "c",
+        long = "channel",
+        help = "The channel the bot should connect to"
+    )]
+    default_channel: Option<String>,
+    #[structopt(
         short = "v",
         long = "verbose",
         help = "Print the content of all packets",
@@ -70,7 +77,7 @@ async fn async_main() {
         Identity::create().expect("Failed to create id")
     };
 
-    let con_config = ConnectOptions::new(args.address)
+    let mut con_config = ConnectOptions::new(args.address)
         .version(tsclientlib::Version::Linux_3_3_2)
         .name(String::from("PokeBot"))
         .identity(id)
@@ -78,13 +85,17 @@ async fn async_main() {
         .log_packets(args.verbose >= 2)
         .log_udp_packets(args.verbose >= 3);
 
+    if let Some(channel) = args.default_channel {
+        con_config = con_config.channel(channel);
+    }
+
     //let (disconnect_send, disconnect_recv) = mpsc::unbounded();
     let conn = Connection::new(con_config).compat().await.unwrap();
 
-    let mut state = State::new(conn.clone());
+    let state = State::new(conn.clone());
     {
         let packet = conn.lock().server.set_subscribed(true);
-        conn.send_packet(packet).compat().await;
+        conn.send_packet(packet).compat().await.unwrap();
     }
     //con.add_on_disconnect(Box::new( || {
     //disconnect_send.unbounded_send(()).unwrap()
@@ -104,19 +115,19 @@ async fn async_main() {
     loop {
         state.poll().await;
     }
-    let ctrl_c = tokio_signal::ctrl_c().flatten_stream();
+    //let ctrl_c = tokio_signal::ctrl_c().flatten_stream();
 
     //let dc_fut = disconnect_recv.into_future().compat().fuse();
     //let ctrlc_fut = ctrl_c.into_future().compat().fuse();
     //ctrlc_fut.await.map_err(|(e, _)| e).unwrap();
 
-    conn.disconnect(DisconnectOptions::new())
-        .compat()
-        .await
-        .unwrap();
+    //conn.disconnect(DisconnectOptions::new())
+        //.compat()
+        //.await
+        //.unwrap();
 
     // TODO Should not be required
-    std::process::exit(0);
+    //std::process::exit(0);
 }
 
 fn handle_event<'a>(state: &State, conn: &ConnectionLock<'a>, event: &Event) {
@@ -158,7 +169,13 @@ fn handle_event<'a>(state: &State, conn: &ConnectionLock<'a>, event: &Event) {
                                         conn.to_mut().set_name("PokeBot - Loading").map_err(|_| ()),
                                     );
                                     let trimmed = url[5..url.len() - 6].to_owned();
-                                    state.add_audio(trimmed);
+                                    let inner_state = state.clone();
+                                    tokio::spawn(
+                                        async move { inner_state.add_audio(trimmed).await }
+                                            .unit_error()
+                                            .boxed()
+                                            .compat(),
+                                    );
                                 } else {
                                     invalid = true;
                                 }
@@ -175,12 +192,29 @@ fn handle_event<'a>(state: &State, conn: &ConnectionLock<'a>, event: &Event) {
                         }
                         Some("volume") => {
                             if let Ok(volume) = f64::from_str(tokens[1]) {
-                                state.volume(volume / 100.0);
+                                if 0.0 <= volume && volume <= 100.0 {
+                                    state.volume(volume);
+                                } else {
+                                    tokio::spawn(
+                                        conn.to_mut()
+                                            .send_message(
+                                                MessageTarget::Channel,
+                                                "Volume must be between 0 and 100",
+                                            )
+                                            .map_err(|_| ()),
+                                    );
+                                }
                             }
                         }
                         Some("play") => {
                             state.play();
                         }
+                        Some("skip") => {
+                            state.next();
+                        }
+                        Some("clear") => {
+                            state.clear();
+                        }
                         Some("pause") => {
                             state.pause();
                         }
diff --git a/src/playlist.rs b/src/playlist.rs
new file mode 100644
index 0000000..e0df816
--- /dev/null
+++ b/src/playlist.rs
@@ -0,0 +1,68 @@
+pub struct Playlist {
+    data: Vec<Option<AudioRequest>>,
+    read: usize,
+    write: usize,
+    is_full: bool,
+}
+
+impl Playlist {
+    pub fn new() -> Self {
+        Self {
+            data: Vec::with_capacity(3),
+            read: 0,
+            write: 0,
+            is_full: false,
+        }
+    }
+
+    pub fn push(&mut self, req: AudioRequest) -> bool {
+        if self.is_full {
+            return false;
+        }
+
+        if self.data.len() < self.data.capacity() {
+            self.data.push(Some(req));
+        } else {
+            self.data[self.write] = Some(req);
+        }
+
+        self.write = (self.write + 1) % self.data.capacity();
+
+        if self.write == self.read {
+            self.is_full = true;
+        }
+
+        true
+    }
+
+    pub fn is_empty(&self) -> bool {
+        !self.is_full && self.write == self.read
+    }
+
+    pub fn is_full(&self) -> bool {
+        self.is_full
+    }
+
+    pub fn pop(&mut self) -> Option<AudioRequest> {
+        if self.is_empty() {
+            None
+        } else {
+            self.read += 1;
+            self.is_full = false;
+            self.data[self.read].take()
+        }
+    }
+
+    pub fn clear(&mut self) {
+        self.data.clear();
+        self.read = 0;
+        self.write = 0;
+        self.is_full = false;
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct AudioRequest {
+    pub title: String,
+    pub address: String,
+}
diff --git a/src/state.rs b/src/state.rs
index dfccd9a..8c6a290 100644
--- a/src/state.rs
+++ b/src/state.rs
@@ -1,27 +1,26 @@
-use std::error::Error;
-use std::process::{Command, Stdio};
-use std::sync::Arc;
-
-use futures::{
-    compat::Future01CompatExt,
-    future::{FutureExt, TryFutureExt},
-};
-use futures01::{future::Future, sink::Sink};
+use std::process::Command;
+use std::sync::{Arc, Mutex};
 
+use futures::compat::Future01CompatExt;
+use futures01::{future::Future, sink::Sink};
 use futures_util::stream::StreamExt;
-use tsclientlib::{Connection, MessageTarget, DisconnectOptions};
+use tokio_process::CommandExt;
+
+use tsclientlib::{Connection, DisconnectOptions, MessageTarget};
 
 use gst::prelude::*;
 use gstreamer as gst;
 use gstreamer_app as gst_app;
-use gstreamer_audio as gst_audio;
 
 use byte_slice_cast::*;
 
+use crate::playlist::{AudioRequest, Playlist};
+
 #[derive(Clone)]
 pub struct State {
     conn: Arc<Connection>,
     pipeline: Arc<gst::Pipeline>,
+    playlist: Arc<Mutex<Playlist>>,
 }
 
 impl State {
@@ -68,6 +67,16 @@ impl State {
             };
 
             if is_audio {
+                let prev_volume = if let Some(volume) = pipeline.get_by_name("volume") {
+                    volume
+                        .get_property("volume")
+                        .expect("Can get volume property")
+                        .get_some::<f64>()
+                        .unwrap_or(0.1)
+                } else {
+                    0.1
+                };
+
                 let names = [
                     "audio-converter",
                     "volume",
@@ -78,8 +87,10 @@ impl State {
                 for element in pipeline.iterate_elements() {
                     if let Ok(element) = element {
                         if names.contains(&&*element.get_name()) {
-                            element.set_state(gst::State::Null);
-                            pipeline.remove_many(&[&element]);
+                            element.set_state(gst::State::Null).unwrap();
+                            pipeline
+                                .remove_many(&[&element])
+                                .expect("Can remove element");
                         }
                     }
                 }
@@ -92,9 +103,12 @@ impl State {
                 let opus_enc = gst::ElementFactory::make("opusenc", Some("opus-encoder")).unwrap();
                 let sink = gst::ElementFactory::make("appsink", Some("app-sink")).unwrap();
 
-                sink.set_property("async", &false);
+                sink.set_property("async", &false)
+                    .expect("Can make app-sink async");
 
-                volume.set_property("volume", &0.2);
+                volume
+                    .set_property("volume", &prev_volume)
+                    .expect("Can change volume");
 
                 {
                     let elements = &[&convert, &volume, &resample, &opus_enc, &sink];
@@ -109,7 +123,7 @@ impl State {
 
                 let appsink = sink
                     .dynamic_cast::<gst_app::AppSink>()
-                    .expect("Sink element is expected to be an appsink!");
+                    .expect("Sink is an Appsink");
 
                 appsink.set_caps(Some(&gst::Caps::new_simple("audio/x-opus", &[])));
 
@@ -144,7 +158,7 @@ impl State {
                                 .get_packet_sink()
                                 .send(packet)
                                 .map(|_| ())
-                                .map_err(|e| println!("Failed to send voice packet"));
+                                .map_err(|e| println!("Failed to send voice packet: {}", e));
 
                             tokio::run(send_packet);
 
@@ -165,68 +179,77 @@ impl State {
         Self {
             conn: conn_arc,
             pipeline: Arc::new(pipeline),
+            playlist: Arc::new(Mutex::new(Playlist::new())),
         }
     }
 
-    pub fn add_audio<'a>(&self, uri: String) {
+    pub async fn add_audio(&self, url: String) {
+        if self
+            .playlist
+            .lock()
+            .expect("Mutex was not poisoned")
+            .is_full()
+        {
+            self.send_message(MessageTarget::Channel, "Playlist is full");
+            return;
+        }
+
         let ytdl_args = [
             "--no-playlist",
             "-f",
             "bestaudio/best",
             "-g",
-            &uri,
+            "--get-filename",
             "-o",
-            "-",
+            "%(title)s",
+            &url,
         ];
 
-        let ytdl_output = Command::new("youtube-dl")
+        let output = Command::new("youtube-dl")
             .args(&ytdl_args)
-            .stdin(Stdio::null())
-            .output()
-            .unwrap();
+            .output_async()
+            .compat()
+            .await
+            .expect("youtube-dl is runnable");
 
-        if ytdl_output.status.success() == false {
-            tokio::spawn(
-                self.conn
-                .lock()
-                .to_mut()
-                .set_name("PokeBot")
-                .map_err(|_| println!("Failed to change name")),
-            );
-            tokio::spawn(
-                self.conn
-                .lock()
-                .to_mut()
-                .send_message(MessageTarget::Channel, "Failed to load url")
-                .map_err(|_| println!("Failed to change name")),
-            );
+        if output.status.success() == false {
+            self.set_name("PokeBot");
+            self.send_message(MessageTarget::Channel, "Failed to load url");
             return;
         }
-        let dl_url: &str = &String::from_utf8(ytdl_output.stderr).unwrap();
-
-        self.pipeline
-            .set_state(gst::State::Ready)
-            .expect("Can set state to ready");
-
-        let http_src = self
-            .pipeline
-            .get_by_name("http-source")
-            .expect("Http source should be registered");
-        http_src
-            .set_property("location", &dl_url)
-            .expect("Can set location on http_dl");
 
-        self.pipeline
-            .set_state(gst::State::Playing)
-            .expect("Can set state to playing");
-
-        tokio::spawn(
-            self.conn
+        let output_string = String::from_utf8(output.stdout).unwrap();
+        let output_lines = output_string.lines().collect::<Vec<_>>();
+        let address = output_lines[0];
+        let title = output_lines[1];
+
+        let req = AudioRequest {
+            title: title.to_owned(),
+            address: address.to_owned(),
+        };
+
+        if gst::State::Null == self.pipeline.get_state(gst::ClockTime(None)).1 {
+            self.set_name("PokeBot - Playing");
+            self.start_audio(req);
+        } else {
+            self.set_name("PokeBot - Playing");
+
+            let title = req.title.clone();
+            if self
+                .playlist
                 .lock()
-                .to_mut()
-                .set_name("PokeBot - Playing")
-                .map_err(|_| println!("Failed to change name")),
-        );
+                .expect("Mutex was not poisoned")
+                .push(req)
+                == false
+            {
+                self.send_message(MessageTarget::Channel, "Playlist is full");
+            } else {
+                self.send_message(
+                    MessageTarget::Channel,
+                    &format!("Added '{}' to the playlist", title),
+                );
+            }
+        }
     }
 
     pub async fn poll(&self) {
@@ -239,8 +262,6 @@ impl State {
         while let Some(msg) = messages.next().await {
             use gst::MessageView;
 
-            // Determine whether we want to quit: on EOS or error message
-            // we quit, otherwise simply continue.
             match msg.view() {
                 MessageView::Eos(..) => break,
                 MessageView::Error(err) => {
@@ -256,39 +277,96 @@ impl State {
             };
         }
 
-        tokio::spawn(
-            self.conn
-                .lock()
-                .to_mut()
-                .set_name("PokeBot")
-                .map_err(|_| println!("Failed to change name")),
-        );
+        self.next();
+    }
 
+    pub fn start_audio(&self, req: AudioRequest) {
         self.pipeline
             .set_state(gst::State::Null)
             .expect("Can set state to null");
+
+        let http_src = self
+            .pipeline
+            .get_by_name("http-source")
+            .expect("Http source should be registered");
+
+        http_src
+            .set_property("location", &&req.address)
+            .expect("Can set location on http_dl");
+
+        self.pipeline
+            .set_state(gst::State::Playing)
+            .expect("Can set state to playing");
+
+        self.set_description(&format!("Currently playing: {}", req.title));
     }
 
     pub fn volume(&self, volume: f64) {
-        self.pipeline
-            .get_by_name("volume")
-            .expect("Volume filter should be registered")
-            .set_property("volume", &volume)
-            .expect("can change volume");
+        if let Some(vol_filter) = self.pipeline.get_by_name("volume") {
+            vol_filter
+                .set_property("volume", &(10.0f64.powf(volume / 50.0 - 2.0)))
+                .expect("can change volume");
+
+            // TODO Reflect volume in name
+        }
     }
 
     pub fn play(&self) {
+        let http_src = self
+            .pipeline
+            .get_by_name("http-source")
+            .expect("Http source should be registered");
+
+        if http_src
+            .get_property("location")
+            .ok()
+            .and_then(|v| v.get::<String>().ok().and_then(|s| s.map(|s| s.is_empty())))
+            .unwrap_or(true)
+        {
+            if self
+                .playlist
+                .lock()
+                .expect("Mutex was not poisoned")
+                .is_empty()
+            {
+                self.send_message(MessageTarget::Channel, "There is nothing to play");
+                return;
+            }
+        }
+
         self.pipeline
             .set_state(gst::State::Playing)
             .expect("can play");
 
-        tokio::spawn(
-            self.conn
-                .lock()
-                .to_mut()
-                .set_name("PokeBot - Playing")
-                .map_err(|_| println!("Failed to change name")),
-        );
+        self.set_name("PokeBot - Playing");
+    }
+
+    pub fn next(&self) {
+        if let Some(req) = self.playlist.lock().expect("Mutex was not poisoned").pop() {
+            self.start_audio(req);
+        } else {
+            self.pipeline
+                .set_state(gst::State::Null)
+                .expect("Can set state to null");
+
+            let http_src = self
+                .pipeline
+                .get_by_name("http-source")
+                .expect("Http source should be registered");
+
+            http_src
+                .set_property("location", &"")
+                .expect("Can set location on http_dl");
+
+            self.set_name("PokeBot");
+        }
+    }
+
+    pub fn clear(&self) {
+        self.playlist
+            .lock()
+            .expect("Mutex was not poisoned")
+            .clear();
     }
 
     pub fn pause(&self) {
@@ -296,13 +374,7 @@ impl State {
             .set_state(gst::State::Paused)
             .expect("can pause");
 
-        tokio::spawn(
-            self.conn
-                .lock()
-                .to_mut()
-                .set_name("PokeBot - Paused")
-                .map_err(|_| println!("Failed to change name")),
-        );
+        self.set_name("PokeBot - Paused");
     }
 
     pub fn stop(&self) {
@@ -310,13 +382,7 @@ impl State {
             .set_state(gst::State::Ready)
             .expect("can stop");
 
-        tokio::spawn(
-            self.conn
-                .lock()
-                .to_mut()
-                .set_name("PokeBot - Stopped")
-                .map_err(|_| println!("Failed to change name")),
-        );
+        self.set_name("PokeBot - Stopped");
     }
 
     pub async fn disconnect(&self) {
@@ -325,4 +391,36 @@ impl State {
             .compat()
             .await;
     }
+
+    pub fn send_message(&self, target: MessageTarget, message: &str) {
+        tokio::spawn(
+            self.conn
+                .lock()
+                .to_mut()
+                .send_message(target, message)
+                .map_err(|e| println!("Failed to send message: {}", e)),
+        );
+    }
+
+    pub fn set_name(&self, name: &str) {
+        tokio::spawn(
+            self.conn
+                .lock()
+                .to_mut()
+                .set_name(name)
+                .map_err(|e| println!("Failed to change name: {}", e)),
+        );
+    }
+
+    pub fn set_description(&self, desc: &str) {
+        tokio::spawn(
+            self.conn
+                .lock()
+                .to_mut()
+                .get_client(&self.conn.lock().own_client)
+                .expect("Can get myself")
+                .set_description(desc)
+                .map_err(|e| println!("Failed to change description: {}", e)),
+        );
+    }
 }