### is_ok_and example Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Example for the `is_ok_and` method, which returns `true` if the result is `Ok` and the value matches a predicate. The last part shows using `as_ref()` to avoid consuming the result. ```rust let x: Result = Ok(2); assert_eq!(x.is_ok_and(|x| x > 1), true); let x: Result = Ok(0); assert_eq!(x.is_ok_and(|x| x > 1), false); let x: Result = Err("hey"); assert_eq!(x.is_ok_and(|x| x > 1), false); let x: Result = Ok("ownership".to_string()); assert_eq!(x.as_ref().is_ok_and(|x| x.len() > 1), true); println!("still alive {:?}", x); ``` -------------------------------- ### is_ok example Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Example for the `is_ok` method, which returns `true` if the result is `Ok`. ```rust let x: Result = Ok(-3); assert_eq!(x.is_ok(), true); let x: Result = Err("Some error message"); assert_eq!(x.is_ok(), false); ``` -------------------------------- ### Start Opus Playout Source: https://docs.rs/cacophony/latest/src/cacophony/connection/handle.rs.html Initiates Opus playout by sending a Start command and returning an OpusPlayout handle. ```rust pub async fn start_opus_playout(&self) -> Result { self.ensure_open()?; let (response, receive) = oneshot::channel(); self.inner .media_tx .send(PlayoutCommand::Start { response: DriverReply::new(response), }) .await .map_err(|_| Error::Closed)?; let id = receive.await.map_err(|_| Error::Closed)??; Ok(OpusPlayout { id, media_tx: self.inner.media_tx.clone(), close: self.inner.close.clone(), finished: false, }) } ``` -------------------------------- ### Arc::new example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs a new Arc with the given value. This is the standard way to create an Arc. ```rust use std::sync::Arc; let five = Arc::new(5); ``` -------------------------------- ### Creating an Arc with Default Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Example for the `Default` implementation of `Arc`. It creates a new `Arc` with the `Default` value for `T`. ```rust use std::sync::Arc; let x: Arc = Default::default(); assert_eq!(*x, 0); ``` -------------------------------- ### FromIterator Result implementation example Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Demonstrates collecting a vector of Results into a Result containing a vector, showing successful collection. ```rust let v = vec![1, 2]; let res: Result, &'static str> = v.iter().map(|x: &u32| x.checked_add(1).ok_or("Overflow!") ).collect(); assert_eq!(res, Ok(vec![2, 3])); ``` -------------------------------- ### is_err example Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Example for the `is_err` method, which returns `true` if the result is `Err`. ```rust let x: Result = Ok(-3); assert_eq!(x.is_err(), false); let x: Result = Err("Some error message"); assert_eq!(x.is_err(), true); ``` -------------------------------- ### FromIterator Result underflow example Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Demonstrates collecting a vector of Results where an underflow error occurs. ```rust let v = vec![1, 2, 0]; let res: Result, &'static str> = v.iter().map(|x: &u32| x.checked_sub(1).ok_or("Underflow!") ).collect(); assert_eq!(res, Err("Underflow!")); ``` -------------------------------- ### Handle Connection Command Source: https://docs.rs/cacophony/latest/src/cacophony/connection/driver.rs.html Asynchronous method to process incoming connection commands, starting with the SetSpeaking variant. ```rust async fn handle_command( &mut self, command: ConnectionCommand, close_rx: &mut tokio::sync::watch::Receiver, ) -> Result<()> { match command { ConnectionCommand::SetSpeaking { flags, delay } => { let ssrc = self.state.internal().ready.ssrc; self.send_voice_gateway_command(GatewayCommand::Speaking(SpeakingCommand { speaking: flags.bits(), delay: Some(delay), ``` -------------------------------- ### Arc::try_new_uninit example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs an Arc with uninitialized contents, returning an error if allocation fails. Requires the nightly-only allocator_api feature. ```rust #![feature(allocator_api)] use std::sync::Arc; let mut five = Arc::::try_new_uninit()?; // Deferred initialization: Arc::get_mut(&mut five).unwrap().write(5); let five = unsafe { five.assume_init() }; assert_eq!(*five, 5); ``` -------------------------------- ### Arc::try_map example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Example for the nightly-only `Arc::try_map` associated function. It maps the value inside an `Arc` using a closure that returns a `Result`, reusing the allocation if successful. Requires the `smart_pointer_try_map` feature gate. ```rust #![feature(smart_pointer_try_map)] use std::sync::Arc; let b = Arc::new(7); let new = Arc::try_map(b, |&i| u32::try_from(i)).unwrap(); assert_eq!(*new, 7); ``` -------------------------------- ### Arc::from_raw Example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs an Arc from a raw pointer. The pointer must have been previously returned by Arc::into_raw. ```rust use std::sync::Arc; let x = Arc::new("hello".to_owned()); let x_ptr = Arc::into_raw(x); unsafe { // Convert back to an `Arc` to prevent leak. let x = Arc::from_raw(x_ptr); assert_eq!(&*x, "hello"); // Further calls to `Arc::from_raw(x_ptr)` would be memory-unsafe. } // The memory was freed when `x` went out of scope above, so `x_ptr` is now dangling! ``` -------------------------------- ### Result iteration and summation example Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Demonstrates using Result with iterators and the sum method, asserting an error result when a negative element is encountered. ```rust let res: Result = v.iter().map(f).sum(); assert_eq!(res, Err("Negative element found")); ``` -------------------------------- ### Arc::clone_from_ref Example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs a new Arc with a clone of the provided value. Requires the nightly-only clone_from_ref feature. ```rust #![feature(clone_from_ref)] use std::sync::Arc; let hello: Arc = Arc::clone_from_ref("hello"); ``` -------------------------------- ### Arc::increment_strong_count Example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Increments the strong reference count on the Arc associated with the provided pointer. ```rust use std::sync::Arc; let five = Arc::new(5); unsafe { let ptr = Arc::into_raw(five); Arc::increment_strong_count(ptr); // This assertion is deterministic because we haven't shared // the `Arc` between threads. let five = Arc::from_raw(ptr); assert_eq!(2, Arc::strong_count(&five)); } ``` -------------------------------- ### transpose on Result, E> Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Example for the `transpose` method on `Result, E>`, which converts a `Result` of an `Option` into an `Option` of a `Result`. ```rust #[derive(Debug, Eq, PartialEq)] struct SomeErr; let x: Result, SomeErr> = Ok(Some(5)); let y: Option> = Some(Ok(5)); assert_eq!(x.transpose(), y); ``` -------------------------------- ### Arc::new_zeroed example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs an Arc with memory filled with zero bytes. The unsafe assume_init is required to finalize the value. ```rust use std::sync::Arc; let zero = Arc::::new_zeroed(); let zero = unsafe { zero.assume_init() }; assert_eq!(*zero, 0) ``` -------------------------------- ### Arc::get_mut_unchecked lifetime mismatch UB Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Example of undefined behavior when using get_mut_unchecked with mismatched lifetimes. ```rust #![feature(get_mut_unchecked)] use std::sync::Arc; let x: Arc<&str> = Arc::new("Hello, world!"); { let s = String::from("Oh, no!"); let mut y: Arc<&str> = x.clone(); unsafe { // this is Undefined Behavior, because x's inner type // is &'long str, not &'short str *Arc::get_mut_unchecked(&mut y) = &s; } } println!("{}", &*x); // Use-after-free ``` -------------------------------- ### Arc::try_new example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs a new Arc, returning an error if allocation fails. Requires the nightly-only allocator_api feature. ```rust #![feature(allocator_api)] use std::sync::Arc; let five = Arc::try_new(5)?; ``` -------------------------------- ### Arc::new_uninit example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs an Arc with uninitialized contents, then initializes it later using Arc::get_mut and write. The unsafe assume_init is required to finalize the value. ```rust use std::sync::Arc; let mut five = Arc::::new_uninit(); // Deferred initialization: Arc::get_mut(&mut five).unwrap().write(5); let five = unsafe { five.assume_init() }; assert_eq!(*five, 5) ``` -------------------------------- ### Cloning an Arc Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Example for the `Clone` implementation of `Arc`. It creates a new pointer to the same allocation, increasing the strong reference count. ```rust use std::sync::Arc; let five = Arc::new(5); let _ = Arc::clone(&five); ``` -------------------------------- ### Arc::get_mut_unchecked type mismatch UB Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Example of undefined behavior when using get_mut_unchecked with mismatched types. ```rust #![feature(get_mut_unchecked)] use std::sync::Arc; let x: Arc = Arc::from("Hello, world!"); let mut y: Arc<[u8]> = x.clone().into(); unsafe { // this is Undefined Behavior, because x's inner type is str, not [u8] Arc::get_mut_unchecked(&mut y).fill(0xff); // 0xff is invalid in UTF-8 } println!("{}", &*x); // Invalid UTF-8 in a str ``` -------------------------------- ### Get raw pointer with as_ptr Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Provides a raw pointer to the data without consuming the Arc or affecting reference counts. ```rust use std::sync::Arc; let x = Arc::new("hello".to_owned()); let y = Arc::clone(&x); let x_ptr = Arc::as_ptr(&x); assert_eq!(x_ptr, Arc::as_ptr(&y)); assert_eq!(unsafe { &*x_ptr }, "hello"); ``` -------------------------------- ### Arc::try_clone_from_ref Example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs a new Arc with a clone of the provided value, returning an error if allocation fails. Requires nightly-only features. ```rust #![feature(clone_from_ref)] #![feature(allocator_api)] use std::sync::Arc; let hello: Arc = Arc::try_clone_from_ref("hello")?; ``` -------------------------------- ### Arc::try_new_zeroed example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs an Arc with memory filled with zero bytes, returning an error if allocation fails. Requires the nightly-only allocator_api feature. ```rust #![feature( allocator_api)] use std::sync::Arc; let zero = Arc::::try_new_zeroed()?; let zero = unsafe { zero.assume_init() }; assert_eq!(*zero, 0); ``` -------------------------------- ### Implementing AsFd-based trait on Arc Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Shows how the `AsFd` implementation for `Arc` allows implementing traits that require `AsFd` on `Arc`. The example defines a trait `MyTrait` and implements it for `Arc` and `Box`. ```rust use std::net::UdpSocket; use std::sync::Arc; trait MyTrait: AsFd {} impl MyTrait for Arc {} impl MyTrait for Box {} ``` -------------------------------- ### Arc::new_cyclic example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Constructs an Arc while providing a Weak to the allocation, allowing T to hold a weak reference to itself. The weak reference cannot be upgraded until the Arc is fully constructed. ```rust use std::sync::{Arc, Weak}; struct Gadget { me: Weak, } impl Gadget { /// Constructs a reference counted Gadget. fn new() -> Arc { // `me` is a `Weak` pointing at the new allocation of the // `Arc` we're constructing. Arc::new_cyclic(|me| { // Create the actual struct here. Gadget { me: me.clone() } }) } /// Returns a reference counted pointer to Self. fn me(&self) -> Arc { self.me.upgrade().unwrap() } } ``` -------------------------------- ### Arc::into_raw Example Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Consumes the Arc and returns the wrapped pointer. Must be converted back to an Arc using Arc::from_raw to avoid memory leaks. ```rust use std::sync::Arc; let x = Arc::new("hello".to_owned()); let x_ptr = Arc::into_raw(x); assert_eq!(unsafe { &*x_ptr }, "hello"); ``` -------------------------------- ### strong_count_from_raw Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Gets the number of strong (`Arc`) pointers to the allocation behind the given raw pointer. This is a nightly-only experimental API and is unsafe to use. ```APIDOC ## unsafe fn strong_count_from_raw(ptr: *const T) -> usize ### Description Gets the number of strong (`Arc`) pointers to the allocation behind the given raw pointer. This method does not consume or drop the `Arc` behind this pointer. ### Method unsafe fn ### Signature `pub unsafe fn strong_count_from_raw(ptr: *const T) -> usize` ### Parameters - **ptr** (*const T) - Required - Pointer to the value inside a live `Arc` allocation, such as a pointer returned by `Arc::into_raw`, `Arc::into_raw_with_allocator`, or `Arc::as_ptr`. ### Returns - **usize** - The number of strong (`Arc`) pointers to the allocation. ### Safety - The pointer must point to (and have valid metadata for) the value inside a live `Arc` allocation. - `T` must have the same alignment as that value. - The associated `Arc` instance must be valid (i.e., the strong count must be at least 1) for the duration of this method. - Another thread can change the strong count at any time, including between calling this method and acting on the result. ### Examples ```rust #![feature(arc_raw_get_strong)] use std::sync::Arc; let five = Arc::new(5); let _also_five = Arc::clone(&five); let ptr = Arc::into_raw(five); unsafe { // This assertion is deterministic because we haven't shared // the `Arc` between threads. assert_eq!(2, Arc::strong_count_from_raw(ptr)); // Convert back to an `Arc` to avoid leaking memory. let five = Arc::from_raw(ptr); assert_eq!(2, Arc::strong_count(&five)); } ``` ``` -------------------------------- ### OpusPlayout: push_packet, try_push_packet, finish (Rust) Source: https://docs.rs/cacophony/latest/src/cacophony/connection/playout.rs.html Defines OpusPlayout, OpusPlayoutStats, and DurationDistribution for the connection playout module. Async push_packet/push_packet_batch and non-blocking try_push_packet/try_push_packet_batch send PlayoutCommand::Frame/Frames over an mpsc channel; try_ variants return Error::Backpressure(BackpressureError::MediaQueueFull) when full and Error::Closed when the channel is closed. finish() sends PlayoutCommand::Finish and awaits a oneshot reply to return OpusPlayoutStats; the displayed listing is cut off at the start of the Drop impl. ```rust use std::{ collections::VecDeque, sync::Arc, time::{Duration, Instant as StdInstant}, }; use tokio::{ sync::{mpsc, oneshot, watch}, time::Instant, }; use crate::{ errors::{BackpressureError, Error, Result}, gateway::{GatewayCommand, SpeakingCommand, SpeakingFlags}, media::OutboundPacket, opus::discord::{Packet, PacketBatch, PacketSpan}, queue::DriverReply, stats::DurationStats, }; use super::{ConnectionClose, driver::ConnectionDriver}; #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct DurationDistribution { pub count: usize, pub min: Option, pub avg: Option, pub p95: Option, pub max: Option, } #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct OpusPlayoutStats { pub packets: usize, pub first_packet: Option, pub last_packet: Option, pub first_packet_sent_at: Option, pub first_packet_sent_after_start: Option, pub wall_elapsed: Duration, pub expected_media: Duration, pub send_call: DurationDistribution, pub inter_frame_gap: DurationDistribution, pub scheduler_lateness: DurationDistribution, pub late_wakes_over_5ms: usize, pub late_wakes_over_20ms: usize, pub underflows: usize, pub underflow_wait: DurationDistribution, pub dropped_stale_frames: usize, pub clock_rebases: usize, pub clock_rebase_wait: DurationDistribution, pub dave_enabled: bool, } pub struct OpusPlayout { pub(super) id: u64, pub(super) media_tx: mpsc::Sender, pub(super) close: ConnectionClose, pub(super) finished: bool, } impl OpusPlayout { fn ensure_open(&self) -> Result<()> { if self.close.is_closed() { Err(Error::Closed) } else { Ok(()) } } pub async fn push_packet(&self, packet: Packet) -> Result<()> { self.ensure_open()?; self.media_tx .send(PlayoutCommand::Frame { id: self.id, frame: PlayoutFrame::packet(packet), }) .await .map_err(|_| Error::Closed) } pub async fn push_packet_batch(&self, packets: PacketBatch) -> Result<()> { self.ensure_open()?; let frames = PlayoutFrame::packet_batch(packets); if frames.is_empty() { return Ok(()); } self.media_tx .send(PlayoutCommand::Frames { id: self.id, frames, }) .await .map_err(|_| Error::Closed) } pub fn try_push_packet(&self, packet: Packet) -> Result<()> { self.ensure_open()?; match self.media_tx.try_send(PlayoutCommand::Frame { id: self.id, frame: PlayoutFrame::packet(packet), }) { Ok(()) => Ok(()), Err(mpsc::error::TrySendError::Closed(_)) => Err(Error::Closed), Err(mpsc::error::TrySendError::Full(_)) => { Err(Error::Backpressure(BackpressureError::MediaQueueFull)) } } } pub fn try_push_packet_batch(&self, packets: PacketBatch) -> Result<()> { self.ensure_open()?; let frames = PlayoutFrame::packet_batch(packets); if frames.is_empty() { return Ok(()); } match self.media_tx.try_send(PlayoutCommand::Frames { id: self.id, frames, }) { Ok(()) => Ok(()), Err(mpsc::error::TrySendError::Closed(_)) => Err(Error::Closed), Err(mpsc::error::TrySendError::Full(_)) => { Err(Error::Backpressure(BackpressureError::MediaQueueFull)) } } } pub async fn finish(mut self) -> Result { let stats = self.finish_inner().await?; self.finished = true; Ok(stats) } async fn finish_inner(&self) -> Result { self.ensure_open()?; let (response, receive) = oneshot::channel(); self.media_tx .send(PlayoutCommand::Finish { id: self.id, response: DriverReply::new(response), }) .await .map_err(|_| Error::Closed)?; receive.await.map_err(|_| Error::Closed)? } } impl Drop for OpusPlayout { fn drop(&mut self) { if !self.finished { let _ = self ``` -------------------------------- ### Implementing AsRawFd-based trait on Arc Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Shows how the `AsRawFd` implementation for `Arc` allows implementing traits that require `AsRawFd` on `Arc`. The example defines a trait `MyTrait` and implements it for `Arc` and `Box`. ```rust use std::net::UdpSocket; use std::sync::Arc; trait MyTrait: AsRawFd { } impl MyTrait for Arc {} impl MyTrait for Box {} ``` -------------------------------- ### Cacophony media.rs imports and dependencies Source: https://docs.rs/cacophony/latest/src/cacophony/media.rs.html No description ```Rust use std::{ collections::VecDeque, fmt, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, time::Duration, }; use aes_gcm::{ Aes256Gcm, aead::{AeadInOut, KeyInit}, }; use chacha20poly1305::XChaCha20Poly1305; use dave::{Codec, MediaType}; use futures_util::{StreamExt, stream::FuturesUnordered}; use tokio::{ net::{TcpStream, UdpSocket}, time::{sleep, timeout}, }; use tokio_tungstenite::{client_async_tls_with_config, tungstenite::client::IntoClientRequest}; use crate::{ AEAD_TAG_LEN, GatewayWebSocketConnectResult, RTP_VERSION, RTPSIZE_NONCE_LEN, WEBSOCKET_ADDRESS_CONNECT_TIMEOUT, WEBSOCKET_ADDRESS_STAGGER, codecs::{self, DiscordCodecDescriptor}, errors::{ Error, InvalidInputError, PayloadKind, ProtocolError, Result, RtpError, TransportCryptoError, }, observer::{DavePendingMediaReason, RtcpHeader}, rtp::RtpPayloadType, rtp_payload::ParsedRtpPayload, secrets::RedactedSecret, state::{EncryptionMode, PendingMediaFrame, PendingMediaPacket}, }; ``` -------------------------------- ### Initialize Connection Driver Source: https://docs.rs/cacophony/latest/src/cacophony/connection/driver.rs.html Initializes the connection state, sets up communication channels, and spawns the connection driver task. ```rust 763 .text_payload()? 764 .into(), 765 )) 766 .await?; 767 768 let (session_description_event, pending_events) = observed_stage_timeout( 769 &observer, 770 ConnectionEvent { 771 endpoint: &endpoint, 772 guild_id, 773 user_id, 774 }, 775 ConnectStage::SessionDescription, 776 SESSION_DESCRIPTION_TIMEOUT, 777 wait_for_session_description(&mut read, &mut last_seq), 778 ) 779 .await?; 780 let session_description: SessionDescription = parse_data(session_description_event.data)?; 781 let rtp_codecs = 782 DiscordRtpCodecMap::new(&session_description, &config.options.codec_preferences)?; 783 let dave_protocol_version = session_description.dave_protocol_version; 784 let dave_identity = config.options.dave_identity.clone(); 785 let runtime_config = config.runtime_config(); 786 let mut dave = DaveInternalState::default(); 787 dave.set_session_protocol(dave_protocol_version); 788 789 let initial_state = ConnectionInternalState { 790 config: runtime_config, 791 heartbeat_interval_ms: hello_data.heartbeat_interval_ms(), 792 last_seq, 793 ready, 794 discovery, 795 selected_mode, 796 session_description: Some(session_description), 797 rtp_codecs: Some(rtp_codecs), 798 connected_user_ids: Arc::new(HashSet::from([user_id])), 799 ssrc_users: Arc::new(HashMap::new()), 800 speaking: Arc::new(HashMap::new()), 801 dave, 802 roster_authoritative: false, 803 resumed: false, 804 }; 805 let mut state = ConnectionStateStore::new(initial_state); 806 replay_pending_voice_events(&mut state, pending_events, &observer)?; 807 let transport_crypto = { 808 let session_description = state 809 .internal() 810 .session_description 811 .as_ref() 812 .ok_or(Error::Protocol(ProtocolError::MissingSessionDescription))?; 813 TransportCrypto::from_config(&session_description.transport_crypto)? 814 }; 815 let (command_tx, command_rx) = mpsc::channel::>(128); 816 let (media_tx, media_rx) = mpsc::channel::(tuning.media_queue_capacity.get()); 817 let close = ConnectionClose::new(); 818 let state_rx = state.subscribe_public(); 819 let task = tokio::spawn( 820 ConnectionDriver { 821 write, 822 read, 823 command_rx, 824 media_rx, 825 close: close.clone(), 826 dave: if let Some(dave_identity) = dave_identity { 827 DaveCoordinator::new_with_identity(user_id, channel_id, dave_identity)? 828 } else { 829 DaveCoordinator::new(user_id, channel_id)? 830 }, 831 receive: ReceivePipeline::new(tuning), 832 send: SendPipeline::new(state.internal())?, 833 transport_crypto, 834 pending_media_ready_waits: DeadlineQueue::default(), 835 pending_media_ready_sends: DeadlineQueue::default(), 836 state, 837 observer: observer.clone(), 838 udp_socket, 839 } 840 .run(), 841 ); 842 let abort = task.abort_handle(); 843 let join_tx = spawn_voice_connection_join_task(task); 844 845 Ok(Connection { 846 inner: Arc::new(ConnectionInner { 847 state_rx, 848 command_tx, 849 media_tx, 850 close, 851 join_tx, 852 abort, 853 observer, 854 }), 855 }) 856} ``` -------------------------------- ### Get strong count from raw pointer with Arc::strong_count_from_raw Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Retrieves the strong reference count from a raw pointer. Requires the arc_raw_get_strong feature gate. ```rust #![feature(arc_raw_get_strong)] use std::sync::Arc; let five = Arc::new(5); let _also_five = Arc::clone(&five); let ptr = Arc::into_raw(five); unsafe { // This assertion is deterministic because we haven't shared // the `Arc` between threads. assert_eq!(2, Arc::strong_count_from_raw(ptr)); // Convert back to an `Arc` to avoid leaking memory. let five = Arc::from_raw(ptr); assert_eq!(2, Arc::strong_count(&five)); } ``` -------------------------------- ### Samples implementation for AsRef<[i16]> Source: https://docs.rs/cacophony/latest/src/cacophony/pcm.rs.html Implements Samples for S16 sample encoding, converting i16 samples to f32 using s16_to_f32. ```rust impl Samples for T where T: AsRef<[i16]>, { fn sample_count(&self) -> Result { Ok(self.as_ref().len()) } fn append_f32(&self, output: &mut Vec) -> Result<()> { let samples = self.as_ref(); output.reserve(samples.len()); output.extend(samples.iter().map(|sample| s16_to_f32(*sample))); Ok(()) } } ``` -------------------------------- ### pub fn weak_count(this: &Arc) -> usize Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Gets the number of `Weak` pointers to this allocation. This method is safe, but using it correctly requires extra care because another thread can change the weak count at any time. ```APIDOC ## `pub fn weak_count(this: &Arc) -> usize` ### Description Gets the number of `Weak` pointers to this allocation. This method is safe, but using it correctly requires extra care because another thread can change the weak count at any time. ### Method Associated function ### Signature `pub fn weak_count(this: &Arc) -> usize` ### Parameters - `this` (`&Arc`) - The `Arc` whose weak count is to be retrieved. ### Returns - `usize` - The number of `Weak` pointers to the allocation. ### Safety This method by itself is safe, but using it correctly requires extra care. Another thread can change the weak count at any time, including potentially between calling this method and acting on the result. ### Examples ```rust use std::sync::Arc; let five = Arc::new(5); let _weak_five = Arc::downgrade(&five); // This assertion is deterministic because we haven't shared // the `Arc` or `Weak` between threads. assert_eq!(1, Arc::weak_count(&five)); ``` ### Availability 1.15.0 ``` -------------------------------- ### pub fn strong_count(this: &Arc) -> usize Source: https://docs.rs/cacophony/latest/cacophony/type.ConnectionStateSnapshot.html Gets the number of strong (`Arc`) pointers to this allocation. This method is safe, but using it correctly requires extra care because another thread can change the strong count at any time. ```APIDOC ## `pub fn strong_count(this: &Arc) -> usize` ### Description Gets the number of strong (`Arc`) pointers to this allocation. This method is safe, but using it correctly requires extra care because another thread can change the strong count at any time. ### Method Associated function ### Signature `pub fn strong_count(this: &Arc) -> usize` ### Parameters - `this` (`&Arc`) - The `Arc` whose strong count is to be retrieved. ### Returns - `usize` - The number of strong (`Arc`) pointers to the allocation. ### Safety This method by itself is safe, but using it correctly requires extra care. Another thread can change the strong count at any time, including potentially between calling this method and acting on the result. ### Examples ```rust use std::sync::Arc; let five = Arc::new(5); let _also_five = Arc::clone(&five); // This assertion is deterministic because we haven't shared // the `Arc` between threads. assert_eq!(2, Arc::strong_count(&five)); ``` ``` -------------------------------- ### FromIterator Result short-circuiting example Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Shows that no further elements are processed after the first error is encountered during collection. ```rust let v = vec![3, 2, 1, 10]; let mut shared = 0; let res: Result, &'static str> = v.iter().map(|x: &u32| { shared += x; x.checked_sub(2).ok_or("Underflow!") }).collect(); assert_eq!(res, Err("Underflow!")); assert_eq!(shared, 6); ``` -------------------------------- ### PlaybackAudio Implementation Methods Source: https://docs.rs/cacophony/latest/src/cacophony/opus.rs.html Methods for creating PlaybackAudio from chunks or mono samples, and accessing packet and source metadata. ```rust impl PlaybackAudio { pub fn from_chunks(chunks: &[PcmChunk]) -> Result { let mut encoder = DynamicPcmEncoder::new(); let mut packets = Vec::new(); for chunk in chunks { encoder.push_chunk_to(chunk, &mut packets)?; } encoder.finish_to(&mut packets)?; Ok(Self { packets, source: PlaybackAudioSource { sample_rate_hz: NonZeroU32::new(encoder.source_sample_rate_hz()) .expect("default Discord source sample rate is nonzero"), channel_count: NonZeroU32::new( u32::try_from(encoder.source_channel_count()) .expect("Discord source channel count fits u32"), ) .expect("default Discord source channel count is nonzero"), pcm_bytes: encoder.source_pcm_bytes(), }, }) } pub fn from_mono_samples(samples: &[f32], source: PlaybackAudioSource) -> Result { Ok(Self { packets: PcmEncoder::::encode_all( source.sample_rate_hz, samples, EncodeConfig::default(), )?, source, }) } pub fn packets(&self) -> &[Packet] { &self.packets } pub fn clone_packets(&self) -> Vec { self.packets.clone() } pub fn source(&self) -> PlaybackAudioSource { self.source } pub fn source_sample_rate_hz(&self) -> u32 { self.source.sample_rate_hz.get() } pub fn source_channel_count(&self) -> u32 { self.source.channel_count.get() } pub fn source_pcm_bytes(&self) -> usize { self.source.pcm_bytes } pub fn packet_count(&self) -> usize { self.packets.len() } } ``` -------------------------------- ### Handle DavePrepareTransition Opcode Source: https://docs.rs/cacophony/latest/src/cacophony/gateway.rs.html Prepares a DAVE transition by updating the state and triggering a retry for pending media. ```rust Some(Opcode::DavePrepareTransition) => { let transition: DavePrepareTransitionEvent = parse_data(event.data)?; self.state_store.update(|state| { state .dave .prepare_transition(transition.transition_id, transition.protocol_version); }); effects .retry_dave_pending_media .include(DavePendingMediaRetry::dave_state()); } ``` -------------------------------- ### DaveInternalState prepare_transition method Source: https://docs.rs/cacophony/latest/src/cacophony/state.rs.html Prepares a transition with the given transition ID and protocol version. Clears pending MLS messages if the transition ID changes, updates the phase to PreparedTransition or PreparedEpoch, and executes the transition if conditions are met. ```rust pub(crate) fn prepare_transition(&mut self, transition_id: u16, protocol_version: u16) { if self.transition_id() != Some(transition_id) { self.clear_pending_mls(); } let protocol = DaveProtocolState::from_prepared(protocol_version); self.phase = match self.phase { DaveGatewayPhase::PreparedEpoch { epoch, seq, .. } => DaveGatewayPhase::PreparedEpoch { protocol, epoch, seq, transition_id: Some(transition_id), }, DaveGatewayPhase::Unknown | DaveGatewayPhase::Session { .. } | DaveGatewayPhase::PreparedTransition { .. } => DaveGatewayPhase::PreparedTransition { transition_id, protocol, }, }; if self.epoch() == Some(1) && transition_id == 0 && protocol.is_dave() { self.execute_transition(transition_id); } } ``` -------------------------------- ### unwrap basic usage Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Demonstrates the basic usage of unwrap and its panic behavior. ```rust let x: Result = Ok(2); assert_eq!(x.unwrap(), 2); ``` ```rust let x: Result = Err("emergency failure"); x.unwrap(); // panics with `emergency failure` ``` -------------------------------- ### and_then usage Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Demonstrates using and_then for control flow. ```rust fn sq_then_to_string(x: u32) -> Result { x.checked_mul(x).map(|sq| sq.to_string()).ok_or("overflowed") } ``` -------------------------------- ### Samples implementation for AsRef<[u8]> Source: https://docs.rs/cacophony/latest/src/cacophony/pcm.rs.html Implements Samples for S16Le (little-endian i16) from raw bytes, checking byte count and converting each 2-byte chunk to f32. ```rust impl Samples for T where T: AsRef<[u8]>, { fn sample_count(&self) -> Result { checked_byte_sample_count::(self.as_ref().len()) } fn append_f32(&self, output: &mut Vec) -> Result<()> { let bytes = self.as_ref(); let samples = checked_byte_sample_count::(bytes.len())?; output.reserve(samples); output.extend( bytes .chunks_exact(2) .map(|sample| s16_to_f32(i16::from_le_bytes([sample[0], sample[1]]))), ); Ok(()) } } ``` -------------------------------- ### Default ConnectionTuning values Source: https://docs.rs/cacophony/latest/src/cacophony/state.rs.html Default values for ConnectionTuning: 10s pending media TTL, 256 interarrival window, 60ms reorder TTL, 32 max reorder frames, u16::MAX UDP buffer, 4096 ready frame buffer, 256 media queue capacity. ```rust impl Default for ConnectionTuning { fn default() -> Self { Self { dave_pending_media_ttl: Duration::from_secs(10), receive_interarrival_window: 256, rtp_reorder_ttl: Duration::from_millis(60), rtp_reorder_buffer_max_frames: 32, udp_receive_buffer_bytes: u16::MAX as usize, ready_frame_buffer_max: 4096, media_queue_capacity: 256, } } } ``` -------------------------------- ### Construct codec table from session description and preferences Source: https://docs.rs/cacophony/latest/src/cacophony/codecs.rs.html Builds a payload codec table and expected payload type list from a session description and codec preferences. Validates the video codec, inserts the audio codec first, then each preferred video codec, and returns the populated structure. ```rust if let Some(codec) = session_description.video_codec.as_deref() { video_codec(codec).ok_or_else(|| UnsupportedCodecError::UnsupportedVideoCodec { codec: codec.to_string(), })?; } let mut payload_codecs = [None; 128]; let mut expected_payload_types = Vec::with_capacity(1 + codec_preferences.video_codecs().len()); Self::insert_codec( &mut payload_codecs, &mut expected_payload_types, audio_codec, ); for codec in codec_preferences.video_codecs().iter().copied() { Self::insert_codec(&mut payload_codecs, &mut expected_payload_types, codec); } Ok(Self { payload_codecs, expected_payload_types, }) ``` ```rust pub(crate) fn detect( &self, payload_type: RtpPayloadType, ) -> Result { if let Some(codec) = self .payload_codecs .get(payload_type.index()) .and_then(|codec| *codec) { return Ok(codec); } if let Some(codec) = codec_for_payload_type(payload_type) { return Err(UnsupportedCodecError::UnexpectedRtpPayloadCodec { payload_type, codec, expected_payload_types: self.expected_payload_types.clone(), }); } Err(UnsupportedCodecError::UnsupportedRtpPayloadType { payload_type, expected_payload_types: self.expected_payload_types.clone(), }) } ``` ```rust fn insert_codec( payload_codecs: &mut [Option; 128], expected_payload_types: &mut Vec, codec: Codec, ) { let rtp_payload_type = payload_type(codec); payload_codecs[rtp_payload_type.index()] = Some(codec); expected_payload_types.push(rtp_payload_type); } ``` -------------------------------- ### CapturedPcmAudio::new Source: https://docs.rs/cacophony/latest/cacophony/opus/struct.CapturedPcmAudio.html Creates a new CapturedPcmAudio instance from a PcmChunk. Returns a Result indicating success or failure. ```APIDOC ## CapturedPcmAudio::new ### Description Creates a new CapturedPcmAudio instance from a PcmChunk. ### Method new ### Signature pub fn new(pcm: PcmChunk) -> Result ### Parameters - **pcm** (PcmChunk) - Required - The PCM audio data to wrap. ### Returns - Result - A Result containing the new CapturedPcmAudio instance on success, or an error on failure. ``` -------------------------------- ### IntoIterator Result implementation example Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Demonstrates consuming a Result into an iterator, yielding the contained value if Ok or nothing if Err. ```rust let x: Result = Ok(5); let v: Vec = x.into_iter().collect(); assert_eq!(v, [5]); let x: Result = Err("nothing!"); let v: Vec = x.into_iter().collect(); assert_eq!(v, []); ``` -------------------------------- ### GatewayReady struct and encryption/stream helpers Source: https://docs.rs/cacophony/latest/src/cacophony/gateway.rs.html Defines the ready state of the gateway and provides methods for selecting encryption modes and identifying primary video streams. ```rust #[derive(Clone, Debug, Deserialize, PartialEq, Eq)] pub struct GatewayReady { pub ssrc: u32, pub ip: String, pub port: u16, #[serde(default)] pub modes: Vec, #[serde(default)] pub heartbeat_interval: Option, #[serde(default)] pub streams: Vec, } impl GatewayReady { pub(crate) fn select_encryption_mode( &self, options: &ValidatedConnectionOptions, ) -> Result { if self.modes.is_empty() { return Err(Error::Protocol(ProtocolError::ReadyMissingEncryptionModes)); } if let Some(required_mode) = options.required_mode { if self.offers_mode(required_mode) { return Ok(required_mode); } return Err(Error::Protocol( ProtocolError::RequiredEncryptionModeUnavailable { required_mode, modes: self.modes.clone(), }, )); } for mode in EncryptionMode::ALL { if self.offers_mode(mode) { return Ok(mode); } } Err(Error::Protocol( ProtocolError::ReadyMissingSupportedEncryptionMode { modes: self.modes.clone(), }, )) } pub(crate) fn primary_video_stream(&self) -> Option { self.streams .iter() .filter(|stream| stream.is_active_video()) .max_by_key(|stream| { ( stream.rid.as_deref() == Some(PRIMARY_VIDEO_RID), stream.quality == Some(PRIMARY_VIDEO_QUALITY), stream.quality.unwrap_or_default(), ) }) .or_else(|| { (self.streams.len() == 1 && self.streams[0].active != Some(false)) .then(|| &self.streams[0]) }) .map(|stream| GatewayReadyVideoStream { ssrc: stream.ssrc, rtx_ssrc: stream.rtx_ssrc, }) } fn offers_mode(&self, mode: EncryptionMode) -> bool { self.modes .iter() .any(|offered| offered.supported() == Some(mode)) } } ``` -------------------------------- ### Default ConnectionRuntimeTuning Source: https://docs.rs/cacophony/latest/src/cacophony/state.rs.html Default implementation that validates the default ConnectionTuning and panics if invalid. ```rust impl Default for ConnectionRuntimeTuning { fn default() -> Self { ConnectionTuning::default() .validate() .expect("default connection tuning is valid") } } ``` -------------------------------- ### cloned on Result<&T, E> Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Example for the `cloned` method on `Result<&T, E>`, which clones the referenced value into a new `Result`. ```rust let val = 12; let x: Result<&i32, i32> = Ok(&val); assert_eq!(x, Ok(&12)); let cloned = x.cloned(); assert_eq!(cloned, Ok(12)); ``` -------------------------------- ### copied on Result<&T, E> Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Example for the `copied` method on `Result<&T, E>`, which copies the referenced value into a new `Result`. ```rust let val = 12; let x: Result<&i32, i32> = Ok(&val); assert_eq!(x, Ok(&12)); let copied = x.copied(); assert_eq!(copied, Ok(12)); ``` -------------------------------- ### and usage Source: https://docs.rs/cacophony/latest/cacophony/type.Result.html Demonstrates the behavior of the and method with different Result combinations. ```rust let x: Result = Ok(2); let y: Result<&str, &str> = Err("late error"); assert_eq!(x.and(y), Err("late error")); let x: Result = Err("early error"); let y: Result<&str, &str> = Ok("foo"); assert_eq!(x.and(y), Err("early error")); let x: Result = Err("not a 2"); let y: Result<&str, &str> = Err("late error"); assert_eq!(x.and(y), Err("not a 2")); let x: Result = Ok(2); let y: Result<&str, &str> = Ok("different result type"); assert_eq!(x.and(y), Ok("different result type")); ``` -------------------------------- ### Samples implementation for AsRef<[u8]> Source: https://docs.rs/cacophony/latest/src/cacophony/pcm.rs.html Implements Samples for F32Le (little-endian f32) from raw bytes, checking byte count and converting each 4-byte chunk to f32 with clamping. ```rust impl Samples for T where T: AsRef<[u8]>, { fn sample_count(&self) -> Result { checked_byte_sample_count::(self.as_ref().len()) } fn append_f32(&self, output: &mut Vec) -> Result<()> { let bytes = self.as_ref(); let samples = checked_byte_sample_count::(bytes.len())?; output.reserve(samples); output.extend(bytes.chunks_exact(4).map(|sample| { f32::from_le_bytes([sample[0], sample[1], sample[2], sample[3]]).clamp(-1.0, 1.0) })); Ok(()) } } ``` -------------------------------- ### OfferedEncryptionMode::new Source: https://docs.rs/cacophony/latest/cacophony/struct.OfferedEncryptionMode.html Creates a new OfferedEncryptionMode from a string-like value. ```APIDOC ## OfferedEncryptionMode::new ### Description Creates a new OfferedEncryptionMode instance from a string-like value. ### Method `pub fn new(mode: impl Into) -> Self` ### Parameters - `mode` (impl Into) - Required - The encryption mode as a string-like value. ### Returns - `Self` - The new OfferedEncryptionMode instance. ```