Add discord voice basics

Joins a voice channel when a call is running, leaves when ended
Autoformat

Cargo.toml

@@ -6,12 +6,17 @@ edition = "2021"
 [dependencies]
 bytes = "1.1.0"
 futures = "0.3.21"
+rand = "0.8.5"
 rsip = "0.4.0"
 rtp = "0.6.5"
 sdp-rs = "0.2.1"
+songbird = { git = "https://github.com/serenity-rs/songbird", branch = "next", default_features = false, features = ["driver", "twilight-rustls", "zlib-stock"] }
 tokio = { version = "1.19.2", features = ["full"] }
 tokio-stream = "0.1.9"
 tokio-util = { version = "0.7.3", features = ["net", "codec"] }
 tracing = "0.1.35"
 tracing-subscriber = "0.3.14"
+twilight-gateway = "0.11.1"
+twilight-http = { version = "0.11.1"}
+twilight-model = "0.11.3"
 webrtc-util = "0.5.4"

flake.nix

@@ -31,6 +31,7 @@
       packages.default = naersk-lib.buildPackage {
         pname = "discosip";
         root = ./.;
+        buildInputs = with pkgs; [libopus pkgconfig];
       };
 
       apps.default = utils.lib.mkApp {drv = packages.default;};
@@ -43,11 +44,22 @@
         };
       in
         pkgs.devshell.mkShell {
-          packages = with pkgs; [
+          devshell.packages = with pkgs; [
             (rust.override {extensions = ["rls"];})
             ffmpeg
             (callPackage ./pjsip {inherit (darwin.apple_sdk.frameworks) AppKit;})
             opusTools
+            libopus
+            pkgconfig
+          ];
+          # Devshell doesn't do any of the automagic that pkgs.mkshell (thanks to
+          # mkderivation) does. So we need to manually tell pkg-config where to find
+          # libopus so that we can `cargo build` in our devshell.
+          env = [
+            {
+              name = "PKG_CONFIG_PATH";
+              value = "${pkgs.libopus.dev}/lib/pkgconfig";
+            }
           ];
         };
       formatter = pkgs.alejandra;

src/codecs.rs

@@ -0,0 +1,62 @@
+use bytes::{Bytes, BytesMut};
+use rsip::message::SipMessage;
+use rtp::packet::Packet as RtpPacket;
+use tokio_util::codec::{Decoder, Encoder};
+use tracing::{event, instrument, Level};
+use webrtc_util::marshal::Unmarshal;
+
+pub struct SipCodec {}
+
+impl Decoder for SipCodec {
+    type Item = SipMessage;
+    type Error = std::io::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if src.is_empty() {
+            Ok(None)
+        } else {
+            // We're assuming we get an entire sip message at once here.
+            match SipMessage::try_from(src.as_ref()) {
+                Ok(msg) => {
+                    src.clear();
+                    Ok(Some(msg))
+                }
+                Err(e) => {
+                    src.clear(); // Still clear the buf to be ready for the next packet
+                    event!(Level::ERROR, error = %e, "Error decoding SIP message.");
+                    Ok(None)
+                }
+            }
+        }
+    }
+}
+
+impl Encoder<SipMessage> for SipCodec {
+    type Error = std::io::Error;
+
+    #[instrument(level = "debug", skip(self, item, dst))]
+    fn encode(&mut self, item: SipMessage, dst: &mut BytesMut) -> Result<(), Self::Error> {
+        let stuff: Bytes = item.into();
+        dst.reserve(stuff.len());
+        dst.extend_from_slice(&stuff);
+        Ok(())
+    }
+}
+
+pub struct RtpCodec {}
+
+impl Decoder for RtpCodec {
+    type Item = RtpPacket;
+    type Error = std::io::Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if src.is_empty() {
+            Ok(None)
+        } else {
+            match RtpPacket::unmarshal(src) {
+                Ok(p) => Ok(Some(p)),
+                Err(e) => Err(Self::Error::new(std::io::ErrorKind::Other, e.to_string())),
+            }
+        }
+    }
+}

src/discord.rs

@@ -0,0 +1,153 @@
+use crate::StdErr;
+use futures::stream::StreamExt;
+use rand::seq::SliceRandom;
+use rand::{prelude::ThreadRng, thread_rng};
+use songbird::Songbird;
+use std::cell::RefCell;
+use std::sync::Arc;
+use tokio::select;
+use tokio::sync::{broadcast, mpsc};
+use tracing::{event, instrument, Level};
+use twilight_gateway::{Cluster, Event};
+use twilight_http::Client;
+use twilight_model::{
+    channel::message::Message,
+    gateway::{
+        payload::outgoing::update_presence::UpdatePresencePayload,
+        presence::{ActivityType, MinimalActivity, Status},
+        Intents,
+    },
+    id::{marker::UserMarker, Id},
+};
+
+thread_local!(static RNG: RefCell<ThreadRng> = RefCell::new(thread_rng()));
+
+const GREETINGS: &'static [&'static str] = &[
+    "You rung?",
+    "Hello there",
+    "Right back atcha",
+    "Yes?",
+    "Wow, rude",
+    "Go stick your head in a pig",
+];
+
+#[derive(Debug)]
+pub struct Call {
+    pub guild_id: u64,
+    pub channel_id: u64,
+    pub done: mpsc::Receiver<()>,
+}
+
+struct State {
+    songbird: Songbird,
+    client: Client,
+}
+
+pub async fn run_discord(
+    token: &str,
+    mut calls: mpsc::Receiver<Call>,
+    mut shutdown: broadcast::Receiver<()>,
+    _done: mpsc::Sender<()>,
+) -> StdErr<()> {
+    event!(Level::INFO, "Starting...");
+    let (cluster, mut events) = Cluster::builder(
+        token.to_owned(),
+        Intents::GUILD_MESSAGES | Intents::MESSAGE_CONTENT | Intents::GUILD_VOICE_STATES,
+    )
+    .presence(UpdatePresencePayload::new(
+        vec![MinimalActivity {
+            kind: ActivityType::Listening,
+            name: "Your INVITEs".to_owned(),
+            url: None,
+        }
+        .into()],
+        false,
+        None,
+        Status::Online,
+    )?)
+    .build()
+    .await?;
+    let cluster = Arc::new(cluster);
+    let client = Client::new(token.to_owned());
+    let me = client.current_user().exec().await?.model().await?.id;
+    let sb = Songbird::twilight(cluster.clone(), me);
+    let state = Arc::new(State {
+        songbird: sb,
+        client,
+    });
+    let cluster_spawn = cluster.clone();
+    tokio::spawn(async move { cluster_spawn.up().await });
+
+    loop {
+        select!(
+            Some((shard_id, ev)) = events.next() => {
+                state.songbird.process(&ev).await;
+                let state_ = state.clone();
+                tokio::spawn(async move {
+                    if let Err(e) = handle_event(ev, shard_id, state_, me).await {
+                        event!(Level::ERROR, err=?e, "Error handling discord event");
+                    }
+                });
+            },
+            Some(call) = calls.recv() => {
+                let state_ = state.clone();
+                tokio::spawn(handle_call(call, state_));
+            },
+            _ = shutdown.recv() => {
+                event!(Level::INFO, "Shutting down...");
+                cluster.down();
+                break;
+            },
+
+        );
+    }
+    event!(Level::INFO, "Done");
+    Ok(())
+}
+
+#[instrument(skip(call, state))]
+async fn handle_call(mut call: Call, state: Arc<State>) {
+    let (_handle, success) = state.songbird.join(call.guild_id, call.channel_id).await;
+    match success {
+        Ok(()) => event!(Level::INFO, %call.guild_id, %call.channel_id, "Joined channel"),
+        Err(err) => {
+            event!(Level::ERROR, %call.guild_id, %call.channel_id, %err, "Error joining channel")
+        }
+    }
+    let _ = call.done.recv().await;
+    match state.songbird.leave(call.guild_id).await {
+        Ok(()) => event!(Level::INFO, %call.guild_id, %call.channel_id, "Left channel"),
+        Err(err) => {
+            event!(Level::INFO, %call.guild_id, %call.channel_id, %err, "Error leaving channel")
+        }
+    }
+}
+
+fn mentions(msg: &Message, us: Id<UserMarker>) -> bool {
+    msg.mentions.iter().filter(|m| m.id == us).next().is_some()
+}
+
+#[instrument(skip(state))]
+async fn handle_event(
+    ev: Event,
+    shard_id: u64,
+    state: Arc<State>,
+    me: Id<UserMarker>,
+) -> StdErr<()> {
+    match ev {
+        Event::MessageCreate(msg) => {
+            if mentions(&msg.0, me) {
+                let greet = RNG.with(|rng| GREETINGS.choose(&mut *rng.borrow_mut()).unwrap());
+                state
+                    .client
+                    .create_message(msg.channel_id)
+                    .content(greet)?
+                    .exec()
+                    .await?;
+            }
+        }
+        Event::ShardConnected(_) => event!(Level::INFO, "Shard connected"),
+        _ => {}
+    }
+    Ok(())
+}

src/main.rs

@@ -1,90 +1,34 @@
-use bytes::{Bytes, BytesMut};
 use futures::{sink::Sink, SinkExt};
-use rtp::packet::Packet as RtpPacket;
-use webrtc_util::marshal::Unmarshal;
 use rsip::common::method::Method as SipMethod;
 use rsip::headers::header::Header as SipHeader;
 use rsip::message::{request::Request, response::Response, SipMessage};
 use sdp_rs::lines::media::{MediaType, ProtoType};
 use sdp_rs::lines::{attribute::Rtpmap, Attribute, Media};
 use sdp_rs::{MediaDescription, SessionDescription};
+use std::env;
+use std::net::IpAddr;
 use std::net::SocketAddr;
 use std::str;
-use std::net::IpAddr;
 use std::time::Duration;
 use tokio::net::UdpSocket;
-use tokio::sync::mpsc;
+use tokio::signal;
+use tokio::sync::{broadcast, mpsc};
 use tokio::time::sleep;
 use tokio_stream::StreamExt;
 use tokio_util::{
-    codec::{Decoder, Encoder},
     sync::{PollSendError, PollSender},
     udp::UdpFramed,
 };
 use tracing::{event, instrument, Level};
 
+mod codecs;
+use codecs::{RtpCodec, SipCodec};
+
+mod discord;
+
 const SIP_PORT: u16 = 5060;
 const BIND_ADDR: &str = "0.0.0.0"; // for now
 
-struct SipCodec {}
-
-impl Decoder for SipCodec {
-    type Item = SipMessage;
-    type Error = std::io::Error;
-
-    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        if src.is_empty() {
-            Ok(None)
-        } else {
-            // We're assuming we get an entire sip message at once here.
-            match SipMessage::try_from(src.as_ref()) {
-                Ok(msg) => {
-                    src.clear();
-                    Ok(Some(msg))
-                }
-                Err(e) => {
-                    src.clear(); // Still clear the buf to be ready for the next packet
-                    event!(Level::ERROR, error = %e, "Error decoding SIP message.");
-                    Ok(None)
-                }
-            }
-        }
-    }
-}
-
-impl Encoder<SipMessage> for SipCodec {
-    type Error = std::io::Error;
-
-    #[instrument(
-        level="debug",
-        skip(self, item, dst)
-    )]
-    fn encode(&mut self, item: SipMessage, dst: &mut BytesMut) -> Result<(), Self::Error> {
-        let stuff: Bytes = item.into();
-        dst.reserve(stuff.len());
-        dst.extend_from_slice(&stuff);
-        Ok(())
-    }
-}
-
-struct RtpCodec{}
-
-impl Decoder for RtpCodec {
-    type Item = RtpPacket;
-    type Error =  std::io::Error;
-
-    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        if src.is_empty() {
-            Ok(None)
-        } else {
-            match RtpPacket::unmarshal(src) {
-                Ok(p) =>Ok(Some(p)),
-                Err(e) => Err(Self::Error::new(std::io::ErrorKind::Other, e.to_string())),
-            }
-        }
-    }
-}
-
 type StdErr<T> = Result<T, Box<dyn std::error::Error>>;
 
 struct Server {}
@@ -96,8 +40,12 @@ struct CurrentCall {
 }
 
 impl Server {
-    #[instrument]
-    async fn run_sip() -> StdErr<()> {
+    #[instrument(skip(call_tx, shutdown, _done))]
+    async fn run_sip(
+        call_tx: mpsc::Sender<discord::Call>,
+        mut shutdown: broadcast::Receiver<()>,
+        _done: mpsc::Sender<()>,
+    ) -> StdErr<()> {
         event!(Level::INFO, "Starting...");
         let socket = UdpSocket::bind(format!("{}:{}", BIND_ADDR, SIP_PORT)).await?;
         let mut framed = UdpFramed::new(socket, SipCodec {});
@@ -109,6 +57,10 @@ impl Server {
         let (response_tx, mut response_rx) = mpsc::channel(10);
         loop {
             tokio::select! {
+                _ = shutdown.recv() => {
+                    event!(Level::INFO, "Shutting down...");
+                    break;
+                }
                 // We got a new response to send
                 Some((res, send_to)) = response_rx.recv() => {
                     event!(Level::INFO, remote=%send_to, "Sending response");
@@ -139,9 +91,10 @@ impl Server {
                                             // remote's address. This means that `handle_call`
                                             // doesn't need to be aware of the remote at all.
                                             let res_tx = response_tx.clone();
+                                            let call_tx = call_tx.clone();
                                             tokio::spawn(async move {
                                                 let res_tx = PollSender::new(res_tx).with::<_, _, _, PollSendError<_>>(|res| {futures::future::ready(Ok((res, remote)))});
-                                                Self::handle_call(&req, res_tx, req_rx).await.unwrap();
+                                                Self::handle_call(&req, res_tx, req_rx, call_tx).await.unwrap();
                                             });
                                         }
                                         Some(call) => {
@@ -188,6 +141,8 @@ impl Server {
                 }
             }
         }
+        event!(Level::INFO, "Done.");
+        Ok(())
     }
 
     // Check that the media description is something we can handle, returning the selected format
@@ -206,12 +161,16 @@ impl Server {
                     if r.encoding_name == "opus" {
                         let prefix = format!("{} ", r.payload_type);
                         // Find the matching fmtp if there is one
-                        let fmtp = md.attributes.iter().filter_map(|a| match a{
+                        let fmtp = md
+                            .attributes
+                            .iter()
+                            .filter_map(|a| match a {
                                 Attribute::Other(fmt, Some(params)) if fmt == "fmtp" => {
                                     params.strip_prefix(&prefix)
-                            },
-                            _ => None
-                        }).next();
+                                }
+                                _ => None,
+                            })
+                            .next();
                         Some((r.clone(), fmtp))
                     } else {
                         None
@@ -226,13 +185,14 @@ impl Server {
     // Handle a call
     #[instrument(
         level = "info",
-        skip(invite, responses, requests)
+        skip(invite, responses, requests, call_tx)
         fields()
     )]
     async fn handle_call<T: Sink<Response> + std::marker::Unpin>(
         invite: &Request,
         mut responses: T,
         mut requests: mpsc::Receiver<Request>,
+        call_tx: mpsc::Sender<discord::Call>,
     ) -> StdErr<()> {
         let mut base_res = Response::default();
         // Copy headers from the invite
@@ -288,8 +248,11 @@ impl Server {
         res.status_code = 200.into();
         // TODO: fix this lmao
         let ip: IpAddr = "10.23.2.134".parse()?;
-        res.headers.push(SipHeader::Contact(format!("sip:{}:{}", ip, SIP_PORT).into()));
-        res.headers.push(SipHeader::ContentType("application/sdp".into()));
+        res.headers.push(SipHeader::Contact(
+            format!("sip:{}:{}", ip, SIP_PORT).into(),
+        ));
+        res.headers
+            .push(SipHeader::ContentType("application/sdp".into()));
         let md = MediaDescription {
             media: Media {
                 media: MediaType::Audio,
@@ -298,16 +261,23 @@ impl Server {
                 proto: ProtoType::RtpAvp,
                 fmt: "101".into(),
             },
-            connections: vec!(sdp_rs::lines::Connection{
+            connections: vec![sdp_rs::lines::Connection {
                 nettype: "IN".into(),
                 addrtype: "IP4".into(),
                 connection_address: ip.into(),
-            }),
-            bandwidths: vec!(sdp_rs::lines::Bandwidth{bwtype:"TIAS".into(), bandwidth: 64000}),
-            attributes: vec!(sdp_rs::lines::Attribute::Other("rtpmap".into(), Some("101 opus/48000/2".into()))),
+            }],
+            bandwidths: vec![sdp_rs::lines::Bandwidth {
+                bwtype: "TIAS".into(),
+                bandwidth: 64000,
+            }],
+            attributes: vec![sdp_rs::lines::Attribute::Other(
+                "rtpmap".into(),
+                Some("101 opus/48000/2".into()),
+            )],
             info: None,
             key: None,
-        }.into();
+        }
+        .into();
         let sd: String = SessionDescription {
             version: sdp_rs::lines::Version::V0,
             origin: sdp_rs::lines::Origin {
@@ -321,26 +291,37 @@ impl Server {
             session_name: "discosip".to_owned().into(),
             session_info: None,
             uri: None,
-            emails: vec!(),
-            phones: vec!(),
+            emails: vec![],
+            phones: vec![],
             connection: None,
-            bandwidths: vec!(sdp_rs::lines::Bandwidth{bwtype:"AS".into(), bandwidth: 84}),
-            times: vec!(sdp_rs::Time{
-                active: sdp_rs::lines::Active{
-                    start: 0,
-                    stop: 0,
-                },
-                repeat: vec!(),
+            bandwidths: vec![sdp_rs::lines::Bandwidth {
+                bwtype: "AS".into(),
+                bandwidth: 84,
+            }],
+            times: vec![sdp_rs::Time {
+                active: sdp_rs::lines::Active { start: 0, stop: 0 },
+                repeat: vec![],
                 zone: None,
-            }).try_into()?,
+            }]
+            .try_into()?,
             key: None,
-            attributes: vec!(),
-            media_descriptions: vec!(md),
-        }.to_string();
+            attributes: vec![],
+            media_descriptions: vec![md],
+        }
+        .to_string();
         res.body = sd.as_bytes().into();
-        res.headers.push(SipHeader::ContentLength((res.body.len() as u32).into()));
+        res.headers
+            .push(SipHeader::ContentLength((res.body.len() as u32).into()));
         responses.send(res).await;
 
+        // Discord
+        let (disc_done, disc_done_rx) = mpsc::channel(1);
+        let disc_call = discord::Call {
+            guild_id: env::var("DISCORD_GUILD")?.parse()?,
+            channel_id: env::var("DISCORD_CHANNEL")?.parse()?,
+            done: disc_done_rx,
+        };
+        call_tx.send(disc_call).await?;
 
         loop {
             tokio::select! {
@@ -356,12 +337,12 @@ impl Server {
                     }
 
                 },
-                Some(rtp_frame) = rtp_framed.next() => {
-                    event!(Level::INFO, "Got RTP Packet!");
+                Some(_) = rtp_framed.next() => {
+                    //event!(Level::INFO, "Got RTP Packet!");
                 },
             }
         }
-
+        disc_done.send(()).await?;
         event!(Level::INFO, "Call handler loop done.");
 
         Ok(())
@@ -371,5 +352,24 @@ impl Server {
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     tracing_subscriber::fmt::init();
-    Server::run_sip().await
+    let discord_token = env::var("DISCORD_TOKEN")?;
+    let (shutdown_tx, shutdown_rx_1) = broadcast::channel(1);
+    let (done_tx, mut done_rx) = mpsc::channel(1);
+    let done_2 = done_tx.clone();
+    let (call_tx, call_rx) = mpsc::channel(1);
+    tokio::spawn(async move {
+        discord::run_discord(&discord_token, call_rx, shutdown_rx_1, done_tx.clone()).await;
+    });
+    let sd_2 = shutdown_tx.subscribe();
+    tokio::spawn(async move {
+        Server::run_sip(call_tx, sd_2, done_2).await;
+    });
+
+    signal::ctrl_c().await?;
+    event!(Level::INFO, "Got interrupt, shutting down");
+    shutdown_tx.send(())?;
+    let _ = done_rx.recv().await;
+    event!(Level::INFO, "Shut down. Goodbye!");
+
+    Ok(())
 }