wip

apps/master/src/commands/testnats.rs

@@ -46,7 +46,7 @@ pub async fn run(
         println!("job {:?}", job);
 
         let response = match nats_client
-            .request("corro-dj.download", to_stdvec(&job).unwrap().into())
+            .request("corro-dj.play", to_stdvec(&job).unwrap().into())
             .await
         {
             Ok(resp) => resp,

apps/master/src/main.rs

@@ -51,7 +51,7 @@ impl EventHandler for Handler {
     async fn ready(&self, ctx: Context, ready: Ready) {
         println!("{} is connected!", ready.user.name);
 
-        let commands = Command::set_global_commands(
+        if let Err(why) = Command::set_global_commands(
             &ctx.http,
             vec![
                 commands::ping::register(),
@@ -59,16 +59,17 @@ impl EventHandler for Handler {
                 commands::testnats::register(),
             ],
         )
-        .await;
+        .await
-
+        {
-        println!("I now have the following slash commands: {commands:#?}");
+            println!("Client error: {why:?}");
+        }
     }
 }
 
 #[tokio::main]
 async fn main() {
     // Configure the client with your Discord bot token in the environment.
-    let token = env::var("DISCORD_TOKEN").expect("Expected a token in the environment");
+    let discord_token = env::var("DISCORD_TOKEN").expect("Expected a token in the environment");
 
     let nats_client = async_nats::connect("nats://localhost:4222")
         .await
@@ -77,7 +78,7 @@ async fn main() {
     let handler = Handler { nats_client };
 
     // Build our client.
-    let mut discord_client = Client::builder(token, GatewayIntents::empty())
+    let mut discord_client = Client::builder(discord_token, GatewayIntents::empty())
         .event_handler(handler)
         .await
         .expect("Error creating discord client");

apps/worker/Cargo.toml

@@ -8,7 +8,9 @@ async-nats = { version = "0.46.0" }
 futures = { version = "0.3.32" }
 futures-executor = { version = "0.3.32" }
 postcard = { version = "1.1.3", features = ["use-std"] }
+serenity = { version = "0.12.5" }
+songbird = { version = "0.5.0", default-features = true }
 tokio = { version = "1.49.0", features = ["macros", "rt-multi-thread"] }
 types = { path = "../../libs/types" }
 which = { version = "8.0.0" }
-yt-dlp = { version = "2.1.0" }
+yt-dlp = { version = "2.4.0" }

apps/worker/src/main.rs

@@ -1,39 +1,97 @@
 mod workers;
 
-use std::env;
+use std::{env, num::NonZeroU64, path::PathBuf, str::FromStr};
 
 use futures::StreamExt;
 use postcard::{from_bytes, to_stdvec};
-use types::jobs::JobResponse;
+use serenity::{
+    Client,
+    all::{Context, EventHandler, GatewayIntents, Ready},
+    async_trait,
+};
+use songbird::SerenityInit;
+use types::{jobs::{JobResponse, Jobs, PlayJob}, misc::new_uuid_v4};
 use which::which;
 use yt_dlp::{Downloader, client::Libraries};
 
+struct Handler {
+    nats_client: async_nats::Client,
+    yt_downloader: Downloader,
+}
+
+#[async_trait]
+impl EventHandler for Handler {
+    async fn ready(&self, ctx: Context, ready: Ready) {
+        println!("{} is connected!", ready.user.name);
+
+        let voice_manager = songbird::get(&ctx).await;
+
+        let mut subscriber = self
+            .nats_client
+            .queue_subscribe("corro-dj.*", "group1".to_string())
+            .await
+            .unwrap();
+
+        // Receive and process messages
+        while let Some(message) = subscriber.next().await {
+            println!("Received message {:?}", message);
+
+            let subject = message.subject.split(".").collect::<Vec<&str>>()[1];
+
+            let result = match subject {
+                "download" => workers::download::download(
+                    &self.yt_downloader,
+                    from_bytes(&message.payload).unwrap(),
+                )
+                .await
+                .map(|res| JobResponse {
+                    content: res.content.map(Jobs::Download),
+                    error: res.error,
+                }),
+                "play" => {
+                    workers::play::play(&voice_manager, PlayJob {
+                        uuid: new_uuid_v4(),
+                        path: PathBuf::new(),
+                        guild_id: NonZeroU64::from_str("302837320250294283").unwrap(),
+                        channel_id: NonZeroU64::from_str("863014198703423499").unwrap()
+                    })
+                        .await
+                        .map(|res| JobResponse {
+                            content: res.content.map(Jobs::Play),
+                            error: res.error,
+                        })
+                }
+                _ => Err(format!("subject {subject} does not exists")),
+            };
+
+            let response = match result {
+                Ok(response) => response,
+                Err(err) => JobResponse {
+                    content: None,
+                    error: Some(err),
+                },
+            };
+
+            println!("response {:?}", response);
+
+            if let Some(reply) = message.reply {
+                self.nats_client
+                    .publish(reply, to_stdvec(&response).unwrap().into())
+                    .await
+                    .unwrap();
+            }
+        }
+    }
+}
+
 #[tokio::main]
 async fn main() {
-    let token = env::var("DISCORD_TOKEN").expect("Expected a token in the environment");
+    let discord_token = env::var("DISCORD_TOKEN").expect("Expected a token in the environment");
-
-    let mut client = Client::builder(&token, intents)
-        .event_handler(Handler)
-        .framework(framework)
-        .register_songbird()
-        // We insert our own HTTP client here to make use of in
-        // `~play`. If we wanted, we could supply cookies and auth
-        // details ahead of time.
-        //
-        // Generally, we don't want to make a new Client for every request!
-        .type_map_insert::<HttpKey>(HttpClient::new())
-        .await
-        .expect("Err creating client");
 
     let nats_client = async_nats::connect("nats://localhost:4222")
         .await
         .expect("Error creating nats client");
 
-    let mut subscriber = nats_client
-        .queue_subscribe("corro-dj.*", "group1".to_string())
-        .await
-        .unwrap();
-
     let yt_downloader = Downloader::builder(
         Libraries::new(which("yt-dlp").unwrap(), which("ffmpeg").unwrap()),
         "output",
@@ -42,33 +100,18 @@ async fn main() {
     .await
     .unwrap();
 
-    // Receive and process messages
+    let handler = Handler {
-    while let Some(message) = subscriber.next().await {
+        nats_client,
-        println!("Received message {:?}", message);
+        yt_downloader,
+    };
 
-        let subject = message.subject.split(".").collect::<Vec<&str>>()[1];
+    let mut discord_client = Client::builder(&discord_token, GatewayIntents::empty())
+        .event_handler(handler)
+        .register_songbird()
+        .await
+        .expect("Error creating discord client");
 
-        let result = match subject {
+    if let Err(why) = discord_client.start().await {
-            "download" => {
+        println!("Client error: {why:?}");
-                workers::download::download(&yt_downloader, from_bytes(&message.payload).unwrap())
-                    .await
-            }
-            _ => Err(format!("subject {subject} does not exists")),
-        };
-
-        let response = match result {
-            Ok(response) => response,
-            Err(err) => JobResponse {
-                content: None,
-                error: Some(err),
-            },
-        };
-
-        if let Some(reply) = message.reply {
-            nats_client
-                .publish(reply, to_stdvec(&response).unwrap().into())
-                .await
-                .unwrap();
-        }
     }
 }

apps/worker/src/workers/mod.rs

@@ -1 +1,2 @@
 pub mod download;
+pub mod play;

apps/worker/src/workers/play.rs

@@ -0,0 +1,25 @@
+use std::sync::Arc;
+
+use songbird::Songbird;
+use types::jobs::{JobResponse, PlayJob, PlayResponse};
+
+pub async fn play(
+    voice_manager: &Option<Arc<Songbird>>,
+    job: PlayJob,
+) -> Result<JobResponse<PlayResponse>, String> {
+    println!("job: {:?}", job);
+
+    if let Some(voice_manager) = voice_manager {
+        if let Err(why) = voice_manager.join(job.guild_id, job.channel_id).await {
+            println!("{why}:?");
+            return Err(format!("{why}"));
+        };
+
+        Ok(JobResponse {
+            content: Some(PlayResponse {}),
+            error: None,
+        })
+    } else {
+        Err("No voice_manager defined".to_string())
+    }
+}

libs/types/src/jobs.rs

@@ -1,8 +1,17 @@
-use std::path::PathBuf;
+use std::{num::NonZeroU64, path::PathBuf};
 
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
+#[derive(Debug, Deserialize, Serialize)]
+pub enum Jobs {
+    Download(DownloadResponse),
+    Play(PlayResponse),
+    // Error(String),
+}
+
+pub type JobResult = Result<Jobs, String>;
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct JobResponse<T> {
     pub content: Option<T>,
@@ -19,3 +28,14 @@ pub struct DownloadJob {
 pub struct DownloadResponse {
     pub path: PathBuf,
 }
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct PlayJob {
+    pub uuid: Uuid,
+    pub path: PathBuf,
+    pub guild_id: NonZeroU64,
+    pub channel_id: NonZeroU64,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct PlayResponse {}