Discord: send a camera, choose a stream's quality, send stickers
Camera (#180): on the voice connection's own camera SSRC from op 2, announced by op 12 and the gateway's self_video, encrypted for the DAVE group and packetised like a stream. Off is op 12 with the stream inactive. Kept across a reconnect. setDiscordCamera / discordCameraReady, and sendDiscordVideoFrame takes kind: camera. Stream quality (#234): startDiscordScreenShare takes the size, rate and bitrate the window encodes at, and op 12 announces those instead of a fixed 720p30. Clamped to the account's tier from READY's premium_type - 720p30 without Nitro or with Basic, 1080p60 Classic, source at 60 with Nitro - and discordStreamLimits says which. Stickers: each guild's stickers are learned as it registers and on GUILD_STICKERS_UPDATE. listDiscordStickers marks the ones that cannot go in this conversation without Nitro; sendDiscordSticker posts sticker_ids. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
@@ -92,6 +92,7 @@ pub fn join_voice(
|
||||
// Recorded before the join, since the handshake it triggers can complete
|
||||
// before this function returns.
|
||||
state.voice.set_options(account_id, options);
|
||||
state.voice.note_flags(account_id, !options.transmit, !options.transmit);
|
||||
|
||||
sender.send(
|
||||
json!({
|
||||
@@ -261,6 +262,7 @@ pub async fn stop_ringing(state: &AppState, account_id: &str, channel_id: &str)
|
||||
/// microphone without saying so leaves everyone else looking at a live
|
||||
/// microphone icon wondering why you have gone quiet.
|
||||
pub fn announce_voice_flags(state: &AppState, account_id: &str, muted: bool, deafened: bool) -> bool {
|
||||
state.voice.note_flags(account_id, muted, deafened);
|
||||
let Some(sender) = state.runtime.discord_gateway_sender(account_id) else { return false };
|
||||
let Some((guild_id, channel_id)) = state.voice.current_channel(account_id) else { return false };
|
||||
sender
|
||||
@@ -273,7 +275,9 @@ pub fn announce_voice_flags(state: &AppState, account_id: &str, muted: bool, dea
|
||||
"channel_id": channel_id,
|
||||
"self_mute": muted,
|
||||
"self_deaf": deafened,
|
||||
"self_video": false
|
||||
// Repeated with every change: the gateway takes the
|
||||
// three together, and leaving it out turns a camera off.
|
||||
"self_video": state.voice.camera_on(account_id)
|
||||
}
|
||||
})
|
||||
.to_string(),
|
||||
|
||||
@@ -321,7 +321,11 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
// it - `premium_type` 1 is Classic, 2 is Nitro, 3 is
|
||||
// Basic, and all three buy the cross-guild use this
|
||||
// cares about; 0 or absent is none.
|
||||
let nitro = d["user"]["premium_type"].as_u64().unwrap_or(0) > 0;
|
||||
let premium = d["user"]["premium_type"].as_u64().unwrap_or(0).min(255) as u8;
|
||||
// And which tier, because a stream's ceiling differs
|
||||
// between them where emoji do not.
|
||||
state.runtime.set_discord_premium(&account_id, premium);
|
||||
let nitro = premium > 0;
|
||||
state.runtime.set_emoji_unrestricted(&account_id, nitro);
|
||||
// What a later reconnect needs to pick this session
|
||||
// back up rather than starting over.
|
||||
@@ -896,7 +900,7 @@ pub(super) async fn run_gateway(state: &AppState, config: &DiscordAccountConfig,
|
||||
// at connect, so the cheapest correct answer is to
|
||||
// register it again - which is what the role events
|
||||
// already do for the same reason.
|
||||
"GUILD_UPDATE" | "GUILD_EMOJIS_UPDATE" => {
|
||||
"GUILD_UPDATE" | "GUILD_EMOJIS_UPDATE" | "GUILD_STICKERS_UPDATE" => {
|
||||
let Some(guild_id) = d["guild_id"].as_str().or_else(|| d["id"].as_str()) else { continue };
|
||||
if guild_context.contains_key(guild_id) {
|
||||
resync_guild(state, config, guild_id, channel_map).await;
|
||||
|
||||
@@ -385,6 +385,22 @@ fn sound_wanted(account_id: &str) -> bool {
|
||||
sound().lock().unwrap().get(account_id).copied().unwrap_or(true)
|
||||
}
|
||||
|
||||
/// What the next stream from each account is sent at, as chosen when it was
|
||||
/// started.
|
||||
fn qualities() -> &'static std::sync::Mutex<std::collections::HashMap<String, super::voiceconn::VideoQuality>> {
|
||||
static QUALITY: std::sync::OnceLock<std::sync::Mutex<std::collections::HashMap<String, super::voiceconn::VideoQuality>>> =
|
||||
std::sync::OnceLock::new();
|
||||
QUALITY.get_or_init(Default::default)
|
||||
}
|
||||
|
||||
pub fn set_quality(account_id: &str, quality: super::voiceconn::VideoQuality) {
|
||||
qualities().lock().unwrap().insert(account_id.to_string(), quality);
|
||||
}
|
||||
|
||||
pub fn quality(account_id: &str) -> super::voiceconn::VideoQuality {
|
||||
qualities().lock().unwrap().get(account_id).copied().unwrap_or_default()
|
||||
}
|
||||
|
||||
/// Puts one encoded frame on the wire.
|
||||
///
|
||||
/// The encoding happens in the window - Chromium has the encoders and this
|
||||
@@ -765,3 +781,37 @@ mod tests {
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/// The best stream an account may send.
|
||||
#[derive(Clone, Copy, Debug, PartialEq)]
|
||||
pub struct StreamLimit {
|
||||
pub max_height: u32,
|
||||
pub max_framerate: u32,
|
||||
/// Whether the capture's own size may be sent unscaled.
|
||||
pub source: bool,
|
||||
}
|
||||
|
||||
/// What Discord's own client offers each tier. `premium_type` 0 is none and
|
||||
/// 3 is Nitro Basic, neither of which buys a better stream than 720p30; 1 is
|
||||
/// Nitro Classic, 1080p60; 2 is Nitro, the source at 60.
|
||||
pub fn stream_limit(premium_type: u8) -> StreamLimit {
|
||||
match premium_type {
|
||||
2 => StreamLimit { max_height: 2160, max_framerate: 60, source: true },
|
||||
1 => StreamLimit { max_height: 1080, max_framerate: 60, source: false },
|
||||
_ => StreamLimit { max_height: 720, max_framerate: 30, source: false },
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod limit_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn an_account_without_nitro_streams_at_720p30() {
|
||||
assert_eq!(stream_limit(0), StreamLimit { max_height: 720, max_framerate: 30, source: false });
|
||||
// Nitro Basic does not include a better stream.
|
||||
assert_eq!(stream_limit(3), stream_limit(0));
|
||||
assert_eq!(stream_limit(1).max_height, 1080);
|
||||
assert!(stream_limit(2).source);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -680,6 +680,17 @@ pub(super) async fn register_guild_channels(state: &AppState, config: &DiscordAc
|
||||
}
|
||||
}
|
||||
|
||||
// Its stickers. Where each may go is worked out when they are listed,
|
||||
// from the conversation's guild.
|
||||
super::stickers::note_guild(
|
||||
&account_id,
|
||||
guild_id,
|
||||
super::stickers::GuildStickers {
|
||||
guild_name: guild_name.clone(),
|
||||
stickers: super::stickers::read_stickers(guild),
|
||||
},
|
||||
);
|
||||
|
||||
spawn_backfill(state.clone(), config.token.clone(), config.user_id.clone(), config.display_name.clone(), new_channels);
|
||||
visible
|
||||
}
|
||||
|
||||
@@ -69,6 +69,7 @@ pub mod voiceconn;
|
||||
pub mod videorx;
|
||||
pub mod search;
|
||||
pub mod send;
|
||||
pub mod stickers;
|
||||
|
||||
pub use calls::*;
|
||||
pub use commands::*;
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
//! Discord stickers, sent.
|
||||
//!
|
||||
//! Reading them was already done (`messages::extract_stickers`); this is the
|
||||
//! other half. A sticker is a message of its own - `sticker_ids` on an
|
||||
//! otherwise empty message - so picking one sends it, the way it does in
|
||||
//! Discord's own client.
|
||||
//!
|
||||
//! The ones offered are each guild's own, learned as the guild is registered.
|
||||
//! Where they may be sent is Discord's rule, applied here so the picker can
|
||||
//! say so before the server refuses: a guild's stickers in that guild's
|
||||
//! channels, and anywhere at all with Nitro.
|
||||
|
||||
use super::*;
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
|
||||
/// One guild's stickers, and where they belong.
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct GuildStickers {
|
||||
pub guild_name: String,
|
||||
pub stickers: Vec<Sticker>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct Sticker {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
/// 1 PNG, 2 APNG, 3 Lottie, 4 GIF.
|
||||
pub format_type: u64,
|
||||
}
|
||||
|
||||
impl Sticker {
|
||||
/// A picture of it, or none for Lottie, which is a vector animation in a
|
||||
/// format nothing here can draw. Still offered - by name - since it can
|
||||
/// still be sent.
|
||||
pub fn url(&self) -> Option<String> {
|
||||
match self.format_type {
|
||||
3 => None,
|
||||
4 => Some(format!("https://media.discordapp.net/stickers/{}.gif", self.id)),
|
||||
_ => Some(format!("https://media.discordapp.net/stickers/{}.png?size=160", self.id)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Account to guild to its stickers.
|
||||
fn known() -> &'static Mutex<HashMap<String, HashMap<String, GuildStickers>>> {
|
||||
static KNOWN: OnceLock<Mutex<HashMap<String, HashMap<String, GuildStickers>>>> = OnceLock::new();
|
||||
KNOWN.get_or_init(Default::default)
|
||||
}
|
||||
|
||||
/// The stickers a guild payload carries - READY's, GUILD_CREATE's, or the
|
||||
/// REST guild a resync reads. Unavailable ones (a guild that lost the boost
|
||||
/// that paid for the slot) are left out, as Discord's client leaves them.
|
||||
pub fn read_stickers(guild: &Value) -> Vec<Sticker> {
|
||||
guild["stickers"]
|
||||
.as_array()
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.filter(|s| s["available"].as_bool().unwrap_or(true))
|
||||
.filter_map(|s| {
|
||||
Some(Sticker {
|
||||
id: s["id"].as_str()?.to_string(),
|
||||
name: s["name"].as_str()?.to_string(),
|
||||
format_type: s["format_type"].as_u64().unwrap_or(1),
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Records a guild's stickers, replacing what was known.
|
||||
pub fn note_guild(account_id: &str, guild_id: &str, entry: GuildStickers) {
|
||||
let mut all = known().lock().unwrap();
|
||||
let guilds = all.entry(account_id.to_string()).or_default();
|
||||
if entry.stickers.is_empty() {
|
||||
guilds.remove(guild_id);
|
||||
} else {
|
||||
guilds.insert(guild_id.to_string(), entry);
|
||||
}
|
||||
}
|
||||
|
||||
/// Every sticker the account has, and whether each can go where it is being
|
||||
/// sent from.
|
||||
pub fn list(state: &AppState, account_id: &str, buffer_id: &str) -> Vec<Value> {
|
||||
let anywhere = state.runtime.emoji_unrestricted(account_id);
|
||||
// Absent in a DM, which belongs to no guild - and where, without Nitro,
|
||||
// no guild's sticker can go.
|
||||
let here = state.runtime.get_discord_guild(buffer_id);
|
||||
let all = known().lock().unwrap();
|
||||
let Some(guilds) = all.get(account_id) else { return Vec::new() };
|
||||
let mut out = Vec::new();
|
||||
// This conversation's own guild first: what is most likely wanted, and
|
||||
// the only ones usable without Nitro.
|
||||
let mut ordered: Vec<(&String, &GuildStickers)> = guilds.iter().collect();
|
||||
ordered.sort_by_key(|(id, g)| (here.as_deref() != Some(id.as_str()), g.guild_name.to_lowercase()));
|
||||
for (guild_id, guild) in ordered {
|
||||
let usable = anywhere || here.as_deref() == Some(guild_id.as_str());
|
||||
for sticker in &guild.stickers {
|
||||
out.push(json!({
|
||||
"id": sticker.id,
|
||||
"name": sticker.name,
|
||||
"pack": guild.guild_name,
|
||||
"body": sticker.name,
|
||||
"url": sticker.url(),
|
||||
"locked": !usable,
|
||||
}));
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Sends a sticker as a message of its own.
|
||||
pub async fn send(state: &AppState, buffer_id: &str, token: &str, sticker_id: &str) -> Result<()> {
|
||||
let channel_id = state.runtime.get_discord_channel(buffer_id).ok_or_else(|| anyhow!("no known Discord channel for this buffer"))?;
|
||||
let resp = send_write(
|
||||
http_client_for(token)
|
||||
.post(format!("{API_BASE}/channels/{channel_id}/messages"))
|
||||
.header("Authorization", token)
|
||||
.json(&json!({ "sticker_ids": [sticker_id] })),
|
||||
)
|
||||
.await
|
||||
.context("sending a Discord sticker")?;
|
||||
if !resp.status().is_success() {
|
||||
let status = resp.status();
|
||||
let text = resp.text().await.unwrap_or_default();
|
||||
bail!("Discord API error {status}: {text}");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn a_guild_offers_its_available_stickers() {
|
||||
let guild = json!({ "stickers": [
|
||||
{ "id": "1", "name": "wave", "format_type": 1 },
|
||||
{ "id": "2", "name": "gone", "format_type": 1, "available": false },
|
||||
{ "id": "3", "name": "dance", "format_type": 4 },
|
||||
{ "id": "4", "name": "vector", "format_type": 3 },
|
||||
]});
|
||||
let stickers = read_stickers(&guild);
|
||||
assert_eq!(stickers.iter().map(|s| s.name.as_str()).collect::<Vec<_>>(), ["wave", "dance", "vector"]);
|
||||
assert!(stickers[1].url().unwrap().ends_with("3.gif"));
|
||||
// Offered by name: it can be sent, just not drawn here.
|
||||
assert_eq!(stickers[2].url(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_guild_with_no_stickers_has_none() {
|
||||
assert!(read_stickers(&json!({})).is_empty());
|
||||
}
|
||||
}
|
||||
@@ -548,6 +548,9 @@ pub async fn connect(
|
||||
// A viewer sends no media, so it claims no SSRCs. Announcing a video
|
||||
// SSRC it will never put a packet on would tell the server to expect a
|
||||
// second picture in a conversation that has one.
|
||||
// At whatever the share was started at - the window encodes to the same
|
||||
// numbers, so what the server tells viewers to expect is what arrives.
|
||||
let quality = super::golive::quality(account_id);
|
||||
let announce_video = json!({
|
||||
"op": 12,
|
||||
"d": {
|
||||
@@ -560,8 +563,8 @@ pub async fn connect(
|
||||
"streams": [{
|
||||
"type": "video", "rid": "100", "ssrc": video_ssrc, "active": true,
|
||||
"quality": 100, "rtx_ssrc": video_ssrc.wrapping_add(1),
|
||||
"max_bitrate": 2_500_000, "max_framerate": 30,
|
||||
"max_resolution": { "type": "fixed", "width": 1280, "height": 720 }
|
||||
"max_bitrate": quality.bitrate, "max_framerate": quality.framerate,
|
||||
"max_resolution": { "type": "fixed", "width": quality.width, "height": quality.height }
|
||||
}],
|
||||
}
|
||||
})
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
use crate::state::AppState;
|
||||
use anyhow::{Context, Result};
|
||||
use serde_json::json;
|
||||
use super::voiceconn::{self, Ended, Handshake, Media, VoiceConn};
|
||||
use super::voiceconn::{self, VideoQuality, Ended, Handshake, Media, VoiceConn};
|
||||
use std::sync::Arc;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Mutex;
|
||||
@@ -93,6 +93,12 @@ pub struct VoiceState {
|
||||
playbacks: Mutex<HashMap<String, Arc<crate::audio::Playback>>>,
|
||||
/// Who is talking right now, per account.
|
||||
speaking: Mutex<HashMap<String, Arc<SpeakingTracker>>>,
|
||||
/// Accounts whose camera is on, and at what. Kept past a connection, so
|
||||
/// one that drops and comes back brings the camera back with it.
|
||||
cameras: Mutex<HashMap<String, VideoQuality>>,
|
||||
/// The microphone and output flags last told to the gateway, which a
|
||||
/// camera turned on has to repeat: the gateway takes the three together.
|
||||
flags: Mutex<HashMap<String, (bool, bool)>>,
|
||||
}
|
||||
|
||||
/// Who is audible in one call, and how loudly.
|
||||
@@ -407,6 +413,20 @@ impl VoiceState {
|
||||
playbacks.get(account_id).map(|p| p.take_level())
|
||||
}
|
||||
|
||||
/// Whether this account's camera is on.
|
||||
pub fn camera_on(&self, account_id: &str) -> bool {
|
||||
self.cameras.lock().unwrap().contains_key(account_id)
|
||||
}
|
||||
|
||||
pub fn note_flags(&self, account_id: &str, muted: bool, deafened: bool) {
|
||||
self.flags.lock().unwrap().insert(account_id.to_string(), (muted, deafened));
|
||||
}
|
||||
|
||||
/// The microphone and output flags last announced, or the cautious pair.
|
||||
pub fn flags(&self, account_id: &str) -> (bool, bool) {
|
||||
self.flags.lock().unwrap().get(account_id).copied().unwrap_or((true, true))
|
||||
}
|
||||
|
||||
/// Who is talking in this account's call right now, loudest first.
|
||||
pub fn speakers(&self, account_id: &str, participants: &[String]) -> Vec<(String, f32)> {
|
||||
let speaking = self.speaking.lock().unwrap();
|
||||
@@ -533,10 +553,43 @@ async fn connect(state: &AppState, account_id: &str, info: &PendingHandshake) ->
|
||||
.await
|
||||
.context("opening the voice connection")?;
|
||||
|
||||
// A camera that was on before the connection dropped is on after it.
|
||||
if let Some(quality) = state.voice.cameras.lock().unwrap().get(account_id).copied() {
|
||||
conn.set_camera(true, quality);
|
||||
}
|
||||
state.voice.conns.lock().unwrap().insert(account_id.to_string(), conn);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Turns this account's camera on or off in the call it is in.
|
||||
///
|
||||
/// Two messages to two servers, both needed: the voice server is told a
|
||||
/// picture is coming on the camera SSRC, and the gateway is told
|
||||
/// `self_video`, which is what puts a tile up on everybody else's screen.
|
||||
pub fn set_camera(state: &AppState, account_id: &str, on: bool, quality: VideoQuality) -> Result<()> {
|
||||
let conn = state.voice.conns.lock().unwrap().get(account_id).cloned().context("not in a call")?;
|
||||
if on {
|
||||
state.voice.cameras.lock().unwrap().insert(account_id.to_string(), quality);
|
||||
} else {
|
||||
state.voice.cameras.lock().unwrap().remove(account_id);
|
||||
}
|
||||
conn.set_camera(on, quality);
|
||||
let (muted, deafened) = state.voice.flags(account_id);
|
||||
super::calls::announce_voice_flags(state, account_id, muted, deafened);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Whether a camera frame handed in now would reach anybody.
|
||||
pub fn camera_ready(state: &AppState, account_id: &str) -> bool {
|
||||
state.voice.conns.lock().unwrap().get(account_id).is_some_and(|c| c.camera_ready())
|
||||
}
|
||||
|
||||
/// One encoded camera frame from the window.
|
||||
pub async fn send_camera_frame(state: &AppState, account_id: &str, frame: &[u8], timestamp_micros: i64) -> Result<()> {
|
||||
let conn = state.voice.conns.lock().unwrap().get(account_id).cloned().context("not in a call")?;
|
||||
conn.send_camera_frame(frame, timestamp_micros).await
|
||||
}
|
||||
|
||||
/// `connection_ended`, boxed with its thread-safety stated.
|
||||
///
|
||||
/// Reconnecting opens a connection that carries this same handler, so the
|
||||
@@ -599,6 +652,7 @@ fn start_transmitting(device_id: Option<&str>, muted: bool) -> Result<(crate::au
|
||||
|
||||
pub async fn disconnect(state: &AppState, account_id: &str) {
|
||||
let conn = state.voice.conns.lock().unwrap().remove(account_id);
|
||||
state.voice.cameras.lock().unwrap().remove(account_id);
|
||||
state.voice.captures.lock().unwrap().remove(account_id);
|
||||
state.voice.playbacks.lock().unwrap().remove(account_id);
|
||||
state.voice.retries.lock().unwrap().remove(account_id);
|
||||
|
||||
@@ -31,7 +31,7 @@ use futures::{SinkExt, StreamExt};
|
||||
use serde_json::{json, Value};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::num::NonZeroU16;
|
||||
use std::sync::atomic::{AtomicU16, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::net::UdpSocket;
|
||||
@@ -95,12 +95,133 @@ pub enum Ended {
|
||||
/// A running connection. Dropping it does not end it; `stop` does.
|
||||
pub struct VoiceConn {
|
||||
stop: watch::Sender<bool>,
|
||||
camera: CameraSender,
|
||||
}
|
||||
|
||||
impl VoiceConn {
|
||||
pub fn stop(&self) {
|
||||
let _ = self.stop.send(true);
|
||||
}
|
||||
|
||||
/// Turns this end's camera on or off, as far as the voice server is
|
||||
/// concerned. The gateway's `self_video` is the other half, and the
|
||||
/// caller's: that is what tells everybody else's client to draw a tile.
|
||||
pub fn set_camera(&self, on: bool, quality: VideoQuality) {
|
||||
let sender = &self.camera;
|
||||
sender.on.store(on, Ordering::Relaxed);
|
||||
let _ = sender.ws.send(WsMessage::Text(announce_camera(sender.audio_ssrc, sender.video_ssrc, on, quality).to_string()));
|
||||
}
|
||||
|
||||
/// Whether a frame handed in now would reach anybody: the camera is on,
|
||||
/// and the group - where there is one - can be encrypted for.
|
||||
pub fn camera_ready(&self) -> bool {
|
||||
self.camera.on.load(Ordering::Relaxed) && group_ready(&self.camera.shared)
|
||||
}
|
||||
|
||||
/// One encoded camera frame, as however many packets it takes.
|
||||
pub async fn send_camera_frame(&self, frame: &[u8], timestamp_micros: i64) -> Result<()> {
|
||||
self.camera.send_frame(frame, timestamp_micros).await
|
||||
}
|
||||
}
|
||||
|
||||
/// What a picture is sent at - a camera or a stream - which the voice server
|
||||
/// passes on to everybody deciding what to ask for.
|
||||
#[derive(Clone, Copy, Debug)]
|
||||
pub struct VideoQuality {
|
||||
pub width: u32,
|
||||
pub height: u32,
|
||||
pub framerate: u32,
|
||||
pub bitrate: u32,
|
||||
}
|
||||
|
||||
impl Default for VideoQuality {
|
||||
/// Discord's own client sends a camera at 720p30 whatever the account,
|
||||
/// and 720p30 is a stream's ceiling on an account without Nitro.
|
||||
fn default() -> Self {
|
||||
VideoQuality { width: 1280, height: 720, framerate: 30, bitrate: 2_500_000 }
|
||||
}
|
||||
}
|
||||
|
||||
/// The `op 12` that says a camera is coming, or that it has gone. Turning it
|
||||
/// off is the same message with the stream inactive and no video SSRC, which
|
||||
/// is what Discord's own client sends - and what makes the far end drop the
|
||||
/// tile rather than hold the last frame.
|
||||
pub fn announce_camera(audio_ssrc: u32, video_ssrc: u32, on: bool, quality: VideoQuality) -> Value {
|
||||
json!({
|
||||
"op": 12,
|
||||
"d": {
|
||||
"audio_ssrc": audio_ssrc,
|
||||
"video_ssrc": if on { video_ssrc } else { 0 },
|
||||
"rtx_ssrc": if on { video_ssrc.wrapping_add(1) } else { 0 },
|
||||
"streams": [{
|
||||
"type": "video", "rid": "100", "ssrc": video_ssrc, "active": on,
|
||||
"quality": 100, "rtx_ssrc": video_ssrc.wrapping_add(1),
|
||||
"max_bitrate": quality.bitrate, "max_framerate": quality.framerate,
|
||||
"max_resolution": { "type": "fixed", "width": quality.width, "height": quality.height }
|
||||
}],
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// The sending half of this end's camera.
|
||||
///
|
||||
/// On the voice connection itself, not a connection of its own as a stream
|
||||
/// is: a camera is part of being in the call, and Discord carries it on the
|
||||
/// SSRC the voice server set aside for it in `op 2`.
|
||||
struct CameraSender {
|
||||
udp: Arc<UdpSocket>,
|
||||
sealer: Arc<Mutex<Sealer>>,
|
||||
shared: Arc<Shared>,
|
||||
ws: mpsc::UnboundedSender<WsMessage>,
|
||||
audio_ssrc: u32,
|
||||
video_ssrc: u32,
|
||||
sequence: AtomicU32,
|
||||
on: AtomicBool,
|
||||
}
|
||||
|
||||
impl CameraSender {
|
||||
async fn send_frame(&self, frame: &[u8], timestamp_micros: i64) -> Result<()> {
|
||||
if !self.on.load(Ordering::Relaxed) {
|
||||
return Ok(());
|
||||
}
|
||||
// As for a stream: encrypted for the group before it is cut into
|
||||
// packets, and dropped while the group is forming, because a frame
|
||||
// nobody can decrypt is worse than none - the next keyframe is two
|
||||
// seconds away.
|
||||
let frame = if self.shared.dave_version.load(Ordering::Relaxed) == 0 {
|
||||
frame.to_vec()
|
||||
} else {
|
||||
let mut group = self.shared.group.lock().unwrap();
|
||||
match group.session.as_mut() {
|
||||
Some(session) if session.is_ready() => session
|
||||
.encrypt(davey::MediaType::VIDEO, davey::Codec::VP8, frame)
|
||||
.map_err(|e| anyhow!("could not encrypt a camera frame: {e:?}"))?
|
||||
.into_owned(),
|
||||
_ => return Ok(()),
|
||||
}
|
||||
};
|
||||
let needed = rtp::packet_count_vp8(&frame);
|
||||
if needed == 0 {
|
||||
return Ok(());
|
||||
}
|
||||
// Reserved before anything is built, so two frames in flight cannot
|
||||
// interleave their sequence numbers.
|
||||
let first = self.sequence.fetch_add(needed as u32, Ordering::Relaxed) as u16;
|
||||
let timestamp = rtp::timestamp_from_micros(timestamp_micros);
|
||||
for packet in rtp::packetise_vp8(&frame, self.video_ssrc, first, timestamp) {
|
||||
let header = rtp::header(&packet);
|
||||
let sealed = self.sealer.lock().unwrap().seal(&header, &packet.payload)?;
|
||||
self.udp.send(&sealed).await.context("sending a camera packet")?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn group_ready(shared: &Shared) -> bool {
|
||||
if shared.dave_version.load(Ordering::Relaxed) == 0 {
|
||||
return true;
|
||||
}
|
||||
shared.group.lock().unwrap().session.as_ref().is_some_and(|s| s.is_ready())
|
||||
}
|
||||
|
||||
/// Which close codes mean "do not come back".
|
||||
@@ -162,6 +283,7 @@ pub async fn connect(
|
||||
// loses the names of everybody already in the call.
|
||||
let mut early: Vec<WsMessage> = Vec::new();
|
||||
let mut ssrc = 0u32;
|
||||
let mut video_ssrc = 0u32;
|
||||
let mut address = String::new();
|
||||
let mut port = 0u16;
|
||||
let mut modes: Vec<String> = Vec::new();
|
||||
@@ -185,6 +307,15 @@ pub async fn connect(
|
||||
address = d["ip"].as_str().unwrap_or_default().to_string();
|
||||
port = d["port"].as_u64().unwrap_or(0) as u16;
|
||||
modes = d["modes"].as_array().into_iter().flatten().filter_map(|m| m.as_str().map(str::to_string)).collect();
|
||||
// This end's camera SSRC, set aside in the streams
|
||||
// list rather than beside the audio one.
|
||||
video_ssrc = d["streams"]
|
||||
.as_array()
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.find(|s| s["type"].as_str() == Some("video"))
|
||||
.and_then(|s| s["ssrc"].as_u64())
|
||||
.unwrap_or(ssrc.wrapping_add(1) as u64) as u32;
|
||||
}
|
||||
_ => early.push(frame.clone()),
|
||||
}
|
||||
@@ -294,7 +425,17 @@ pub async fn connect(
|
||||
let sealer = Arc::new(Mutex::new(Sealer::new(mode, &key)?));
|
||||
|
||||
tokio::spawn(receive(shared.clone(), udp.clone(), sealer.clone(), ssrc, mode, key.clone(), stop_rx.clone()));
|
||||
tokio::spawn(clock(shared.clone(), udp.clone(), sealer, ssrc, media.playback, media.mic, ws_tx.clone(), stop_rx.clone()));
|
||||
tokio::spawn(clock(shared.clone(), udp.clone(), sealer.clone(), ssrc, media.playback, media.mic, ws_tx.clone(), stop_rx.clone()));
|
||||
let camera = CameraSender {
|
||||
udp: udp.clone(),
|
||||
sealer,
|
||||
shared: shared.clone(),
|
||||
ws: ws_tx.clone(),
|
||||
audio_ssrc: ssrc,
|
||||
video_ssrc,
|
||||
sequence: AtomicU32::new(rand::random::<u16>() as u32),
|
||||
on: AtomicBool::new(false),
|
||||
};
|
||||
tokio::spawn(async move {
|
||||
let ended = socket_loop(shared.clone(), write, read, early, beat_every, ws_rx, stop_rx).await;
|
||||
// Whatever cameras were showing are not any more.
|
||||
@@ -302,7 +443,7 @@ pub async fn connect(
|
||||
on_end(ended);
|
||||
});
|
||||
|
||||
Ok(Arc::new(VoiceConn { stop: stop_tx }))
|
||||
Ok(Arc::new(VoiceConn { stop: stop_tx, camera }))
|
||||
}
|
||||
|
||||
fn describe_close(reason: &Option<tokio_tungstenite::tungstenite::protocol::CloseFrame>) -> String {
|
||||
@@ -1192,6 +1333,21 @@ impl Default for Jitter {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
/// Turning a camera off is the same message with the stream inactive and
|
||||
/// no video SSRC - what makes the far end drop the tile rather than hold
|
||||
/// the last frame.
|
||||
#[test]
|
||||
fn a_camera_is_announced_on_and_off() {
|
||||
let on = announce_camera(10, 11, true, VideoQuality::default());
|
||||
assert_eq!(on["d"]["video_ssrc"], 11);
|
||||
assert_eq!(on["d"]["streams"][0]["active"], true);
|
||||
assert_eq!(on["d"]["streams"][0]["max_resolution"]["height"], 720);
|
||||
let off = announce_camera(10, 11, false, VideoQuality::default());
|
||||
assert_eq!(off["d"]["video_ssrc"], 0);
|
||||
assert_eq!(off["d"]["streams"][0]["active"], false);
|
||||
assert_eq!(off["d"]["audio_ssrc"], 10);
|
||||
}
|
||||
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
|
||||
+94
-1
@@ -4622,6 +4622,21 @@ pub async fn dispatch(
|
||||
// window says otherwise.
|
||||
let sound = params.get("audio").and_then(|v| v.as_bool()).unwrap_or(true);
|
||||
backend::discord::golive::set_sound(&account_id, sound);
|
||||
// The size, rate and bitrate the window will encode at, so the
|
||||
// server tells viewers the truth. Clamped to what this
|
||||
// account's tier allows, which is what Discord itself enforces
|
||||
// in its own client.
|
||||
let limit = backend::discord::golive::stream_limit(state.runtime.discord_premium(&account_id));
|
||||
let default = backend::discord::voiceconn::VideoQuality::default();
|
||||
backend::discord::golive::set_quality(
|
||||
&account_id,
|
||||
backend::discord::voiceconn::VideoQuality {
|
||||
width: p_i64(params, "width", default.width as i64).clamp(16, 3840) as u32,
|
||||
height: (p_i64(params, "height", default.height as i64) as u32).clamp(16, limit.max_height),
|
||||
framerate: (p_i64(params, "framerate", default.framerate as i64) as u32).clamp(1, limit.max_framerate),
|
||||
bitrate: p_i64(params, "bitrate", default.bitrate as i64).clamp(100_000, 10_000_000) as u32,
|
||||
},
|
||||
);
|
||||
backend::discord::golive::start(state, &account_id, guild_id.as_deref(), &channel_id)
|
||||
} else {
|
||||
backend::discord::golive::close(&account_id);
|
||||
@@ -4706,12 +4721,61 @@ pub async fn dispatch(
|
||||
let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(frame) else {
|
||||
return (None, Some("that frame is not base64".to_string()));
|
||||
};
|
||||
match backend::discord::golive::send_frame(account_id, &bytes, timestamp).await {
|
||||
// A camera goes on the voice connection, a screen on the stream's
|
||||
// own; the encoder in the window is the same for both.
|
||||
let sent = if p_str(params, "kind", "screen") == "camera" {
|
||||
backend::discord::voice::send_camera_frame(state, account_id, &bytes, timestamp).await
|
||||
} else {
|
||||
backend::discord::golive::send_frame(account_id, &bytes, timestamp).await
|
||||
};
|
||||
match sent {
|
||||
Ok(()) => (Some(ok_node()), None),
|
||||
Err(e) => (None, Some(format!("{e:#}"))),
|
||||
}
|
||||
}
|
||||
|
||||
// This end's camera in a Discord call, on or off. The picture itself
|
||||
// follows as `sendDiscordVideoFrame` with `kind: "camera"`.
|
||||
"setDiscordCamera" => {
|
||||
let Some(account_id) = p_str_opt(params, "accountId") else {
|
||||
return (None, Some("setDiscordCamera requires \"accountId\"".to_string()));
|
||||
};
|
||||
let on = p_bool(params, "on", false);
|
||||
let default = backend::discord::voiceconn::VideoQuality::default();
|
||||
let quality = backend::discord::voiceconn::VideoQuality {
|
||||
width: p_i64(params, "width", default.width as i64).clamp(16, 3840) as u32,
|
||||
height: p_i64(params, "height", default.height as i64).clamp(16, 2160) as u32,
|
||||
framerate: p_i64(params, "framerate", default.framerate as i64).clamp(1, 60) as u32,
|
||||
bitrate: p_i64(params, "bitrate", default.bitrate as i64).clamp(100_000, 10_000_000) as u32,
|
||||
};
|
||||
match backend::discord::voice::set_camera(state, account_id, on, quality) {
|
||||
Ok(()) => (Some(ok_node()), None),
|
||||
Err(e) => (None, Some(format!("{e:#}"))),
|
||||
}
|
||||
}
|
||||
|
||||
// Whether a camera frame sent now would reach anybody - the group
|
||||
// has formed - so the window opens the camera only once it would.
|
||||
"discordCameraReady" => {
|
||||
let Some(account_id) = p_str_opt(params, "accountId") else {
|
||||
return (None, Some("discordCameraReady requires \"accountId\"".to_string()));
|
||||
};
|
||||
(Some(serde_json::json!({ "ready": backend::discord::voice::camera_ready(state, account_id) })), None)
|
||||
}
|
||||
|
||||
// How good a stream this account may send. Discord ties it to the
|
||||
// account's tier, and its own client offers nothing above that.
|
||||
"discordStreamLimits" => {
|
||||
let Some(account_id) = p_str_opt(params, "accountId") else {
|
||||
return (None, Some("discordStreamLimits requires \"accountId\"".to_string()));
|
||||
};
|
||||
let limit = backend::discord::golive::stream_limit(state.runtime.discord_premium(account_id));
|
||||
(
|
||||
Some(serde_json::json!({ "maxHeight": limit.max_height, "maxFramerate": limit.max_framerate, "source": limit.source })),
|
||||
None,
|
||||
)
|
||||
}
|
||||
|
||||
// Whether the stream connection is up and wants frames. The window
|
||||
// asks before it starts encoding, so a capture is not opened against
|
||||
// a connection that never arrived.
|
||||
@@ -5110,6 +5174,35 @@ pub async fn dispatch(
|
||||
(Some(serde_json::json!(out)), None)
|
||||
}
|
||||
|
||||
// A Discord account's stickers: every guild's, each marked with
|
||||
// whether it can go in this conversation.
|
||||
"listDiscordStickers" => {
|
||||
let Some(buffer_id) = p_str_opt(params, "bufferId") else {
|
||||
return (None, Some("listDiscordStickers requires \"bufferId\"".to_string()));
|
||||
};
|
||||
let Some(buffer) = state.runtime.get_buffer(buffer_id) else {
|
||||
return (None, Some("no such buffer".to_string()));
|
||||
};
|
||||
(Some(serde_json::json!(backend::discord::stickers::list(state, &buffer.account_id, buffer_id))), None)
|
||||
}
|
||||
|
||||
"sendDiscordSticker" => {
|
||||
let (buffer_id, sticker_id) = match (p_str_opt(params, "bufferId"), p_str_opt(params, "stickerId")) {
|
||||
(Some(b), Some(s)) => (b, s),
|
||||
_ => return (None, Some("sendDiscordSticker requires \"bufferId\" and \"stickerId\"".to_string())),
|
||||
};
|
||||
let Some(buffer) = state.runtime.get_buffer(buffer_id) else {
|
||||
return (None, Some("no such buffer".to_string()));
|
||||
};
|
||||
let Some(cfg) = state.accounts.get_discord(&buffer.account_id) else {
|
||||
return (None, Some("account not connected".to_string()));
|
||||
};
|
||||
match backend::discord::stickers::send(state, buffer_id, &cfg.token, sticker_id).await {
|
||||
Ok(()) => (Some(ok_node()), None),
|
||||
Err(e) => (None, Some(format!("{e:#}"))),
|
||||
}
|
||||
}
|
||||
|
||||
"sendMatrixSticker" => {
|
||||
let (buffer_id, mxc) = match (p_str_opt(params, "bufferId"), p_str_opt(params, "mxc")) {
|
||||
(Some(b), Some(m)) => (b, m),
|
||||
|
||||
@@ -451,6 +451,8 @@ pub struct Runtime {
|
||||
/// Whether this account's subscription lets it use emoji away from where
|
||||
/// they live - Discord Nitro. Absent means no.
|
||||
emoji_unrestricted: Mutex<std::collections::HashSet<String>>,
|
||||
/// Each Discord account's `premium_type`, as READY gave it.
|
||||
discord_premium: Mutex<HashMap<String, u8>>,
|
||||
/// Which Kick livestreams each account is claiming to watch, by buffer -
|
||||
/// see backend/kick/watch.rs.
|
||||
kick_watching: Mutex<HashMap<String, crate::backend::kick::watch::Watching>>,
|
||||
@@ -997,6 +999,7 @@ impl Runtime {
|
||||
discord_buffer_emojis: Mutex::new(HashMap::new()),
|
||||
emoji_catalogue: Mutex::new(HashMap::new()),
|
||||
emoji_unrestricted: Mutex::new(std::collections::HashSet::new()),
|
||||
discord_premium: Mutex::new(HashMap::new()),
|
||||
kick_watching: Mutex::new(HashMap::new()),
|
||||
matrix_emoticons: Mutex::new(HashMap::new()),
|
||||
discord_friends: Mutex::new(HashMap::new()),
|
||||
@@ -1736,6 +1739,15 @@ impl Runtime {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn set_discord_premium(&self, account_id: &str, premium_type: u8) {
|
||||
self.discord_premium.lock().unwrap().insert(account_id.to_string(), premium_type);
|
||||
}
|
||||
|
||||
/// The account's Nitro tier, or none where READY has not said.
|
||||
pub fn discord_premium(&self, account_id: &str) -> u8 {
|
||||
self.discord_premium.lock().unwrap().get(account_id).copied().unwrap_or(0)
|
||||
}
|
||||
|
||||
pub fn emoji_unrestricted(&self, account_id: &str) -> bool {
|
||||
self.emoji_unrestricted.lock().unwrap().contains(account_id)
|
||||
}
|
||||
@@ -4706,6 +4718,7 @@ impl Runtime {
|
||||
matrix_verifications, matrix_ignored, matrix_verification_peers,
|
||||
matrix_push_rules, irc_away, kick_pins,
|
||||
matrix_server_facts, irc_isupport, sneedchat_rooms,
|
||||
discord_premium,
|
||||
);
|
||||
// Keyed by (account, something) or (buffer, something): the first
|
||||
// element carries the account either way.
|
||||
|
||||
Reference in New Issue
Block a user