mollusk/peerspeak
Public Access
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases 10 Wiki Activity

Initialize project and implement decentralized voice chat client (PipeWire, Opus, Iroh, Iced)

Browse Source
This commit is contained in:
mollusk
2026-05-27 05:18:56 -04:00
commit 1220d94e91
15 changed files with 9177 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -0,0 +1 @@
/target
Cargo.lock
Generated
+7327
View File
File diff suppressed because it is too large Load Diff
Cargo.toml
+23
View File
@@ -0,0 +1,23 @@
[package]
name = "peerspeak"
version = "0.1.0"
edition = "2024"
[dependencies]
anyhow = "1.0.102"
async-trait = "0.1.89"
base64 = "0.22.1"
bytes = "1.11.1"
iced = "0.14.0"
iroh = "1.0.0-rc.0"
iroh-gossip = "0.99.0"
iroh-tickets = "1.0.0-rc.0"
opus = "0.3.1"
pipewire = "0.9"
rand = "0.10.1"
ringbuf = "0.5.0"
serde = { version = "1.0.228", features = ["derive"] }
serde_json = "1.0.150"
thiserror = "2.0.18"
tokio = { version = "1.52.3", features = ["full"] }
tokio-stream = "0.1.18"
antigravity.toml
+31
View File
@@ -0,0 +1,31 @@
# Example entry in an antigravity.toml configuration file
[agent]
model = "gemini-3.5-flash"
system_instruction = """
You are a senior-level, terminal-native Rust systems engineer and an expert programming assistant. Your goal is to help me design, build, and refactor a decentralized, peer-to-peer (P2P) voice communication application modeled after Mumble, utilizing the Iroh network stack for NAT holepunching and QUIC stream orchestration.
### 1. Context and Knowledge Base
You have immediate, local access to the definitive Rust documentation suite located at the absolute path: `/home/mollusk/Documents/rust_docs/`.
Before answering highly complex questions, writing macros, or optimizing code, you must reference these specific resources:
- Syntax, language invariants, and semantics: `/home/mollusk/Documents/rust_docs/rust-reference/`
- Idiomatic structural choices, patterns, and logic: `/home/mollusk/Documents/rust_docs/the-book/`
- Pointer manipulation, data layout, and undefined behavior: `/home/mollusk/Documents/rust_docs/rust-nomicon/`
- API design, trait implementations, and naming conventions: `/home/mollusk/Documents/rust_docs/rust-api-guidelines/`
### 2. Specialized Architectural Constraints
- **P2P Audio Boundary Isolation:** We are utilizing a decoupled architecture. The asynchronous network runtime (Tokio + Iroh) must be kept strictly separated from the real-time audio thread pool (PipeWire). Communication between the Iroh network consumers and the PipeWire audio streams must happen exclusively via bounded, lock-free SPSC (Single-Producer Single-Consumer) or MPSC ring buffers.
- **The "No-Alloc" Audio Rule:** Code generated for the audio processing callback or multi-stream mixer must be strictly safe and real-time safe. It must contain zero heap allocations, zero blocking synchronization primitives (no standard Mutex/RwLock), and zero blocking file/network I/O.
- **Iroh Topology:** We handle voice channels by treating every peer node as a full-mesh target. Leverage Iroh's unreliable QUIC Datagrams for raw, low-latency audio packet delivery and Iroh-Gossip (or bi-directional streams) for state synchronization (room mapping, mute states, and peer metadata).
### 3. Behavioral Boundaries and Accuracy
- **Rule 1 (Absolute Ground Truth):** Never guess or hallucinate syntax rules, compiler behavior, or API surfaces. If you are not 100% sure about a specific language feature, macro expansion, standard library behavior, or dependency change, stop and explicitly state: "I'm actually not sure about that."
- **Rule 2 (No "C in Rust"):** Do not write C-style logic wrapped in Rust syntax. Prioritize idiomatic Rust patterns (e.g., using algebraic data types, proper trait bounds, combinators like `.map()` or `.and_then()`, and precise error handling with `Result` and `Option`).
- **Rule 3 (Safe by Default):** Always default to safe, idiomatic Rust code. Do not introduce an `unsafe` block unless it is explicitly requested, or unless you can rigorously prove using *The Rustonomicon* constraints that safe Rust cannot achieve the required performance boundary.
### 4. Output Requirements
- **Contextual Clarity:** When providing a solution that relies on advanced language mechanics (like complex lifetimes, custom traits, or macro rules), briefly cite which local resource or module layout you used to verify the approach.
- **Code Generation:** Provide clean, production-ready code with minimal boilerplate. Use standard formatting rules (`rustfmt` styles). Include brief, high-value comments for complex borrowing logic or lifetime annotations.
- **Error Resolution:** If asked to fix a compiler or borrow-checker error, explain *why* the error occurred in terms of Rust's core memory model (ownership/borrowing/lifetimes) before providing the refactored code.
Acknowledge these operational parameters, summarize your core objectives, and ask me for the details of our new project.
"""
src/app/mod.rs
+490
View File
@@ -0,0 +1,490 @@
use crate::core::{CoreController, messages::{CoreCommand, UiEvent}};
use crate::network::PeerState;
use iced::widget::{
container, column, row, text, button, text_input, scrollable, Column,
};
use iced::{
Color, Background, Border, Element, Subscription, Task, Theme,
};
use iroh::EndpointId;
use std::collections::HashMap;
use std::sync::{Arc, OnceLock};
use tokio::sync::Mutex;
static UI_RX: OnceLock<Mutex<Option<tokio::sync::mpsc::Receiver<UiEvent>>>> = OnceLock::new();
#[derive(Debug, Clone)]
pub enum AppMessage {
NicknameChanged(String),
TicketInputChanged(String),
JoinPressed,
CreatePressed,
LeavePressed,
ToggleMutePressed,
ToggleDeafenPressed,
UiEventReceived(UiEvent),
CopyToClipboard,
}
fn core_subscription() -> impl iced::futures::Stream<Item = UiEvent> {
iced::stream::channel(100, |mut output: iced::futures::channel::mpsc::Sender<UiEvent>| async move {
if let Some(rx_lock) = UI_RX.get() {
let mut guard = rx_lock.lock().await;
if let Some(mut rx) = guard.take() {
use iced::futures::sink::SinkExt;
while let Some(event) = rx.recv().await {
let _ = output.send(event).await;
}
}
}
})
}
pub struct AppState {
name: String,
ticket_input: String,
status_message: String,
self_id: String,
ticket: String,
is_muted: bool,
is_deafened: bool,
peers: HashMap<EndpointId, PeerState>,
audio_levels: HashMap<EndpointId, f32>,
controller: Arc<CoreController>,
}
impl Default for AppState {
fn default() -> Self {
let (ui_tx, ui_rx) = tokio::sync::mpsc::channel(100);
let controller = Arc::new(CoreController::new(ui_tx));
let _ = UI_RX.set(Mutex::new(Some(ui_rx)));
Self {
name: "Peer".to_string(),
ticket_input: "".to_string(),
status_message: "Ready to connect".to_string(),
self_id: "".to_string(),
ticket: "".to_string(),
is_muted: false,
is_deafened: false,
peers: HashMap::new(),
audio_levels: HashMap::new(),
controller,
}
}
}
fn theme(_state: &AppState) -> Theme {
Theme::Dark
}
pub fn run_gui() -> iced::Result {
iced::application(AppState::default, update, view)
.title("PeerSpeak P2P Voice Chat")
.theme(theme)
.subscription(subscription)
.window(iced::window::Settings {
size: iced::Size::new(900.0, 600.0),
position: iced::window::Position::Centered,
exit_on_close_request: true,
..Default::default()
})
.run()
}
fn subscription(_state: &AppState) -> Subscription<AppMessage> {
Subscription::run(core_subscription).map(AppMessage::UiEventReceived)
}
fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
match message {
AppMessage::NicknameChanged(val) => {
state.name = val;
}
AppMessage::TicketInputChanged(val) => {
state.ticket_input = val;
}
AppMessage::JoinPressed => {
state.status_message = "Connecting to room...".to_string();
let _ = state.controller.send(CoreCommand::Join {
name: state.name.clone(),
ticket: state.ticket_input.clone(),
});
}
AppMessage::CreatePressed => {
state.status_message = "Creating room...".to_string();
let _ = state.controller.send(CoreCommand::Join {
name: state.name.clone(),
ticket: "".to_string(),
});
}
AppMessage::LeavePressed => {
let _ = state.controller.send(CoreCommand::Leave);
}
AppMessage::ToggleMutePressed => {
let _ = state.controller.send(CoreCommand::ToggleMute);
state.is_muted = !state.is_muted;
}
AppMessage::ToggleDeafenPressed => {
let _ = state.controller.send(CoreCommand::ToggleDeafen);
state.is_deafened = !state.is_deafened;
}
AppMessage::UiEventReceived(event) => {
match event {
UiEvent::RoomJoined { ticket, self_id } => {
state.ticket = ticket;
state.self_id = self_id;
state.status_message = "Connected to room".to_string();
state.peers.clear();
state.audio_levels.clear();
}
UiEvent::RoomLeft => {
state.ticket = "".to_string();
state.peers.clear();
state.audio_levels.clear();
state.status_message = "Ready to connect".to_string();
}
UiEvent::PeerJoined { id, state: peer_state } => {
state.peers.insert(id, peer_state);
}
UiEvent::PeerLeft { id } => {
state.peers.remove(&id);
state.audio_levels.remove(&id);
}
UiEvent::PeerUpdated { id, state: peer_state } => {
state.peers.insert(id, peer_state);
}
UiEvent::AudioLevels(levels) => {
for (id, val) in levels {
state.audio_levels.insert(id, val);
}
}
UiEvent::Error(err) => {
state.status_message = format!("Error: {}", err);
}
}
}
AppMessage::CopyToClipboard => {
if !state.ticket.is_empty() {
return iced::clipboard::write(state.ticket.clone());
}
}
}
Task::none()
}
fn horizontal_space() -> iced::widget::Space {
iced::widget::Space::new().width(iced::Length::Fill)
}
fn vertical_space(height: f32) -> iced::widget::Space {
iced::widget::Space::new().height(height)
}
fn view(state: &AppState) -> Element<'_, AppMessage> {
// Theme Colors
let color_crust = Color::from_rgb8(17, 17, 27);
let color_mantle = Color::from_rgb8(24, 24, 37);
let color_base = Color::from_rgb8(30, 30, 46);
let color_surface = Color::from_rgb8(49, 50, 68);
let color_text = Color::from_rgb8(205, 214, 244);
let color_subtext = Color::from_rgb8(166, 173, 200);
let color_blue = Color::from_rgb8(137, 180, 250);
let color_lavender = Color::from_rgb8(180, 190, 254);
let color_red = Color::from_rgb8(243, 139, 168);
let color_maroon = Color::from_rgb8(233, 146, 160);
let color_green = Color::from_rgb8(166, 227, 161);
// Style Helpers
let c_style = move |bg: Color, b_color: Color, radius: f32| {
move |_theme: &Theme| container::Style {
text_color: Some(color_text),
background: Some(Background::Color(bg)),
border: Border {
color: b_color,
width: if b_color == Color::TRANSPARENT { 0.0 } else { 1.0 },
radius: radius.into(),
},
..Default::default()
}
};
let b_style = move |bg: Color, hover_bg: Color, text_c: Color, radius: f32| {
move |_theme: &Theme, status: button::Status| {
let active_bg = match status {
button::Status::Hovered => hover_bg,
_ => bg,
};
button::Style {
background: Some(Background::Color(active_bg)),
text_color: text_c,
border: Border {
color: Color::TRANSPARENT,
width: 0.0,
radius: radius.into(),
},
..Default::default()
}
}
};
let t_style = move |_theme: &Theme, _status: text_input::Status| {
text_input::Style {
background: Background::Color(color_crust),
border: Border {
color: color_surface,
width: 1.0,
radius: 6.0.into(),
},
icon: color_subtext,
placeholder: Color::from_rgb8(108, 112, 134),
value: color_text,
selection: color_blue,
}
};
if state.ticket.is_empty() {
// --- HOME SCREEN ---
let logo = text("PEERSPEAK")
.size(36)
.color(color_blue);
let subtitle = text("NAT-traversing full-mesh voice chat")
.size(16)
.color(color_subtext);
let nickname_input = column![
text("Nickname").size(14).color(color_subtext),
vertical_space(4.0),
text_input("Enter nickname...", &state.name)
.on_input(AppMessage::NicknameChanged)
.style(t_style)
.padding(10)
];
let create_btn = button(
text("Create New Room")
.size(16)
.align_x(iced::alignment::Horizontal::Center)
)
.on_press(AppMessage::CreatePressed)
.style(b_style(color_blue, color_lavender, color_crust, 8.0))
.padding(12)
.width(iced::Length::Fill);
let join_group = column![
text("Join Existing Room").size(14).color(color_subtext),
vertical_space(4.0),
text_input("Paste room ticket here...", &state.ticket_input)
.on_input(AppMessage::TicketInputChanged)
.style(t_style)
.padding(10),
vertical_space(8.0),
button(
text("Join Room")
.size(16)
.align_x(iced::alignment::Horizontal::Center)
)
.on_press(AppMessage::JoinPressed)
.style(b_style(color_surface, color_blue, color_text, 8.0))
.padding(12)
.width(iced::Length::Fill)
];
let status = text(&state.status_message)
.size(14)
.color(color_subtext);
let content = container(
column![
logo,
subtitle,
vertical_space(20.0),
nickname_input,
vertical_space(16.0),
create_btn,
vertical_space(16.0),
text("— OR —").size(12).color(color_surface).align_x(iced::alignment::Horizontal::Center),
vertical_space(16.0),
join_group,
vertical_space(20.0),
status
]
.spacing(10)
.align_x(iced::alignment::Horizontal::Center)
)
.style(c_style(color_mantle, color_surface, 12.0))
.padding(30)
.width(420);
container(content)
.width(iced::Length::Fill)
.height(iced::Length::Fill)
.align_x(iced::alignment::Horizontal::Center)
.align_y(iced::alignment::Vertical::Center)
.style(c_style(color_crust, Color::TRANSPARENT, 0.0))
.into()
} else {
// --- ROOM SCREEN ---
let header = row![
text("PEERSPEAK")
.size(20)
.color(color_blue),
horizontal_space(),
text(format!("My ID: {}", &state.self_id[..8]))
.size(14)
.color(color_subtext),
horizontal_space(),
button(text("Copy Ticket").size(12))
.on_press(AppMessage::CopyToClipboard)
.style(b_style(color_surface, color_blue, color_text, 6.0))
.padding(6)
]
.align_y(iced::alignment::Vertical::Center);
let header_container = container(header)
.style(c_style(color_mantle, color_surface, 8.0))
.padding(15)
.width(iced::Length::Fill);
// Peers Column
let mut peers_list = Column::new().spacing(10);
// Add ourselves
let self_card = container(
row![
text(format!("{} (You)", &state.name)).size(16).color(color_text),
horizontal_space(),
if state.is_muted {
text("[Muted]").size(14).color(color_red)
} else {
text("[Active]").size(14).color(color_green)
}
]
.align_y(iced::alignment::Vertical::Center)
)
.style(c_style(color_base, color_surface, 6.0))
.padding(12);
peers_list = peers_list.push(self_card);
for (peer_id, peer) in &state.peers {
let level = state.audio_levels.get(peer_id).copied().unwrap_or(0.0);
let is_speaking = level > 0.01;
let indicator = if peer.is_muted {
text("[Muted]").size(14).color(color_red)
} else if is_speaking {
text("[Speaking]").size(14).color(color_green)
} else {
text("[Idle]").size(14).color(color_subtext)
};
let card = container(
row![
column![
text(&peer.name).size(16).color(color_text),
text(format!("ID: {}", &peer_id.to_string()[..8])).size(11).color(color_subtext)
],
horizontal_space(),
indicator
]
.align_y(iced::alignment::Vertical::Center)
)
.style(c_style(
if is_speaking { color_base } else { color_mantle },
if is_speaking { color_green } else { color_surface },
6.0
))
.padding(12);
peers_list = peers_list.push(card);
}
let scroll_peers = scrollable(peers_list);
let peers_panel = container(
column![
text("Room Participants").size(18).color(color_blue),
vertical_space(10.0),
scroll_peers
]
)
.style(c_style(color_mantle, Color::TRANSPARENT, 0.0))
.padding(15)
.width(iced::Length::FillPortion(2))
.height(iced::Length::Fill);
// Control Panel Column
let mute_text = if state.is_muted { "Unmute Mic" } else { "Mute Mic" };
let mute_bg = if state.is_muted { color_red } else { color_surface };
let mute_hover = if state.is_muted { color_maroon } else { color_blue };
let mute_fg = if state.is_muted { color_crust } else { color_text };
let deafen_text = if state.is_deafened { "Undeafen Audio" } else { "Deafen Audio" };
let deafen_bg = if state.is_deafened { color_red } else { color_surface };
let deafen_hover = if state.is_deafened { color_maroon } else { color_blue };
let deafen_fg = if state.is_deafened { color_crust } else { color_text };
let ctrl_buttons = column![
button(
text(mute_text)
.size(16)
.align_x(iced::alignment::Horizontal::Center)
)
.on_press(AppMessage::ToggleMutePressed)
.style(b_style(mute_bg, mute_hover, mute_fg, 8.0))
.padding(14)
.width(iced::Length::Fill),
vertical_space(10.0),
button(
text(deafen_text)
.size(16)
.align_x(iced::alignment::Horizontal::Center)
)
.on_press(AppMessage::ToggleDeafenPressed)
.style(b_style(deafen_bg, deafen_hover, deafen_fg, 8.0))
.padding(14)
.width(iced::Length::Fill),
vertical_space(30.0),
button(
text("Leave Room")
.size(16)
.align_x(iced::alignment::Horizontal::Center)
)
.on_press(AppMessage::LeavePressed)
.style(b_style(color_red, color_maroon, color_crust, 8.0))
.padding(14)
.width(iced::Length::Fill)
];
let control_panel = container(
column![
text("Controls").size(18).color(color_blue),
vertical_space(15.0),
ctrl_buttons,
vertical_space(20.0),
text(&state.status_message).size(12).color(color_subtext)
]
)
.style(c_style(color_mantle, Color::TRANSPARENT, 0.0))
.padding(15)
.width(iced::Length::FillPortion(1))
.height(iced::Length::Fill);
let main_layout = row![peers_panel, control_panel].spacing(15);
container(
column![
header_container,
vertical_space(15.0),
main_layout
]
)
.padding(15)
.width(iced::Length::Fill)
.height(iced::Length::Fill)
.style(c_style(color_crust, Color::TRANSPARENT, 0.0))
.into()
}
}
src/audio/mod.rs
+31
View File
@@ -0,0 +1,31 @@
use std::sync::mpsc::{Sender, Receiver};
use thiserror::Error;
#[derive(Error, Debug)]
pub enum AudioError {
#[error("Failed to initialize audio backend: {0}")]
Init(String),
#[error("Audio device error: {0}")]
Device(String),
#[error("Stream error: {0}")]
Stream(String),
#[error("Audio buffer overflow/underflow")]
BufferError,
#[error("Other audio error: {0}")]
Other(String),
}
pub trait AudioBackend: Send + Sync {
/// Starts capturing raw PCM audio from the input device (microphone),
/// sending chunks of samples (e.g. `Vec<i16>`) to the provided Sender.
fn start_capture(&self, tx: Sender<Vec<i16>>) -> Result<(), AudioError>;
/// Starts playing back raw PCM audio to the output device (speaker),
/// reading mixed/incoming chunks of samples from the provided Receiver.
fn start_playback(&self, rx: Receiver<Vec<i16>>) -> Result<(), AudioError>;
/// Stops both capture and playback streams.
fn stop(&self) -> Result<(), AudioError>;
}
pub mod pipewire_impl;
src/audio/pipewire_impl.rs
+328
View File
@@ -0,0 +1,328 @@
use crate::audio::{AudioBackend, AudioError};
use std::sync::mpsc::{Sender, Receiver};
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread::{self, JoinHandle};
use std::time::Duration;
use pipewire as pw;
use pw::{properties::properties, spa};
use spa::pod::Pod;
use ringbuf::{HeapRb, traits::{Consumer, Producer, Split}};
pub struct PipeWireBackend {
capture_state: Mutex<Option<CaptureState>>,
playback_state: Mutex<Option<PlaybackState>>,
}
struct CaptureState {
cmd_tx: pw::channel::Sender<()>,
thread: JoinHandle<()>,
}
struct PlaybackState {
cmd_tx: pw::channel::Sender<()>,
thread: JoinHandle<()>,
}
impl PipeWireBackend {
pub fn new() -> Self {
pw::init();
Self {
capture_state: Mutex::new(None),
playback_state: Mutex::new(None),
}
}
}
impl AudioBackend for PipeWireBackend {
fn start_capture(&self, tx: Sender<Vec<i16>>) -> Result<(), AudioError> {
let mut capture_guard = self.capture_state.lock().unwrap();
if capture_guard.is_some() {
return Err(AudioError::Stream("Capture already started".to_string()));
}
let (cmd_tx, cmd_rx) = pw::channel::channel::<()>();
let tx_clone = tx.clone();
let thread = thread::Builder::new()
.name("peerspeak-capture".to_string())
.spawn(move || {
if let Err(e) = run_capture(cmd_rx, tx_clone) {
eprintln!("Capture thread error: {:?}", e);
}
})
.map_err(|e| AudioError::Init(e.to_string()))?;
*capture_guard = Some(CaptureState { cmd_tx, thread });
Ok(())
}
fn start_playback(&self, rx: Receiver<Vec<i16>>) -> Result<(), AudioError> {
let mut playback_guard = self.playback_state.lock().unwrap();
if playback_guard.is_some() {
return Err(AudioError::Stream("Playback already started".to_string()));
}
let (cmd_tx, cmd_rx) = pw::channel::channel::<()>();
let thread = thread::Builder::new()
.name("peerspeak-playback".to_string())
.spawn(move || {
if let Err(e) = run_playback(cmd_rx, rx) {
eprintln!("Playback thread error: {:?}", e);
}
})
.map_err(|e| AudioError::Init(e.to_string()))?;
*playback_guard = Some(PlaybackState { cmd_tx, thread });
Ok(())
}
fn stop(&self) -> Result<(), AudioError> {
// Stop capture
let mut capture_guard = self.capture_state.lock().unwrap();
if let Some(state) = capture_guard.take() {
let _ = state.cmd_tx.send(());
let _ = state.thread.join();
}
// Stop playback
let mut playback_guard = self.playback_state.lock().unwrap();
if let Some(state) = playback_guard.take() {
let _ = state.cmd_tx.send(());
let _ = state.thread.join();
}
Ok(())
}
}
fn run_capture(cmd_rx: pw::channel::Receiver<()>, tx: Sender<Vec<i16>>) -> Result<(), AudioError> {
let mainloop = pw::main_loop::MainLoopRc::new(None)
.map_err(|e| AudioError::Init(e.to_string()))?;
let context = pw::context::ContextRc::new(&mainloop, None)
.map_err(|e| AudioError::Init(e.to_string()))?;
let core = context.connect_rc(None)
.map_err(|e| AudioError::Init(e.to_string()))?;
// Ring buffer setup: 9600 samples (200ms capacity for mono 48kHz)
let rb = HeapRb::<i16>::new(9600);
let (producer, mut consumer) = rb.split();
// Command receiver to quit main loop
let mainloop_clone = mainloop.clone();
let _cmd_recv = cmd_rx.attach(mainloop.loop_(), move |_| {
mainloop_clone.quit();
});
let props = properties! {
*pw::keys::MEDIA_TYPE => "Audio",
*pw::keys::MEDIA_CATEGORY => "Capture",
*pw::keys::MEDIA_ROLE => "Communication",
};
let stream = pw::stream::StreamBox::new(&core, "peerspeak-capture-stream", props)
.map_err(|e| AudioError::Init(e.to_string()))?;
struct CaptureUserData<P: ringbuf::traits::Producer<Item = i16>> {
producer: P,
}
let _listener = stream
.add_local_listener_with_user_data(CaptureUserData { producer })
.process(|stream, user_data| {
if let Some(mut buffer) = stream.dequeue_buffer() {
let datas = buffer.datas_mut();
if !datas.is_empty() {
let data = &mut datas[0];
let size = data.chunk().size() as usize;
if let Some(slice) = data.data() {
// Each sample is 2 bytes (S16LE)
for chunk in slice[..size].chunks_exact(2) {
let sample = i16::from_le_bytes([chunk[0], chunk[1]]);
let _ = user_data.producer.try_push(sample);
}
}
}
}
})
.register()
.map_err(|e| AudioError::Init(e.to_string()))?;
let mut audio_info = spa::param::audio::AudioInfoRaw::new();
audio_info.set_format(spa::param::audio::AudioFormat::S16LE);
audio_info.set_rate(48000);
audio_info.set_channels(1); // Mono
let obj = pw::spa::pod::Object {
type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
id: pw::spa::param::ParamType::EnumFormat.as_raw(),
properties: audio_info.into(),
};
let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
std::io::Cursor::new(Vec::new()),
&pw::spa::pod::Value::Object(obj),
)
.unwrap()
.0
.into_inner();
let mut params = [Pod::from_bytes(&values).unwrap()];
stream.connect(
spa::utils::Direction::Input,
None,
pw::stream::StreamFlags::AUTOCONNECT
| pw::stream::StreamFlags::MAP_BUFFERS
| pw::stream::StreamFlags::RT_PROCESS,
&mut params,
)
.map_err(|e| AudioError::Stream(e.to_string()))?;
// Spawn the worker thread to pop from consumer and send Vec<i16> frames
let running = Arc::new(AtomicBool::new(true));
let running_clone = running.clone();
let worker_handle = thread::spawn(move || {
let mut frame = Vec::with_capacity(960);
while running_clone.load(Ordering::Relaxed) {
let mut popped = false;
while let Some(sample) = consumer.try_pop() {
popped = true;
frame.push(sample);
if frame.len() == 960 {
if tx.send(frame).is_err() {
return;
}
frame = Vec::with_capacity(960);
}
}
if !popped {
thread::sleep(Duration::from_millis(5));
}
}
});
mainloop.run();
running.store(false, Ordering::Relaxed);
let _ = worker_handle.join();
Ok(())
}
fn run_playback(cmd_rx: pw::channel::Receiver<()>, rx: Receiver<Vec<i16>>) -> Result<(), AudioError> {
let mainloop = pw::main_loop::MainLoopRc::new(None)
.map_err(|e| AudioError::Init(e.to_string()))?;
let context = pw::context::ContextRc::new(&mainloop, None)
.map_err(|e| AudioError::Init(e.to_string()))?;
let core = context.connect_rc(None)
.map_err(|e| AudioError::Init(e.to_string()))?;
// Ring buffer setup: 9600 samples (200ms capacity for mono 48kHz)
let rb = HeapRb::<i16>::new(9600);
let (mut producer, consumer) = rb.split();
// Command receiver to quit main loop
let mainloop_clone = mainloop.clone();
let _cmd_recv = cmd_rx.attach(mainloop.loop_(), move |_| {
mainloop_clone.quit();
});
let props = properties! {
*pw::keys::MEDIA_TYPE => "Audio",
*pw::keys::MEDIA_CATEGORY => "Playback",
*pw::keys::MEDIA_ROLE => "Communication",
};
let stream = pw::stream::StreamBox::new(&core, "peerspeak-playback-stream", props)
.map_err(|e| AudioError::Init(e.to_string()))?;
struct PlaybackUserData<C: ringbuf::traits::Consumer<Item = i16>> {
consumer: C,
}
let _listener = stream
.add_local_listener_with_user_data(PlaybackUserData { consumer })
.process(|stream, user_data| {
if let Some(mut buffer) = stream.dequeue_buffer() {
let datas = buffer.datas_mut();
if !datas.is_empty() {
let data = &mut datas[0];
let mut total_size = 0;
if let Some(slice) = data.data() {
let stride = 2; // S16LE Mono = 2 bytes per frame
let n_frames = slice.len() / stride;
for i in 0..n_frames {
let val = user_data.consumer.try_pop().unwrap_or(0);
let bytes = val.to_le_bytes();
let start = i * stride;
slice[start] = bytes[0];
slice[start + 1] = bytes[1];
}
total_size = n_frames * stride;
}
let chunk = data.chunk_mut();
*chunk.offset_mut() = 0;
*chunk.stride_mut() = 2;
*chunk.size_mut() = total_size as _;
}
}
})
.register()
.map_err(|e| AudioError::Init(e.to_string()))?;
let mut audio_info = spa::param::audio::AudioInfoRaw::new();
audio_info.set_format(spa::param::audio::AudioFormat::S16LE);
audio_info.set_rate(48000);
audio_info.set_channels(1); // Mono
let obj = pw::spa::pod::Object {
type_: pw::spa::utils::SpaTypes::ObjectParamFormat.as_raw(),
id: pw::spa::param::ParamType::EnumFormat.as_raw(),
properties: audio_info.into(),
};
let values: Vec<u8> = pw::spa::pod::serialize::PodSerializer::serialize(
std::io::Cursor::new(Vec::new()),
&pw::spa::pod::Value::Object(obj),
)
.unwrap()
.0
.into_inner();
let mut params = [Pod::from_bytes(&values).unwrap()];
stream.connect(
spa::utils::Direction::Output,
None,
pw::stream::StreamFlags::AUTOCONNECT
| pw::stream::StreamFlags::MAP_BUFFERS
| pw::stream::StreamFlags::RT_PROCESS,
&mut params,
)
.map_err(|e| AudioError::Stream(e.to_string()))?;
// Spawn a worker thread to read from rx and push to producer
let running = Arc::new(AtomicBool::new(true));
let running_clone = running.clone();
let worker_handle = thread::spawn(move || {
while running_clone.load(Ordering::Relaxed) {
if let Ok(frame) = rx.recv() {
for &sample in &frame {
// Try to push. If buffer is full, drop to avoid growing latency.
if producer.try_push(sample).is_err() {
break;
}
}
} else {
return;
}
}
});
mainloop.run();
running.store(false, Ordering::Relaxed);
let _ = worker_handle.join();
Ok(())
}
src/codec/mod.rs
+25
View File
@@ -0,0 +1,25 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum CodecError {
#[error("Failed to initialize codec: {0}")]
Init(String),
#[error("Encoding failed: {0}")]
Encode(String),
#[error("Decoding failed: {0}")]
Decode(String),
}
pub trait AudioEncoder: Send {
/// Encodes raw PCM samples into compressed bytes.
fn encode(&mut self, pcm: &[i16]) -> Result<Vec<u8>, CodecError>;
}
pub trait AudioDecoder: Send {
/// Decodes compressed bytes back into raw PCM samples.
/// If `compressed` is `None` (or `Some(&[])`), it indicates packet loss,
/// enabling the decoder to perform packet loss concealment (PLC).
fn decode(&mut self, compressed: Option<&[u8]>) -> Result<Vec<i16>, CodecError>;
}
pub mod opus_impl;
src/codec/opus_impl.rs
+74
View File
@@ -0,0 +1,74 @@
use crate::codec::{AudioEncoder, AudioDecoder, CodecError};
use opus::{Encoder, Decoder, Application, Channels};
pub struct OpusEncoder {
encoder: Encoder,
}
impl OpusEncoder {
/// Creates a new Opus encoder.
/// Standard voice parameters: sample_rate = 48000, channels = Channels::Mono, application = Application::Voip
pub fn new(sample_rate: u32, channels: Channels, application: Application) -> Result<Self, CodecError> {
let encoder = Encoder::new(sample_rate, channels, application)
.map_err(|e| CodecError::Init(format!("Failed to create Opus encoder: {}", e)))?;
Ok(Self { encoder })
}
}
impl AudioEncoder for OpusEncoder {
fn encode(&mut self, pcm: &[i16]) -> Result<Vec<u8>, CodecError> {
// We allocate a buffer for the compressed output.
// A maximum packet size of 4000 bytes is more than enough for a single voice frame.
let mut compressed = vec![0u8; 4000];
let len = self.encoder.encode(pcm, &mut compressed)
.map_err(|e| CodecError::Encode(format!("Opus encoding failed: {}", e)))?;
compressed.truncate(len);
Ok(compressed)
}
}
pub struct OpusDecoder {
decoder: Decoder,
channels: Channels,
}
impl OpusDecoder {
/// Creates a new Opus decoder.
/// Standard voice parameters: sample_rate = 48000, channels = Channels::Mono
pub fn new(sample_rate: u32, channels: Channels) -> Result<Self, CodecError> {
let decoder = Decoder::new(sample_rate, channels)
.map_err(|e| CodecError::Init(format!("Failed to create Opus decoder: {}", e)))?;
Ok(Self { decoder, channels })
}
}
impl AudioDecoder for OpusDecoder {
fn decode(&mut self, compressed: Option<&[u8]>) -> Result<Vec<i16>, CodecError> {
// Maximum Opus frame size is 120ms. At 48kHz, this is 5760 samples per channel.
let channels_count = match self.channels {
Channels::Mono => 1,
Channels::Stereo => 2,
};
let max_samples = 5760 * channels_count;
let mut pcm = vec![0i16; max_samples];
let decoded_samples_per_channel = match compressed {
Some(data) if !data.is_empty() => {
// Normal decode
self.decoder.decode(data, &mut pcm, false)
.map_err(|e| CodecError::Decode(format!("Opus decoding failed: {}", e)))?
}
_ => {
// Packet Loss Concealment (PLC)
// In opus-rs, passing an empty slice triggers PLC.
self.decoder.decode(&[], &mut pcm, false)
.map_err(|e| CodecError::Decode(format!("Opus PLC decoding failed: {}", e)))?
}
};
let total_samples = decoded_samples_per_channel * channels_count;
pcm.truncate(total_samples);
Ok(pcm)
}
}
src/core/messages.rs
+21
View File
@@ -0,0 +1,21 @@
use crate::network::PeerState;
use iroh::EndpointId;
#[derive(Debug, Clone)]
pub enum CoreCommand {
Join { name: String, ticket: String },
Leave,
ToggleMute,
ToggleDeafen,
}
#[derive(Debug, Clone)]
pub enum UiEvent {
RoomJoined { ticket: String, self_id: String },
RoomLeft,
PeerJoined { id: EndpointId, state: PeerState },
PeerLeft { id: EndpointId },
PeerUpdated { id: EndpointId, state: PeerState },
AudioLevels(Vec<(EndpointId, f32)>),
Error(String),
}
src/core/mod.rs
+382
View File
@@ -0,0 +1,382 @@
pub mod messages;
use crate::audio::{AudioBackend, pipewire_impl::PipeWireBackend};
use crate::codec::{AudioEncoder, AudioDecoder, opus_impl::{OpusEncoder, OpusDecoder}};
use crate::network::{
NetworkTransport, RoomState, PeerState, RoomEvent, PeerSpeakTicket,
iroh_impl::IrohTransport,
gossip::IrohGossipState,
};
use crate::core::messages::{CoreCommand, UiEvent};
use iroh::{Endpoint, EndpointId, endpoint::presets, protocol::Router};
use iroh_gossip::net::Gossip;
use tokio::sync::{mpsc, Mutex};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
pub struct CoreController {
cmd_tx: mpsc::Sender<CoreCommand>,
}
impl CoreController {
pub fn new(ui_tx: mpsc::Sender<UiEvent>) -> Self {
let (cmd_tx, cmd_rx) = mpsc::channel(100);
tokio::spawn(async move {
if let Err(e) = run_core_loop(cmd_rx, ui_tx).await {
eprintln!("App core loop failed: {:?}", e);
}
});
Self { cmd_tx }
}
pub fn send(&self, cmd: CoreCommand) -> Result<(), mpsc::error::TrySendError<CoreCommand>> {
self.cmd_tx.try_send(cmd)
}
}
struct ActiveSession {
router: Router,
room_state: Arc<IrohGossipState>,
capture_thread: std::thread::JoinHandle<()>,
datagram_task: tokio::task::JoinHandle<()>,
mixer_task: tokio::task::JoinHandle<()>,
event_task: tokio::task::JoinHandle<()>,
}
impl ActiveSession {
async fn shutdown(self, audio_backend: Arc<PipeWireBackend>) {
// Abort asynchronous tasks first
self.datagram_task.abort();
self.mixer_task.abort();
self.event_task.abort();
// Stop the audio backend in a blocking thread to avoid blocking the async executor
let audio_backend_clone = audio_backend.clone();
let _ = tokio::task::spawn_blocking(move || {
let _ = audio_backend_clone.stop();
}).await;
// Leave room
let _ = self.room_state.leave().await;
// Shut down router
let _ = self.router.shutdown().await;
// Clean up capture thread
let _ = self.capture_thread.join();
}
}
async fn run_core_loop(
mut cmd_rx: mpsc::Receiver<CoreCommand>,
ui_tx: mpsc::Sender<UiEvent>,
) -> Result<(), anyhow::Error> {
let memory_lookup = iroh::address_lookup::memory::MemoryLookup::new();
let secret_key = iroh::SecretKey::generate();
// Bind local endpoint on a random port
let endpoint = Endpoint::builder(presets::N0)
.secret_key(secret_key)
.address_lookup(memory_lookup.clone())
.bind()
.await?;
endpoint.online().await;
let audio_backend = Arc::new(PipeWireBackend::new());
let is_muted = Arc::new(AtomicBool::new(false));
let is_deafened = Arc::new(AtomicBool::new(false));
let mut current_name = "Anonymous".to_string();
let mut active_session: Option<ActiveSession> = None;
while let Some(cmd) = cmd_rx.recv().await {
match cmd {
CoreCommand::Join { name, ticket } => {
current_name = name.clone();
// Clean up any existing session
if let Some(session) = active_session.take() {
session.shutdown(audio_backend.clone()).await;
}
// Determine target ticket
let ticket_str = if ticket.trim().is_empty() || ticket == "create" {
let topic_id: [u8; 32] = rand::random();
let host_addr = endpoint.addr();
let ticket = PeerSpeakTicket { host_addr, topic_id };
ticket.to_string()
} else {
ticket.trim().to_string()
};
// Initialize Gossip and Transport
let gossip = Gossip::builder().spawn(endpoint.clone());
let (transport, audio_proto) = IrohTransport::new(endpoint.clone());
let transport = Arc::new(transport);
// Start Router
let router = iroh::protocol::Router::builder(endpoint.clone())
.accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone())
.accept(b"peerspeak-audio", audio_proto)
.spawn();
let room_state = Arc::new(IrohGossipState::new(
endpoint.clone(),
gossip.clone(),
memory_lookup.clone(),
));
let self_state = PeerState {
name: current_name.clone(),
is_muted: is_muted.load(Ordering::Relaxed),
addr: endpoint.addr(),
};
if let Err(e) = room_state.join(&ticket_str, self_state.clone()).await {
let _ = ui_tx.send(UiEvent::Error(format!("Failed to join room: {}", e))).await;
let _ = router.shutdown().await;
continue;
}
// Setup raw audio channels
let (capture_tx, capture_rx) = std::sync::mpsc::channel();
let (playback_tx, playback_rx) = std::sync::mpsc::channel();
if let Err(e) = audio_backend.start_capture(capture_tx) {
let _ = ui_tx.send(UiEvent::Error(format!("Failed to start capture: {}", e))).await;
let _ = room_state.leave().await;
let _ = router.shutdown().await;
continue;
}
if let Err(e) = audio_backend.start_playback(playback_rx) {
let _ = ui_tx.send(UiEvent::Error(format!("Failed to start playback: {}", e))).await;
let _ = audio_backend.stop();
let _ = room_state.leave().await;
let _ = router.shutdown().await;
continue;
}
let queues: Arc<Mutex<HashMap<EndpointId, VecDeque<i16>>>> = Arc::new(Mutex::new(HashMap::new()));
// 1. Capture & encoding thread
let is_muted_clone = is_muted.clone();
let transport_clone = transport.clone();
let room_state_clone = room_state.clone();
let tokio_handle = tokio::runtime::Handle::current();
let capture_thread = std::thread::spawn(move || {
use opus::{Channels, Application};
let mut encoder = match OpusEncoder::new(48000, Channels::Mono, Application::Voip) {
Ok(enc) => enc,
Err(e) => {
eprintln!("Capture thread error: {:?}", e);
return;
}
};
while let Ok(pcm) = capture_rx.recv() {
if is_muted_clone.load(Ordering::Relaxed) {
continue;
}
if let Ok(encoded) = encoder.encode(&pcm) {
let bytes = bytes::Bytes::from(encoded);
let active = room_state_clone.active_peers();
for (peer_id, _) in active {
let transport = transport_clone.clone();
let bytes = bytes.clone();
tokio_handle.spawn(async move {
let _ = transport.send_datagram(peer_id, bytes).await;
});
}
}
}
});
// 2. Receiver & decoding task
let transport_recv = transport.clone();
let queues_recv = queues.clone();
let datagram_task = tokio::spawn(async move {
use opus::Channels;
let mut datagram_rx = match transport_recv.receive_datagrams().await {
Ok(rx) => rx,
Err(e) => {
eprintln!("Receiver task error: {:?}", e);
return;
}
};
let mut decoders: HashMap<EndpointId, OpusDecoder> = HashMap::new();
while let Some((from_peer, bytes)) = datagram_rx.recv().await {
let decoder = match decoders.entry(from_peer) {
std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
std::collections::hash_map::Entry::Vacant(entry) => {
match OpusDecoder::new(48000, Channels::Mono) {
Ok(dec) => entry.insert(dec),
Err(e) => {
eprintln!("Failed to initialize decoder for {:?}: {:?}", from_peer, e);
continue;
}
}
}
};
match decoder.decode(Some(&bytes)) {
Ok(pcm) => {
let mut guard = queues_recv.lock().await;
let queue = guard.entry(from_peer).or_insert_with(VecDeque::new);
queue.extend(pcm);
}
Err(e) => {
eprintln!("Failed to decode packet from {:?}: {:?}", from_peer, e);
}
}
}
});
// 3. Mixing & level extraction loop task
let queues_mixer = queues.clone();
let is_deafened_clone = is_deafened.clone();
let ui_tx_mixer = ui_tx.clone();
let mixer_task = tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_millis(20));
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
interval.tick().await;
let mut guard = queues_mixer.lock().await;
let mut mixed = vec![0i16; 960];
let mut active_levels = Vec::new();
let mut peer_frames = Vec::new();
for (&peer_id, queue) in guard.iter_mut() {
let mut frame = vec![0i16; 960];
let len = queue.len();
if len >= 960 {
if len > 9600 {
let drain = len - 960;
queue.drain(0..drain);
}
for sample in frame.iter_mut() {
*sample = queue.pop_front().unwrap_or(0);
}
} else {
for sample in frame.iter_mut().take(len) {
*sample = queue.pop_front().unwrap_or(0);
}
}
// Calculate speaking level (RMS normalized)
let sum_sq: f32 = frame.iter().map(|&x| (x as f32).powi(2)).sum();
let rms = (sum_sq / 960.0).sqrt();
let level = (rms / 32768.0).clamp(0.0, 1.0);
active_levels.push((peer_id, level));
peer_frames.push(frame);
}
if !peer_frames.is_empty() {
for i in 0..960 {
let mut sum = 0i32;
for f in &peer_frames {
sum += f[i] as i32;
}
mixed[i] = sum.clamp(i16::MIN as i32, i16::MAX as i32) as i16;
}
}
let frame_to_send = if is_deafened_clone.load(Ordering::Relaxed) {
vec![0i16; 960]
} else {
mixed
};
if playback_tx.send(frame_to_send).is_err() {
break;
}
let _ = ui_tx_mixer.send(UiEvent::AudioLevels(active_levels)).await;
}
});
// 4. Room event subscriber task
let mut room_events = match room_state.subscribe_events().await {
Ok(rx) => rx,
Err(e) => {
let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe events: {}", e))).await;
continue;
}
};
let ui_tx_events = ui_tx.clone();
let queues_events = queues.clone();
let event_task = tokio::spawn(async move {
while let Some(event) = room_events.recv().await {
match event {
RoomEvent::PeerJoined(peer_id, state) => {
queues_events.lock().await.entry(peer_id).or_insert_with(VecDeque::new);
let _ = ui_tx_events.send(UiEvent::PeerJoined { id: peer_id, state }).await;
}
RoomEvent::PeerLeft(peer_id) => {
queues_events.lock().await.remove(&peer_id);
let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await;
}
RoomEvent::PeerUpdated(peer_id, state) => {
let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await;
}
}
}
});
let session = ActiveSession {
router,
room_state: room_state.clone(),
capture_thread,
datagram_task,
mixer_task,
event_task,
};
let self_id = endpoint.id().to_string();
let _ = ui_tx.send(UiEvent::RoomJoined { ticket: ticket_str, self_id }).await;
active_session = Some(session);
}
CoreCommand::Leave => {
if let Some(session) = active_session.take() {
session.shutdown(audio_backend.clone()).await;
let _ = ui_tx.send(UiEvent::RoomLeft).await;
}
}
CoreCommand::ToggleMute => {
let current = is_muted.load(Ordering::Relaxed);
let new_state = !current;
is_muted.store(new_state, Ordering::Relaxed);
if let Some(session) = &active_session {
let self_state = PeerState {
name: current_name.clone(),
is_muted: new_state,
addr: endpoint.addr(),
};
let _ = session.room_state.update_self_state(self_state).await;
}
}
CoreCommand::ToggleDeafen => {
let current = is_deafened.load(Ordering::Relaxed);
is_deafened.store(!current, Ordering::Relaxed);
}
}
}
Ok(())
}
src/main.rs
+12
View File
@@ -0,0 +1,12 @@
pub mod audio;
pub mod codec;
pub mod network;
pub mod core;
pub mod app;
#[tokio::main]
async fn main() {
if let Err(e) = app::run_gui() {
eprintln!("Error running GUI: {:?}", e);
}
}
src/network/gossip.rs
+214
View File
@@ -0,0 +1,214 @@
use crate::network::{RoomState, NetError, PeerState, RoomEvent, PeerSpeakTicket};
use iroh::{Endpoint, EndpointId};
use iroh_gossip::net::Gossip;
use iroh_gossip::proto::TopicId;
use tokio::sync::{mpsc, Mutex};
use tokio::sync::mpsc::Receiver;
use std::sync::Arc;
use std::collections::HashMap;
use async_trait::async_trait;
use tokio_stream::StreamExt;
use serde::{Serialize, Deserialize};
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct GossipPayload {
pub author: EndpointId,
pub msg: GossipMessage,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub enum GossipMessage {
Announce(PeerState),
Leave,
}
pub struct IrohGossipState {
_endpoint: Endpoint,
gossip: Gossip,
address_lookup: iroh::address_lookup::memory::MemoryLookup,
self_state: Arc<Mutex<Option<PeerState>>>,
peers: Arc<Mutex<HashMap<EndpointId, PeerState>>>,
event_tx: mpsc::Sender<RoomEvent>,
event_rx: Mutex<Option<mpsc::Receiver<RoomEvent>>>,
active_topic: Mutex<Option<tokio::task::JoinHandle<()>>>,
active_topic_id: Mutex<Option<TopicId>>,
active_sender: Mutex<Option<iroh_gossip::api::GossipSender>>,
}
impl IrohGossipState {
pub fn new(
endpoint: Endpoint,
gossip: Gossip,
address_lookup: iroh::address_lookup::memory::MemoryLookup,
) -> Self {
let (event_tx, event_rx) = mpsc::channel(100);
Self {
_endpoint: endpoint,
gossip,
address_lookup,
self_state: Arc::new(Mutex::new(None)),
peers: Arc::new(Mutex::new(HashMap::new())),
event_tx,
event_rx: Mutex::new(Some(event_rx)),
active_topic: Mutex::new(None),
active_topic_id: Mutex::new(None),
active_sender: Mutex::new(None),
}
}
}
#[async_trait]
impl RoomState for IrohGossipState {
async fn join(&self, ticket_str: &str, self_state: PeerState) -> Result<(), NetError> {
let ticket = ticket_str.parse::<PeerSpeakTicket>()?;
let topic_id = TopicId::from_bytes(ticket.topic_id);
// Stop any currently running topic
let _ = self.leave().await;
// Add the host to the address book
self.address_lookup.add_endpoint_info(ticket.host_addr.clone());
// Join the gossip topic. If we are the host, bootstrap list will be empty
// or contain ourselves (which is fine), but let's bootstrap to the ticket host.
let bootstrap_peers = if ticket.host_addr.id == self_state.addr.id {
vec![]
} else {
vec![ticket.host_addr.id]
};
let gossip_topic = self.gossip.subscribe(topic_id, bootstrap_peers).await
.map_err(|e| NetError::Gossip(format!("Failed to join gossip topic: {}", e)))?;
let (gossip_sender, mut gossip_receiver) = gossip_topic.split();
*self.self_state.lock().await = Some(self_state.clone());
*self.active_topic_id.lock().await = Some(topic_id);
*self.active_sender.lock().await = Some(gossip_sender.clone());
let event_tx = self.event_tx.clone();
let peers = self.peers.clone();
let address_lookup = self.address_lookup.clone();
let self_state_clone = self.self_state.clone();
let gossip_sender_clone = gossip_sender.clone();
let handle = tokio::spawn(async move {
// Broadcast initial state
let payload = GossipPayload {
author: self_state_clone.lock().await.as_ref().unwrap().addr.id,
msg: GossipMessage::Announce(self_state_clone.lock().await.clone().unwrap()),
};
if let Ok(bytes) = serde_json::to_vec(&payload) {
let _ = gossip_sender_clone.broadcast(bytes.into()).await;
}
// Stream topic messages
while let Some(res) = gossip_receiver.next().await {
match res {
Ok(iroh_gossip::api::Event::Received(msg)) => {
if let Ok(payload) = serde_json::from_slice::<GossipPayload>(&msg.content) {
match payload.msg {
GossipMessage::Announce(state) => {
if payload.author == self_state_clone.lock().await.as_ref().unwrap().addr.id {
continue; // Ignore our own announcements
}
let mut peer_map = peers.lock().await;
let is_new = !peer_map.contains_key(&payload.author);
let state_changed = peer_map.get(&payload.author) != Some(&state);
if is_new {
address_lookup.add_endpoint_info(state.addr.clone());
peer_map.insert(payload.author, state.clone());
let _ = event_tx.send(RoomEvent::PeerJoined(payload.author, state)).await;
} else if state_changed {
peer_map.insert(payload.author, state.clone());
let _ = event_tx.send(RoomEvent::PeerUpdated(payload.author, state)).await;
}
}
GossipMessage::Leave => {
let mut peer_map = peers.lock().await;
if peer_map.remove(&payload.author).is_some() {
let _ = event_tx.send(RoomEvent::PeerLeft(payload.author)).await;
}
}
}
}
}
Ok(iroh_gossip::api::Event::NeighborUp(_peer_id)) => {
// Resend state on new neighbor connection to guarantee synchronization
if let Some(state) = self_state_clone.lock().await.as_ref() {
let payload = GossipPayload {
author: state.addr.id,
msg: GossipMessage::Announce(state.clone()),
};
if let Ok(bytes) = serde_json::to_vec(&payload) {
let _ = gossip_sender_clone.broadcast(bytes.into()).await;
}
}
}
_ => {}
}
}
});
*self.active_topic.lock().await = Some(handle);
Ok(())
}
async fn update_self_state(&self, self_state: PeerState) -> Result<(), NetError> {
let mut self_guard = self.self_state.lock().await;
*self_guard = Some(self_state.clone());
if let Some(sender) = self.active_sender.lock().await.as_ref() {
let payload = GossipPayload {
author: self_state.addr.id,
msg: GossipMessage::Announce(self_state),
};
if let Ok(bytes) = serde_json::to_vec(&payload) {
sender.broadcast(bytes.into()).await
.map_err(|e| NetError::Gossip(e.to_string()))?;
}
}
Ok(())
}
async fn leave(&self) -> Result<(), NetError> {
let mut handle_guard = self.active_topic.lock().await;
if let Some(handle) = handle_guard.take() {
handle.abort();
}
let mut topic_id_guard = self.active_topic_id.lock().await;
let _ = topic_id_guard.take();
let mut sender_guard = self.active_sender.lock().await;
if let Some(sender) = sender_guard.take() {
if let Some(self_state) = self.self_state.lock().await.as_ref() {
let payload = GossipPayload {
author: self_state.addr.id,
msg: GossipMessage::Leave,
};
if let Ok(bytes) = serde_json::to_vec(&payload) {
let _ = sender.broadcast(bytes.into()).await;
}
}
}
self.peers.lock().await.clear();
Ok(())
}
fn active_peers(&self) -> Vec<(EndpointId, PeerState)> {
let guard = self.peers.blocking_lock();
guard.iter().map(|(k, v)| (*k, v.clone())).collect()
}
async fn subscribe_events(&self) -> Result<Receiver<RoomEvent>, NetError> {
let mut rx_guard = self.event_rx.lock().await;
if let Some(rx) = rx_guard.take() {
Ok(rx)
} else {
Err(NetError::Other("Events already subscribed".to_string()))
}
}
}
src/network/iroh_impl.rs
+125
View File
@@ -0,0 +1,125 @@
use crate::network::{NetworkTransport, NetError};
use iroh::{Endpoint, EndpointId};
use iroh::endpoint::Connection;
use bytes::Bytes;
use tokio::sync::{mpsc, Mutex};
use tokio::sync::mpsc::Receiver;
use std::sync::Arc;
use std::collections::HashMap;
use async_trait::async_trait;
#[derive(Debug, Clone)]
pub struct AudioProtocol {
incoming_tx: mpsc::Sender<(EndpointId, Bytes)>,
connections: Arc<Mutex<HashMap<EndpointId, Connection>>>,
}
impl iroh::protocol::ProtocolHandler for AudioProtocol {
fn accept(
&self,
connection: Connection,
) -> impl std::future::Future<Output = Result<(), iroh::protocol::AcceptError>> + Send {
let peer_id = connection.remote_id();
let incoming_tx = self.incoming_tx.clone();
let connections = self.connections.clone();
async move {
connections.lock().await.insert(peer_id, connection.clone());
loop {
match connection.read_datagram().await {
Ok(bytes) => {
if incoming_tx.send((peer_id, bytes)).await.is_err() {
break;
}
}
Err(_) => {
connections.lock().await.remove(&peer_id);
break;
}
}
}
Ok(())
}
}
}
pub struct IrohTransport {
endpoint: Endpoint,
connections: Arc<Mutex<HashMap<EndpointId, Connection>>>,
incoming_tx: mpsc::Sender<(EndpointId, Bytes)>,
incoming_rx: Mutex<Option<mpsc::Receiver<(EndpointId, Bytes)>>>,
}
impl IrohTransport {
pub fn new(endpoint: Endpoint) -> (Self, AudioProtocol) {
let (incoming_tx, incoming_rx) = mpsc::channel(1000);
let connections = Arc::new(Mutex::new(HashMap::new()));
let audio_proto = AudioProtocol {
incoming_tx: incoming_tx.clone(),
connections: connections.clone(),
};
let transport = Self {
endpoint,
connections,
incoming_tx,
incoming_rx: Mutex::new(Some(incoming_rx)),
};
(transport, audio_proto)
}
}
#[async_trait]
impl NetworkTransport for IrohTransport {
async fn send_datagram(&self, peer_id: EndpointId, data: Bytes) -> Result<(), NetError> {
let mut conns = self.connections.lock().await;
let conn = if let Some(conn) = conns.get(&peer_id) {
conn.clone()
} else {
// Establish a new connection.
// We use the same audio ALPN: b"peerspeak-audio"
let alpn = b"peerspeak-audio";
let conn = self.endpoint.connect(peer_id, alpn).await
.map_err(|e| NetError::Connection(e.to_string()))?;
conns.insert(peer_id, conn.clone());
let incoming_tx_inner = self.incoming_tx.clone();
let connections_inner = self.connections.clone();
let conn_clone = conn.clone();
tokio::spawn(async move {
loop {
match conn_clone.read_datagram().await {
Ok(bytes) => {
if let Err(_) = incoming_tx_inner.send((peer_id, bytes)).await {
break;
}
}
Err(_) => {
connections_inner.lock().await.remove(&peer_id);
break;
}
}
}
});
conn
};
conn.send_datagram(data)
.map_err(|e| NetError::Connection(e.to_string()))?;
Ok(())
}
async fn receive_datagrams(&self) -> Result<Receiver<(EndpointId, Bytes)>, NetError> {
let mut rx_guard = self.incoming_rx.lock().await;
if let Some(rx) = rx_guard.take() {
Ok(rx)
} else {
Err(NetError::Other("Datagram receiver already subscribed".to_string()))
}
}
}
src/network/mod.rs
+93
View File
@@ -0,0 +1,93 @@
use iroh::EndpointId;
use bytes::Bytes;
use thiserror::Error;
use tokio::sync::mpsc::Receiver;
use async_trait::async_trait;
use serde::{Serialize, Deserialize};
use std::str::FromStr;
#[derive(Error, Debug)]
pub enum NetError {
#[error("Failed to initialize network: {0}")]
Init(String),
#[error("Failed to connect/dial peer: {0}")]
Connection(String),
#[error("Gossip swarm error: {0}")]
Gossip(String),
#[error("Serialization / Deserialization error: {0}")]
Serde(String),
#[error("Invalid ticket: {0}")]
InvalidTicket(String),
#[error("Other network error: {0}")]
Other(String),
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PeerState {
pub name: String,
pub is_muted: bool,
pub addr: iroh::EndpointAddr,
}
#[derive(Debug, Clone)]
pub enum RoomEvent {
PeerJoined(EndpointId, PeerState),
PeerLeft(EndpointId),
PeerUpdated(EndpointId, PeerState),
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct PeerSpeakTicket {
pub host_addr: iroh::EndpointAddr,
pub topic_id: [u8; 32],
}
impl ToString for PeerSpeakTicket {
fn to_string(&self) -> String {
let serialized = serde_json::to_vec(self).unwrap();
// Convert to base64 URL-safe string
base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, &serialized)
}
}
impl FromStr for PeerSpeakTicket {
type Err = NetError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let decoded = base64::Engine::decode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, s)
.map_err(|e| NetError::InvalidTicket(e.to_string()))?;
let ticket: PeerSpeakTicket = serde_json::from_slice(&decoded)
.map_err(|e| NetError::InvalidTicket(e.to_string()))?;
Ok(ticket)
}
}
#[async_trait]
pub trait NetworkTransport: Send + Sync {
/// Send a low-latency unreliable datagram to a specific peer (for audio).
async fn send_datagram(&self, peer_id: EndpointId, data: Bytes) -> Result<(), NetError>;
/// Subscribes to incoming datagrams from any peer.
async fn receive_datagrams(&self) -> Result<Receiver<(EndpointId, Bytes)>, NetError>;
}
#[async_trait]
pub trait RoomState: Send + Sync {
/// Joins a room using a gossip ticket string and announces our state.
async fn join(&self, ticket: &str, self_state: PeerState) -> Result<(), NetError>;
/// Updates our local state (e.g. when user mutes/unmutes) and broadcasts it.
async fn update_self_state(&self, self_state: PeerState) -> Result<(), NetError>;
/// Leaves the room and announces departure.
async fn leave(&self) -> Result<(), NetError>;
/// Returns a list of currently active peers in the room.
fn active_peers(&self) -> Vec<(EndpointId, PeerState)>;
/// Subscribes to room events (peer joined, peer left, peer updated).
async fn subscribe_events(&self) -> Result<Receiver<RoomEvent>, NetError>;
}
pub mod iroh_impl;
pub mod gossip;