Add discord voice basics

Joins a voice channel when a call is running, leaves when ended
Autoformat
This commit is contained in:
Sam W 2022-07-17 13:47:27 +01:00
parent 6042cc7a82
commit 26c481deba
4 changed files with 333 additions and 573 deletions

698
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -10,13 +10,13 @@ rand = "0.8.5"
rsip = "0.4.0"
rtp = "0.6.5"
sdp-rs = "0.2.1"
songbird = { version = "0.2.2", features = ["builtin-queue"]}
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 = "0.11.1"
twilight-http = { version = "0.11.1"}
twilight-model = "0.11.3"
webrtc-util = "0.5.4"

View File

@ -2,6 +2,7 @@ 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;
@ -21,18 +22,37 @@ use twilight_model::{
thread_local!(static RNG: RefCell<ThreadRng> = RefCell::new(thread_rng()));
const GREETINGS: &'static [&'static str] =
&["You rung?", "Hello there", "Right back atcha", "Yes?"];
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_MESSAGES | Intents::MESSAGE_CONTENT | Intents::GUILD_VOICE_STATES,
)
.presence(UpdatePresencePayload::new(
vec![MinimalActivity {
@ -47,21 +67,32 @@ pub async fn run_discord(
)?)
.build()
.await?;
let cluster = Arc::new(cluster);
let cluster_spawn = Arc::clone(&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 });
let client = Arc::new(Client::new(token.to_owned()));
loop {
select!(
Some((shard_id, ev)) = events.next() => {
let client = client.clone();
state.songbird.process(&ev).await;
let state_ = state.clone();
tokio::spawn(async move {
if let Err(e) = handle_event(ev, shard_id, client).await {
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();
@ -74,18 +105,41 @@ pub async fn run_discord(
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(client))]
async fn handle_event(ev: Event, shard_id: u64, client: Arc<Client>) -> StdErr<()> {
let me = client.current_user().exec().await?.model().await?;
#[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.id) {
if mentions(&msg.0, me) {
let greet = RNG.with(|rng| GREETINGS.choose(&mut *rng.borrow_mut()).unwrap());
client
state
.client
.create_message(msg.channel_id)
.content(greet)?
.exec()

View File

@ -1,18 +1,18 @@
use futures::{sink::Sink, SinkExt};
use tokio::sync::{broadcast, mpsc};
use tokio::signal;
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::net::SocketAddr;
use std::str;
use std::env;
use std::net::IpAddr;
use std::net::SocketAddr;
use std::str;
use std::time::Duration;
use tokio::net::UdpSocket;
use tokio::signal;
use tokio::sync::{broadcast, mpsc};
use tokio::time::sleep;
use tokio_stream::StreamExt;
use tokio_util::{
@ -20,17 +20,15 @@ use tokio_util::{
udp::UdpFramed,
};
use tracing::{event, instrument, Level};
use songbird::driver::Driver;
mod codecs;
use codecs::{SipCodec, RtpCodec};
use codecs::{RtpCodec, SipCodec};
mod discord;
const SIP_PORT: u16 = 5060;
const BIND_ADDR: &str = "0.0.0.0"; // for now
type StdErr<T> = Result<T, Box<dyn std::error::Error>>;
struct Server {}
@ -42,8 +40,12 @@ struct CurrentCall {
}
impl Server {
#[instrument(skip(shutdown, _done))]
async fn run_sip(mut shutdown: broadcast::Receiver<()>, _done: mpsc::Sender<()>) -> 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 {});
@ -89,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) => {
@ -158,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{
Attribute::Other(fmt, Some(params)) if fmt == "fmtp" => {
params.strip_prefix(&prefix)
},
_ => None
}).next();
let fmtp = md
.attributes
.iter()
.filter_map(|a| match a {
Attribute::Other(fmt, Some(params)) if fmt == "fmtp" => {
params.strip_prefix(&prefix)
}
_ => None,
})
.next();
Some((r.clone(), fmtp))
} else {
None
@ -178,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
@ -216,7 +224,7 @@ impl Server {
.media_descriptions
.iter()
.filter_map(|md| match Self::select_format(md) {
Ok((rtpmap, fmtp)) => Some((md, rtpmap, fmtp)),
Ok((rtpmap, fmtp)) => Some((md, rtpmap, fmtp)),
Err(e) => {
event!(Level::INFO, why=e, media_description=%md, "Rejected media description.");
None
@ -230,7 +238,7 @@ impl Server {
let socket = UdpSocket::bind(format!("{}:0", BIND_ADDR)).await?;
let rtp_port = socket.local_addr()?.port();
event!(Level::INFO, rtp_port, "Bound RTP port");
let mut rtp_framed = UdpFramed::new(socket, RtpCodec{});
let mut rtp_framed = UdpFramed::new(socket, RtpCodec {});
let mut res = base_res.clone();
res.status_code = 180.into();
responses.send(res).await;
@ -240,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,
@ -250,19 +261,26 @@ 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{
origin: sdp_rs::lines::Origin {
username: "-".into(),
sess_id: "foobar".into(),
sess_version: "foobar".into(),
@ -273,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! {
@ -308,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(())
@ -327,12 +356,13 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let (shutdown_tx, shutdown_rx_1) = broadcast::channel(1);
let (done_tx, mut done_rx) = mpsc::channel(1);
let done_2 = done_tx.clone();
tokio::spawn( async move {
discord::run_discord(&discord_token, shutdown_rx_1, done_tx.clone()).await;
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(sd_2, done_2).await;
Server::run_sip(call_tx, sd_2, done_2).await;
});
signal::ctrl_c().await?;