diff options
| -rw-r--r-- | .gitignore | 4 | ||||
| -rw-r--r-- | src/audio_player.rs | 73 | ||||
| -rw-r--r-- | src/bot.rs | 549 | ||||
| -rw-r--r-- | src/command.rs | 2 | ||||
| -rw-r--r-- | src/main.rs | 429 | ||||
| -rw-r--r-- | src/teamspeak.rs | 76 | ||||
| -rw-r--r-- | src/youtube_dl.rs | 12 |
7 files changed, 736 insertions, 409 deletions
@@ -1,4 +1,4 @@ /target **/*.rs.bk -id.toml -log/
\ No newline at end of file +config.toml +log/ diff --git a/src/audio_player.rs b/src/audio_player.rs index 97a61cd..2ff7c11 100644 --- a/src/audio_player.rs +++ b/src/audio_player.rs @@ -6,7 +6,7 @@ use gstreamer as gst; use gstreamer_app::{AppSink, AppSinkCallbacks}; use gstreamer_audio::{StreamVolume, StreamVolumeFormat}; -use crate::{ApplicationMessage, State}; +use crate::bot::{MusicBotMessage, State}; use glib::BoolError; use log::{debug, error, info, warn}; use std::sync::{Arc, Mutex}; @@ -14,22 +14,10 @@ use tokio02::sync::mpsc::UnboundedSender; static GST_INIT: Once = Once::new(); -#[derive(Debug)] -pub enum AudioPlayerError { - GStreamerError(glib::error::BoolError), - StateChangeFailed, -} - -impl From<glib::error::BoolError> for AudioPlayerError { - fn from(err: BoolError) -> Self { - AudioPlayerError::GStreamerError(err) - } -} - -impl From<gst::StateChangeError> for AudioPlayerError { - fn from(_err: gst::StateChangeError) -> Self { - AudioPlayerError::StateChangeFailed - } +#[derive(PartialEq, Eq, Debug, Clone, Copy)] +pub enum PollResult { + Continue, + Quit, } pub struct AudioPlayer { @@ -38,7 +26,7 @@ pub struct AudioPlayer { http_src: gst::Element, volume: gst::Element, - sender: Arc<Mutex<UnboundedSender<ApplicationMessage>>>, + sender: Arc<Mutex<UnboundedSender<MusicBotMessage>>>, } fn make_element(factoryname: &str, display_name: &str) -> Result<gst::Element, AudioPlayerError> { @@ -87,7 +75,7 @@ fn add_decode_bin_new_pad_callback( impl AudioPlayer { pub fn new( - sender: Arc<Mutex<UnboundedSender<ApplicationMessage>>>, + sender: Arc<Mutex<UnboundedSender<MusicBotMessage>>>, callback: Option<Box<dyn FnMut(&[u8]) + Send>>, ) -> Result<Self, AudioPlayerError> { GST_INIT.call_once(|| gst::init().unwrap()); @@ -239,13 +227,27 @@ impl AudioPlayer { Ok(()) } + pub fn quit(&self, reason: String) { + info!("Quitting audio player"); + + if let Err(e) = self + .bus + .post(&gst::Message::new_application(gst::Structure::new_empty("quit")).build()) + { + warn!("Failed to send \"quit\" app event: {}", e); + } + + let sender = self.sender.lock().unwrap(); + sender.send(MusicBotMessage::Quit(reason)).unwrap(); + } + fn send_state(&self, state: State) { info!("Sending state {:?} to application", state); let sender = self.sender.lock().unwrap(); - sender.send(ApplicationMessage::StateChange(state)).unwrap(); + sender.send(MusicBotMessage::StateChange(state)).unwrap(); } - pub fn poll(&self) { + pub fn poll(&self) -> PollResult { debug!("Polling GStreamer"); 'outer: loop { while let Some(msg) = self.bus.timed_pop(gst::ClockTime(None)) { @@ -308,12 +310,39 @@ impl AudioPlayer { ); break 'outer; } + MessageView::Application(content) => { + if let Some(s) = content.get_structure() { + if s.get_name() == "quit" { + return PollResult::Quit; + } + } + } _ => { - // debug!("{:?}", msg) + //debug!("{:?}", msg) } }; } } debug!("Left GStreamer message loop"); + + PollResult::Continue + } +} + +#[derive(Debug)] +pub enum AudioPlayerError { + GStreamerError(glib::error::BoolError), + StateChangeFailed, +} + +impl From<glib::error::BoolError> for AudioPlayerError { + fn from(err: BoolError) -> Self { + AudioPlayerError::GStreamerError(err) + } +} + +impl From<gst::StateChangeError> for AudioPlayerError { + fn from(_err: gst::StateChangeError) -> Self { + AudioPlayerError::StateChangeFailed } } diff --git a/src/bot.rs b/src/bot.rs new file mode 100644 index 0000000..38f8628 --- /dev/null +++ b/src/bot.rs @@ -0,0 +1,549 @@ +use std::future::Future; +use std::io::BufRead; +use std::sync::{Arc, Mutex}; +use std::thread; + +use futures::future::{FutureExt, TryFutureExt}; +use futures01::future::Future as Future01; +use log::{debug, info}; +use serde::{Deserialize, Serialize}; +use structopt::StructOpt; +use tokio02::sync::mpsc::UnboundedSender; +use tsclientlib::{ClientId, ConnectOptions, Identity, Invoker, MessageTarget}; + +use crate::audio_player::*; +use crate::command::Command; +use crate::playlist::*; +use crate::teamspeak::*; +use crate::youtube_dl::AudioMetadata; + +use crate::Args; + +#[derive(Debug)] +pub struct Message { + pub target: MessageTarget, + pub invoker: Invoker, + pub text: String, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum State { + Playing, + Paused, + Stopped, + EndOfStream, +} + +#[derive(Debug)] +pub enum MusicBotMessage { + TextMessage(Message), + StateChange(State), + Quit(String), +} + +pub struct MusicBot { + name: String, + player: Arc<AudioPlayer>, + teamspeak: Option<Arc<TeamSpeakConnection>>, + playlist: Arc<Mutex<Playlist>>, + state: Arc<Mutex<State>>, +} + +impl MusicBot { + pub async fn new(args: MusicBotArgs) -> (Arc<Self>, impl Future) { + let (tx, mut rx) = tokio02::sync::mpsc::unbounded_channel(); + let tx = Arc::new(Mutex::new(tx)); + let (player, connection) = if args.local { + info!("Starting in CLI mode"); + let audio_player = AudioPlayer::new(tx.clone(), None).unwrap(); + + (audio_player, None) + } else { + info!("Starting in TeamSpeak mode"); + + let con_config = ConnectOptions::new(args.address) + .version(tsclientlib::Version::Linux_3_3_2) + .name(args.name.clone()) + .identity(args.id) + .log_commands(args.verbose >= 1) + .log_packets(args.verbose >= 2) + .log_udp_packets(args.verbose >= 3) + .channel(args.channel); + + let connection = Arc::new( + TeamSpeakConnection::new(tx.clone(), con_config) + .await + .unwrap(), + ); + let cconnection = connection.clone(); + let audio_player = AudioPlayer::new( + tx.clone(), + Some(Box::new(move |samples| { + cconnection.send_audio_packet(samples); + })), + ) + .unwrap(); + + (audio_player, Some(connection)) + }; + + player.set_volume(0.5).unwrap(); + let player = Arc::new(player); + let playlist = Arc::new(Mutex::new(Playlist::new())); + + spawn_gstreamer_thread(player.clone(), tx.clone()); + + if args.local { + spawn_stdin_reader(tx); + } + + let bot = Arc::new(Self { + name: args.name, + player, + teamspeak: connection, + playlist, + state: Arc::new(Mutex::new(State::Stopped)), + }); + + let cbot = bot.clone(); + let msg_loop = async move { + 'outer: loop { + while let Some(msg) = rx.recv().await { + if let MusicBotMessage::Quit(reason) = msg { + cbot.with_teamspeak(|ts| ts.disconnect(&reason)); + break 'outer; + } + cbot.on_message(msg).await.unwrap(); + } + } + debug!("Left message loop"); + }; + + (bot, msg_loop) + } + + #[inline(always)] + fn with_teamspeak<F: Fn(&TeamSpeakConnection)>(&self, func: F) { + if let Some(ts) = &self.teamspeak { + func(&ts); + } + } + + fn start_playing_audio(&self, metadata: AudioMetadata) { + if let Some(title) = metadata.title { + self.send_message(&format!("Playing '{}'", title)); + self.set_description(&format!("Currently playing '{}'", title)); + } else { + self.send_message("Playing unknown title"); + self.set_description("Currently playing"); + } + self.player.reset().unwrap(); + self.player.set_source_url(metadata.url).unwrap(); + self.player.play().unwrap(); + } + + pub async fn add_audio(&self, url: String) { + match crate::youtube_dl::get_audio_download_url(url).await { + Ok(metadata) => { + info!("Found audio url: {}", metadata.url); + + let mut playlist = self.playlist.lock().expect("Mutex was not poisoned"); + playlist.push(metadata.clone()); + + if !self.player.is_started() { + if let Some(request) = playlist.pop() { + self.start_playing_audio(request); + } + } else { + if let Some(title) = metadata.title { + self.send_message(&format!("Added '{}' to playlist", title)); + } else { + self.send_message("Added to playlist"); + } + } + } + Err(e) => { + info!("Failed to find audio url: {}", e); + + self.send_message(&format!("Failed to find url: {}", e)); + } + } + } + + fn send_message(&self, text: &str) { + debug!("Sending message to TeamSpeak: {}", text); + + self.with_teamspeak(|ts| ts.send_message_to_channel(text)); + } + + fn set_nickname(&self, name: &str) { + info!("Setting TeamsSpeak nickname to {}", name); + + self.with_teamspeak(|ts| ts.set_nickname(name)); + } + + fn set_description(&self, desc: &str) { + info!("Setting TeamsSpeak description to {}", desc); + + self.with_teamspeak(|ts| ts.set_description(desc)); + } + + async fn on_text(&self, message: Message) -> Result<(), AudioPlayerError> { + let msg = message.text; + if msg.starts_with("!") { + let tokens = msg[1..].split_whitespace().collect::<Vec<_>>(); + + match Command::from_iter_safe(&tokens) { + Ok(args) => self.on_command(args).await?, + Err(e) if e.kind == structopt::clap::ErrorKind::HelpDisplayed => { + self.send_message(&format!("\n{}", e.message)); + } + _ => (), + } + } + + Ok(()) + } + + async fn on_command(&self, command: Command) -> Result<(), AudioPlayerError> { + match command { + Command::Play => { + let playlist = self.playlist.lock().expect("Mutex was not poisoned"); + + if !self.player.is_started() { + if !playlist.is_empty() { + self.player.stop_current(); + } + } else { + self.player.play()?; + } + } + Command::Add { url } => { + // strip bbcode tags from url + let url = url.replace("[URL]", "").replace("[/URL]", ""); + + self.add_audio(url.to_string()).await; + } + Command::Pause => { + self.player.pause()?; + } + Command::Stop => { + self.player.reset()?; + } + Command::Next => { + let playlist = self.playlist.lock().expect("Mutex was not poisoned"); + if !playlist.is_empty() { + info!("Skipping to next track"); + self.player.stop_current(); + } else { + info!("Playlist empty, cannot skip"); + self.player.reset()?; + } + } + Command::Clear => { + self.playlist + .lock() + .expect("Mutex was not poisoned") + .clear(); + } + Command::Volume { percent: volume } => { + let volume = volume.max(0.0).min(100.0) * 0.01; + self.player.set_volume(volume)?; + } + Command::Leave => { + self.quit(String::from("Leaving")); + } + } + + Ok(()) + } + + fn on_state(&self, state: State) -> Result<(), AudioPlayerError> { + let mut current_state = self.state.lock().unwrap(); + if *current_state != state { + match state { + State::Playing => { + self.set_nickname(&format!("{} - Playing", self.name)); + } + State::Paused => { + self.set_nickname(&format!("{} - Paused", self.name)); + } + State::Stopped => { + self.set_nickname(&self.name); + self.set_description(""); + } + State::EndOfStream => { + let next_track = self.playlist.lock().expect("Mutex was not poisoned").pop(); + if let Some(request) = next_track { + info!("Advancing playlist"); + + self.start_playing_audio(request); + } else { + self.set_nickname(&self.name); + self.set_description(""); + } + } + } + } + + *current_state = state; + + Ok(()) + } + + async fn on_message(&self, message: MusicBotMessage) -> Result<(), AudioPlayerError> { + match message { + MusicBotMessage::TextMessage(message) => { + if MessageTarget::Channel == message.target { + self.on_text(message).await?; + } + } + MusicBotMessage::StateChange(state) => { + self.on_state(state)?; + } + MusicBotMessage::Quit(_) => (), + } + + Ok(()) + } + + pub fn quit(&self, reason: String) { + self.player.quit(reason); + } +} + +pub struct MasterBot { + config: MasterConfig, + teamspeak: Option<Arc<TeamSpeakConnection>>, + connected_bots: Arc<Mutex<Vec<Arc<MusicBot>>>>, +} + +impl MasterBot { + pub async fn new(args: MasterArgs) -> (Arc<Self>, impl Future) { + let (tx, mut rx) = tokio02::sync::mpsc::unbounded_channel(); + let tx = Arc::new(Mutex::new(tx)); + let connection = if args.local { + info!("Starting in CLI mode"); + + None + } else { + info!("Starting in TeamSpeak mode"); + + let mut con_config = ConnectOptions::new(args.address.clone()) + .version(tsclientlib::Version::Linux_3_3_2) + .name(args.name.clone()) + .identity(args.id) + .log_commands(args.verbose >= 1) + .log_packets(args.verbose >= 2) + .log_udp_packets(args.verbose >= 3); + + if let Some(channel) = args.channel { + con_config = con_config.channel(channel); + } + + let connection = Arc::new( + TeamSpeakConnection::new(tx.clone(), con_config) + .await + .unwrap(), + ); + + Some(connection) + }; + + let config = MasterConfig { + address: args.address, + bots: args.bots, + local: args.local, + verbose: args.verbose, + }; + + let bot = Arc::new(Self { + config, + teamspeak: connection, + connected_bots: Arc::new(Mutex::new(Vec::new())), + }); + + let cbot = bot.clone(); + let msg_loop = async move { + loop { + while let Some(msg) = rx.recv().await { + cbot.on_message(msg).await.unwrap(); + } + } + }; + + (bot, msg_loop) + } + + async fn spawn_bot(&self, id: ClientId) { + let channel_name = if let Some(ts) = &self.teamspeak { + ts.channel_path_of_user(id) + } else { + String::from("local") + }; + + info!("Connecting to {} on {}", channel_name, self.config.address); + let preset = self.config.bots[0].clone(); + let bot_args = MusicBotArgs { + name: preset.name, + owner: preset.owner, + local: self.config.local, + address: self.config.address.clone(), + id: preset.id, + channel: channel_name, + verbose: self.config.verbose, + }; + + let (app, fut) = MusicBot::new(bot_args).await; + tokio::spawn(fut.unit_error().boxed().compat().map(|_| ())); + let mut bots = self.connected_bots.lock().expect("Mutex was not poisoned"); + bots.push(app); + } + + async fn on_message(&self, message: MusicBotMessage) -> Result<(), AudioPlayerError> { + if let MusicBotMessage::TextMessage(message) = message { + if let MessageTarget::Poke(who) = message.target { + info!("Poked by {}, creating bot for their channel", who); + self.spawn_bot(who).await; + } + } + + Ok(()) + } +} + +fn spawn_stdin_reader(tx: Arc<Mutex<UnboundedSender<MusicBotMessage>>>) { + thread::spawn(move || { + let stdin = ::std::io::stdin(); + let lock = stdin.lock(); + for line in lock.lines() { + let line = line.unwrap(); + + let message = MusicBotMessage::TextMessage(Message { + target: MessageTarget::Server, + invoker: Invoker { + name: String::from("stdin"), + id: ClientId(0), + uid: None, + }, + text: line, + }); + + let tx = tx.lock().unwrap(); + tx.send(message).unwrap(); + } + }); +} + +fn spawn_gstreamer_thread( + player: Arc<AudioPlayer>, + tx: Arc<Mutex<UnboundedSender<MusicBotMessage>>>, +) { + thread::spawn(move || loop { + if player.poll() == PollResult::Quit { + break; + } + + tx.lock() + .unwrap() + .send(MusicBotMessage::StateChange(State::EndOfStream)) + .unwrap(); + }); +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct MasterArgs { + #[serde(default = "default_name")] + pub name: String, + #[serde(default = "default_local")] + pub local: bool, + pub address: String, + pub channel: Option<String>, + #[serde(default = "default_verbose")] + pub verbose: u8, + pub id: Identity, + pub bots: Vec<BotConfig>, +} + +fn default_name() -> String { + String::from("PokeBot") +} + +fn default_local() -> bool { + false +} + +fn default_verbose() -> u8 { + 0 +} + +impl MasterArgs { + pub fn merge(self, args: Args) -> Self { + let address = args.address.unwrap_or(self.address); + let local = args.local || self.local; + let channel = args.master_channel.or(self.channel); + let verbose = if args.verbose > 0 { + args.verbose + } else { + self.verbose + }; + + Self { + name: self.name, + bots: self.bots, + local, + address, + id: self.id, + channel, + verbose, + } + } +} + +#[derive(Debug)] +pub struct MusicBotArgs { + name: String, + owner: Option<ClientId>, + local: bool, + address: String, + id: Identity, + channel: String, + verbose: u8, +} + +pub struct MasterConfig { + pub address: String, + pub bots: Vec<BotConfig>, + pub local: bool, + pub verbose: u8, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct BotConfig { + pub name: String, + #[serde( + deserialize_with = "client_id_deserialize", + serialize_with = "client_id_serialize" + )] + pub owner: Option<ClientId>, + pub id: Identity, +} + +fn client_id_serialize<S>(c: &Option<ClientId>, s: S) -> Result<S::Ok, S::Error> +where + S: serde::Serializer, +{ + match c { + Some(c) => s.serialize_some(&c.0), + None => s.serialize_none(), + } +} + +fn client_id_deserialize<'de, D>(deserializer: D) -> Result<Option<ClientId>, D::Error> +where + D: serde::Deserializer<'de>, +{ + let id: Option<u16> = Deserialize::deserialize(deserializer)?; + + Ok(id.map(|id| ClientId(id))) +} diff --git a/src/command.rs b/src/command.rs index fbc714c..3a39290 100644 --- a/src/command.rs +++ b/src/command.rs @@ -27,4 +27,6 @@ pub enum Command { Clear, /// Changes the volume to the specified value Volume { percent: f64 }, + /// Leaves the channel + Leave, } diff --git a/src/main.rs b/src/main.rs index f4f7559..b8ccf78 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,52 +1,48 @@ -use std::io::{BufRead, Read}; +use std::fs::File; +use std::io::{Read, Write}; use std::path::PathBuf; -use std::sync::{Arc, Mutex}; -use std::thread; use futures::future::{FutureExt, TryFutureExt}; +use futures01::future::Future as Future01; use log::{debug, info}; use structopt::clap::AppSettings; use structopt::StructOpt; -use tokio02::sync::mpsc::UnboundedSender; -use tsclientlib::{ClientId, ConnectOptions, Identity, Invoker, MessageTarget}; +use tsclientlib::Identity; mod audio_player; +mod bot; mod command; mod playlist; mod teamspeak; mod youtube_dl; -use audio_player::*; -use playlist::*; -use teamspeak::*; -use youtube_dl::AudioMetadata; -use command::Command; +use bot::{BotConfig, MasterArgs, MasterBot}; #[derive(StructOpt, Debug)] #[structopt(raw(global_settings = "&[AppSettings::ColoredHelp]"))] -struct Args { +pub struct Args { #[structopt(short = "l", long = "local", help = "Run locally in text mode")] local: bool, #[structopt( + short = "g", + long = "generate-identities", + help = "Generate 'count' identities" + )] + gen_id_count: Option<u8>, + #[structopt( short = "a", long = "address", - default_value = "localhost", help = "The address of the server to connect to" )] - address: String, + address: Option<String>, + #[structopt(help = "Configuration file", parse(from_os_str), default_value = "config.toml")] + config_path: PathBuf, #[structopt( - short = "i", - long = "id", - help = "Identity file - good luck creating one", - parse(from_os_str) + short = "d", + long = "master_channel", + help = "The channel the master bot should connect to" )] - id_path: Option<PathBuf>, - #[structopt( - short = "c", - long = "channel", - help = "The channel the bot should connect to" - )] - default_channel: Option<String>, + master_channel: Option<String>, #[structopt( short = "v", long = "verbose", @@ -60,352 +56,77 @@ struct Args { // 3. Print udp packets } -#[derive(Debug)] -pub struct Message { - pub target: MessageTarget, - pub invoker: Invoker, - pub text: String, -} - -#[derive(Debug, PartialEq, Eq)] -pub enum State { - Playing, - Paused, - Stopped, - EndOfStream, -} - -#[derive(Debug)] -pub enum ApplicationMessage { - TextMessage(Message), - StateChange(State), -} - -struct Application { - player: Arc<AudioPlayer>, - teamspeak: Option<Arc<TeamSpeakConnection>>, - playlist: Arc<Mutex<Playlist>>, - state: Arc<Mutex<State>>, -} - -impl Application { - pub fn new( - player: Arc<AudioPlayer>, - playlist: Arc<Mutex<Playlist>>, - teamspeak: Option<Arc<TeamSpeakConnection>>, - ) -> Self { - Self { - player, - teamspeak, - playlist, - state: Arc::new(Mutex::new(State::Stopped)), - } - } - - #[inline(always)] - fn with_teamspeak<F: Fn(&TeamSpeakConnection)>(&self, func: F) { - if let Some(ts) = &self.teamspeak { - func(&ts); - } - } - - fn start_playing_audio(&self, metadata: AudioMetadata) { - if let Some(title) = metadata.title { - self.send_message(&format!("Playing '{}'", title)); - self.set_description(&format!("Currently playing '{}'", title)); - } else { - self.send_message("Playing unknown title"); - self.set_description("Currently playing"); - } - self.player.reset().unwrap(); - self.player.set_source_url(metadata.url).unwrap(); - self.player.play().unwrap(); - } - - pub async fn add_audio(&self, url: String) { - match youtube_dl::get_audio_download_url(url).await { - Ok(metadata) => { - info!("Found audio url: {}", metadata.url); - - let mut playlist = self.playlist.lock().expect("Mutex was not poisoned"); - playlist.push(metadata.clone()); - - if !self.player.is_started() { - if let Some(request) = playlist.pop() { - self.start_playing_audio(request); - } - } else { - if let Some(title) = metadata.title { - self.send_message(&format!("Added '{}' to playlist", title)); - } else { - self.send_message("Added to playlist"); - } - } - } - Err(e) => { - info!("Failed to find audio url: {}", e); - - self.send_message(&format!("Failed to find url: {}", e)); - } - } - } - - fn send_message(&self, text: &str) { - debug!("Sending message to TeamSpeak: {}", text); - - self.with_teamspeak(|ts| ts.send_message_to_channel(text)); - } - - fn set_nickname(&self, name: &str) { - info!("Setting TeamsSpeak nickname to {}", name); - - self.with_teamspeak(|ts| ts.set_nickname(name)); - } - - fn set_description(&self, desc: &str) { - info!("Setting TeamsSpeak description to {}", desc); - - self.with_teamspeak(|ts| ts.set_description(desc)); - } - - async fn on_text(&self, message: Message) -> Result<(), AudioPlayerError> { - let msg = message.text; - if msg.starts_with("!") { - let tokens = msg[1..].split_whitespace().collect::<Vec<_>>(); - - match Command::from_iter_safe(&tokens) { - Ok(args) => self.on_command(args).await?, - Err(e) if e.kind == structopt::clap::ErrorKind::HelpDisplayed => { - self.send_message(&format!("\n{}", e.message)); - } - _ => (), - } - } - - Ok(()) - } - - async fn on_command(&self, command: Command) -> Result<(), AudioPlayerError> { - match command { - Command::Play => { - let playlist = self.playlist.lock().expect("Mutex was not poisoned"); - - if !self.player.is_started() { - if !playlist.is_empty() { - self.player.stop_current()?; - } - } else { - self.player.play()?; - } - } - Command::Add { url } => { - // strip bbcode tags from url - let url = url.replace("[URL]", "").replace("[/URL]", ""); - - self.add_audio(url.to_string()).await; - } - Command::Pause => { - self.player.pause()?; - } - Command::Stop => { - self.player.reset()?; - } - Command::Next => { - let playlist = self.playlist.lock().expect("Mutex was not poisoned"); - if !playlist.is_empty() { - info!("Skipping to next track"); - self.player.stop_current()?; - } else { - info!("Playlist empty, cannot skip"); - self.player.reset()?; - } - } - Command::Clear => { - self.playlist - .lock() - .expect("Mutex was not poisoned") - .clear(); - } - Command::Volume { percent: volume } => { - let volume = volume.max(0.0).min(100.0) * 0.01; - self.player.set_volume(volume)?; - } - } - - Ok(()) - } - - fn on_state(&self, state: State) -> Result<(), AudioPlayerError> { - let mut current_state = self.state.lock().unwrap(); - if *current_state != state { - match state { - State::Playing => { - self.set_nickname("PokeBot - Playing"); - } - State::Paused => { - self.set_nickname("PokeBot - Paused"); - } - State::Stopped => { - self.set_nickname("PokeBot"); - self.set_description(""); - } - State::EndOfStream => { - let next_track = self.playlist.lock().expect("Mutex was not poisoned").pop(); - if let Some(request) = next_track { - info!("Advancing playlist"); - - self.start_playing_audio(request); - } else { - self.set_nickname("PokeBot"); - self.set_description(""); - } - } - } - } - - *current_state = state; - - Ok(()) - } - - pub async fn on_message(&self, message: ApplicationMessage) -> Result<(), AudioPlayerError> { - match message { - ApplicationMessage::TextMessage(message) => { - if let MessageTarget::Poke(who) = message.target { - info!("Poked by {}, joining their channel", who); - self.with_teamspeak(|ts| ts.join_channel_of_user(who)); - } else { - self.on_text(message).await?; - } - } - ApplicationMessage::StateChange(state) => { - self.on_state(state)?; - } - } - - Ok(()) +fn main() { + //let example = BotConfig { + //name: String::from("asd"), + //id: Identity::create().unwrap(), + //owner: Some(ClientId(12)), + //}; + //let bots = vec![example]; + //println!( + //"{}", + //toml::to_string(&MasterArgs { + //name: String::from("PokeBot"), + //id: Identity::create().unwrap(), + //address: String::from("localhost"), + //channel: Some(String::from("Poke If Needed")), + //local: false, + //verbose: 0, + //bots, + //}) + //.map_err(|e| panic!(e.to_string())) + //.unwrap() + //); + //panic!(); + if let Err(e) = run() { + println!("Error: {}", e); } } -fn main() { +fn run() -> Result<(), Box<dyn std::error::Error>> { log4rs::init_file("log4rs.yml", Default::default()).unwrap(); - tokio::run(async_main().unit_error().boxed().compat()); -} - -async fn async_main() { info!("Starting PokeBot!"); // Parse command line options let args = Args::from_args(); - debug!("Received CLI arguments: {:?}", std::env::args()); - - let (tx, mut rx) = tokio02::sync::mpsc::unbounded_channel(); - let tx = Arc::new(Mutex::new(tx)); - let (player, connection) = if args.local { - info!("Starting in CLI mode"); - let audio_player = AudioPlayer::new(tx.clone(), None).unwrap(); - - (audio_player, None) - } else { - info!("Starting in TeamSpeak mode"); - - let id = if let Some(path) = args.id_path { - let mut file = std::fs::File::open(path).expect("Failed to open id file"); - let mut content = String::new(); - file.read_to_string(&mut content) - .expect("Failed to read id file"); - - toml::from_str(&content).expect("Failed to parse id file") - } else { - Identity::create().expect("Failed to create id") - }; - - let mut con_config = ConnectOptions::new(args.address) - .version(tsclientlib::Version::Linux_3_3_2) - .name(String::from("PokeBot")) - .identity(id) - .log_commands(args.verbose >= 1) - .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 mut file = File::open(&args.config_path)?; + let mut toml = String::new(); + file.read_to_string(&mut toml)?; + + let mut config: MasterArgs = toml::from_str(&toml)?; + + if let Some(count) = args.gen_id_count { + for i in 0..count { + let id = Identity::create().expect("Failed to create id"); + let bot = BotConfig { + name: format!("{}", i), + owner: None, + id, + }; + config.bots.push(bot); } - let connection = Arc::new( - TeamSpeakConnection::new(tx.clone(), con_config) - .await - .unwrap(), - ); - let cconnection = connection.clone(); - let audio_player = AudioPlayer::new( - tx.clone(), - Some(Box::new(move |samples| { - cconnection.send_audio_packet(samples); - })), - ) - .unwrap(); - - (audio_player, Some(connection)) - }; - - player.set_volume(0.5).unwrap(); - let player = Arc::new(player); - let playlist = Arc::new(Mutex::new(Playlist::new())); - let application = Arc::new(Application::new( - player.clone(), - playlist.clone(), - connection, - )); - - spawn_gstreamer_thread(player, tx.clone()); - - if args.local { - spawn_stdin_reader(tx); + let toml = toml::to_string(&config)?; + let mut file = File::create(&args.config_path)?; + file.write_all(toml.as_bytes())?; + return Ok(()); } - loop { - while let Some(msg) = rx.recv().await { - application.on_message(msg).await.unwrap(); - } - } -} + let bot_args = config.merge(args); -fn spawn_stdin_reader(tx: Arc<Mutex<UnboundedSender<ApplicationMessage>>>) { - thread::spawn(move || { - let stdin = ::std::io::stdin(); - let lock = stdin.lock(); - for line in lock.lines() { - let line = line.unwrap(); - - let message = ApplicationMessage::TextMessage(Message { - target: MessageTarget::Server, - invoker: Invoker { - name: String::from("stdin"), - id: ClientId(0), - uid: None, - }, - text: line, - }); + debug!("Received CLI arguments: {:?}", std::env::args()); - let tx = tx.lock().unwrap(); - tx.send(message).unwrap(); + tokio::run( + async { + let (_, fut) = MasterBot::new(bot_args).await; + tokio::spawn(fut.unit_error().boxed().compat().map(|_| ())); } - }); -} - -fn spawn_gstreamer_thread( - player: Arc<AudioPlayer>, - tx: Arc<Mutex<UnboundedSender<ApplicationMessage>>>, -) { - thread::spawn(move || loop { - player.poll(); + .unit_error() + .boxed() + .compat(), + ); - tx.lock() - .unwrap() - .send(ApplicationMessage::StateChange(State::EndOfStream)) - .unwrap(); - }); + Ok(()) } diff --git a/src/teamspeak.rs b/src/teamspeak.rs index 79dc1bc..f1abaec 100644 --- a/src/teamspeak.rs +++ b/src/teamspeak.rs @@ -1,14 +1,20 @@ +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + use futures::compat::Future01CompatExt; use futures01::{future::Future, sink::Sink}; use tokio02::sync::mpsc::UnboundedSender; -use crate::{ApplicationMessage, Message}; -use std::sync::{Arc, Mutex}; use tsclientlib::Event::ConEvents; -use tsclientlib::{events::Event, ClientId, ConnectOptions, Connection, MessageTarget}; +use tsclientlib::{ + events::Event, ChannelId, ClientId, ConnectOptions, Connection, DisconnectOptions, + MessageTarget, Reason, +}; use log::error; +use crate::bot::{Message, MusicBotMessage}; + pub struct TeamSpeakConnection { conn: Connection, } @@ -30,7 +36,7 @@ fn get_message<'a>(event: &Event) -> Option<Message> { impl TeamSpeakConnection { pub async fn new( - tx: Arc<Mutex<UnboundedSender<ApplicationMessage>>>, + tx: Arc<Mutex<UnboundedSender<MusicBotMessage>>>, options: ConnectOptions, ) -> Result<TeamSpeakConnection, tsclientlib::Error> { let conn = Connection::new(options).compat().await?; @@ -44,7 +50,7 @@ impl TeamSpeakConnection { for event in *events { if let Some(msg) = get_message(event) { let tx = tx.lock().unwrap(); - tx.send(ApplicationMessage::TextMessage(msg)).unwrap(); + tx.send(MusicBotMessage::TextMessage(msg)).unwrap(); } } } @@ -72,23 +78,34 @@ impl TeamSpeakConnection { tokio::run(send_packet); } - pub fn join_channel_of_user(&self, id: ClientId) { - let channel = self - .conn - .lock() - .clients - .get(&id) - .expect("can find poke sender") - .channel; - tokio::spawn( - self.conn - .lock() - .to_mut() - .get_client(&self.conn.lock().own_client) - .expect("can get myself") - .set_channel(channel) - .map_err(|e| error!("Failed to switch channel: {}", e)), - ); + pub fn channel_path_of_user(&self, id: ClientId) -> String { + let conn = self.conn.lock(); + + let channel_id = conn.clients.get(&id).expect("can find poke sender").channel; + + let mut channel = conn + .channels + .get(&channel_id) + .expect("can find user channel"); + + let mut names = vec![&channel.name[..]]; + + // Channel 0 is the root channel + while channel.parent != ChannelId(0) { + names.push("/"); + channel = conn + .channels + .get(&channel.parent) + .expect("can find user channel"); + names.push(&channel.name); + } + + let mut path = String::new(); + while let Some(name) = names.pop() { + path.push_str(name); + } + + path } pub fn set_nickname(&self, name: &str) { @@ -122,4 +139,19 @@ impl TeamSpeakConnection { .map_err(|e| error!("Failed to send message: {}", e)), ); } + + pub fn disconnect(&self, reason: &str) { + let opt = DisconnectOptions::new() + .reason(Reason::Clientdisconnect) + .message(reason); + tokio::spawn( + self.conn + .disconnect(opt) + .map_err(|e| error!("Failed to send message: {}", e)), + ); + // Might or might not be required to keep tokio running while the bot disconnects + tokio::spawn( + tokio::timer::Delay::new(Instant::now() + Duration::from_secs(1)).map_err(|_| ()), + ); + } } diff --git a/src/youtube_dl.rs b/src/youtube_dl.rs index a917c54..c6012f0 100644 --- a/src/youtube_dl.rs +++ b/src/youtube_dl.rs @@ -1,8 +1,8 @@ +use futures::compat::Future01CompatExt; use std::process::{Command, Stdio}; use tokio_process::CommandExt; -use futures::compat::Future01CompatExt; -use serde::{Serialize, Deserialize}; +use serde::{Deserialize, Serialize}; use log::debug; @@ -13,13 +13,7 @@ pub struct AudioMetadata { } pub async fn get_audio_download_url(uri: String) -> Result<AudioMetadata, String> { - let ytdl_args = [ - "--no-playlist", - "-f", - "bestaudio/best", - "-j", - &uri, - ]; + let ytdl_args = ["--no-playlist", "-f", "bestaudio/best", "-j", &uri]; let mut cmd = Command::new("youtube-dl"); cmd.args(&ytdl_args); |