Skip to repository content1887 lines · 69.8 KB · rust
tenant.openagents/omega
No repository description is available.
OpenAgents Git authority 2026-07-28T06:49:08.059Z Public web read
NIP-34 coordinate
30617:7649603503856e5148d571eac2766b288a8ff1e9e35d380337a1d2b0015b4f92:omegaMaintainersHidden in public view
References2 branches · 1 tag
Read-only clone
git clone https://openagents.com/git/tenant.openagents/omega.gitBrowse files
room.rs
1use crate::{
2 call_settings::CallSettings,
3 participant::{LocalParticipant, RemoteParticipant},
4};
5use anyhow::{Context as _, Result, anyhow};
6use audio::{Audio, Sound};
7use client::{
8 ChannelId, Client, ParticipantIndex, TypedEnvelope, User, UserStore,
9 proto::{self, PeerId},
10};
11use collections::{BTreeMap, HashMap, HashSet};
12use feature_flags::FeatureFlagAppExt;
13use fs::Fs;
14use futures::StreamExt;
15use gpui::{
16 App, AppContext as _, AsyncApp, Context, Entity, EventEmitter, FutureExt as _,
17 ScreenCaptureSource, ScreenCaptureStream, Task, TaskExt, Timeout, WeakEntity,
18};
19use gpui_tokio::Tokio;
20use language::LanguageRegistry;
21use livekit::{LocalTrackPublication, ParticipantIdentity, RoomEvent};
22use livekit_client::{self as livekit, AudioStream, TrackSid};
23use postage::{sink::Sink, stream::Stream, watch};
24use project::{CURRENT_PROJECT_FEATURES, Project};
25use settings::Settings as _;
26use std::sync::atomic::AtomicU64;
27use std::{future::Future, mem, rc::Rc, sync::Arc, time::Duration, time::Instant};
28
29use super::diagnostics::CallDiagnostics;
30use util::{ResultExt, TryFutureExt, paths::PathStyle, post_inc};
31use workspace::ParticipantLocation;
32
33pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30);
34
35#[derive(Clone, Debug, PartialEq, Eq)]
36pub enum Event {
37 RoomJoined {
38 channel_id: Option<ChannelId>,
39 },
40 ParticipantLocationChanged {
41 participant_id: proto::PeerId,
42 },
43 RemoteVideoTracksChanged {
44 participant_id: proto::PeerId,
45 },
46 RemoteVideoTrackUnsubscribed {
47 sid: TrackSid,
48 },
49 RemoteAudioTracksChanged {
50 participant_id: proto::PeerId,
51 },
52 RemoteProjectShared {
53 owner: Arc<User>,
54 project_id: u64,
55 worktree_root_names: Vec<String>,
56 },
57 RemoteProjectUnshared {
58 project_id: u64,
59 },
60 RemoteProjectJoined {
61 project_id: u64,
62 },
63 RemoteProjectInvitationDiscarded {
64 project_id: u64,
65 },
66 RoomLeft {
67 channel_id: Option<ChannelId>,
68 },
69 LocalScreenShareStarted,
70 LocalScreenShareStopped,
71}
72
73pub struct Room {
74 id: u64,
75 channel_id: Option<ChannelId>,
76 live_kit: Option<LiveKitRoom>,
77 diagnostics: Option<Entity<CallDiagnostics>>,
78 status: RoomStatus,
79 shared_projects: HashSet<WeakEntity<Project>>,
80 joined_projects: HashSet<WeakEntity<Project>>,
81 local_participant: LocalParticipant,
82 remote_participants: BTreeMap<u64, RemoteParticipant>,
83 pending_participants: Vec<Arc<User>>,
84 participant_user_ids: HashSet<u64>,
85 pending_call_count: usize,
86 leave_when_empty: bool,
87 client: Arc<Client>,
88 user_store: Entity<UserStore>,
89 follows_by_leader_id_project_id: HashMap<(PeerId, u64), Vec<PeerId>>,
90 client_subscriptions: Vec<client::Subscription>,
91 _subscriptions: Vec<gpui::Subscription>,
92 room_update_completed_tx: watch::Sender<Option<()>>,
93 room_update_completed_rx: watch::Receiver<Option<()>>,
94 pending_room_update: Option<Task<()>>,
95 maintain_connection: Option<Task<Option<()>>>,
96 created: Instant,
97}
98
99impl EventEmitter<Event> for Room {}
100
101impl Room {
102 pub fn channel_id(&self) -> Option<ChannelId> {
103 self.channel_id
104 }
105
106 pub fn is_sharing_project(&self) -> bool {
107 !self.shared_projects.is_empty()
108 }
109
110 pub fn is_connected(&self, _: &App) -> bool {
111 if let Some(live_kit) = self.live_kit.as_ref() {
112 live_kit.room.connection_state() == livekit::ConnectionState::Connected
113 } else {
114 false
115 }
116 }
117
118 fn new(
119 id: u64,
120 channel_id: Option<ChannelId>,
121 livekit_connection_info: Option<proto::LiveKitConnectionInfo>,
122 client: Arc<Client>,
123 user_store: Entity<UserStore>,
124 cx: &mut Context<Self>,
125 ) -> Self {
126 spawn_room_connection(livekit_connection_info, cx);
127
128 let maintain_connection = cx.spawn({
129 let client = client.clone();
130 async move |this, cx| {
131 Self::maintain_connection(this, client.clone(), cx)
132 .log_err()
133 .await
134 }
135 });
136
137 Audio::play_sound(Sound::Joined, cx);
138
139 let (room_update_completed_tx, room_update_completed_rx) = watch::channel();
140
141 Self {
142 id,
143 channel_id,
144 live_kit: None,
145 diagnostics: None,
146 status: RoomStatus::Online,
147 shared_projects: Default::default(),
148 joined_projects: Default::default(),
149 participant_user_ids: Default::default(),
150 local_participant: Default::default(),
151 remote_participants: Default::default(),
152 pending_participants: Default::default(),
153 pending_call_count: 0,
154 client_subscriptions: vec![
155 client.add_message_handler(cx.weak_entity(), Self::handle_room_updated),
156 ],
157 _subscriptions: vec![
158 cx.on_release(Self::released),
159 cx.on_app_quit(Self::app_will_quit),
160 ],
161 leave_when_empty: false,
162 pending_room_update: None,
163 client,
164 user_store,
165 follows_by_leader_id_project_id: Default::default(),
166 maintain_connection: Some(maintain_connection),
167 room_update_completed_tx,
168 room_update_completed_rx,
169 created: cx.background_executor().now(),
170 }
171 }
172
173 pub(crate) fn create(
174 called_user_id: u64,
175 initial_project: Option<Entity<Project>>,
176 client: Arc<Client>,
177 user_store: Entity<UserStore>,
178 cx: &mut App,
179 ) -> Task<Result<Entity<Self>>> {
180 cx.spawn(async move |cx| {
181 let response = client.request(proto::CreateRoom {}).await?;
182 let room_proto = response.room.context("invalid room")?;
183 let room = cx.new(|cx| {
184 let mut room = Self::new(
185 room_proto.id,
186 None,
187 response.live_kit_connection_info,
188 client,
189 user_store,
190 cx,
191 );
192 if let Some(participant) = room_proto.participants.first() {
193 room.local_participant.role = participant.role()
194 }
195 room
196 });
197
198 let initial_project_id = if let Some(initial_project) = initial_project {
199 let initial_project_id = room
200 .update(cx, |room, cx| {
201 room.share_project(initial_project.clone(), cx)
202 })
203 .await?;
204 Some(initial_project_id)
205 } else {
206 None
207 };
208
209 let did_join = room
210 .update(cx, |room, cx| {
211 room.leave_when_empty = true;
212 room.call(called_user_id, initial_project_id, cx)
213 })
214 .await;
215 match did_join {
216 Ok(()) => Ok(room),
217 Err(error) => Err(error.context("room creation failed")),
218 }
219 })
220 }
221
222 pub(crate) async fn join_channel(
223 channel_id: ChannelId,
224 client: Arc<Client>,
225 user_store: Entity<UserStore>,
226 cx: AsyncApp,
227 ) -> Result<Entity<Self>> {
228 Self::from_join_response(
229 client
230 .request(proto::JoinChannel {
231 channel_id: channel_id.0,
232 })
233 .await?,
234 client,
235 user_store,
236 cx,
237 )
238 }
239
240 pub(crate) async fn join(
241 room_id: u64,
242 client: Arc<Client>,
243 user_store: Entity<UserStore>,
244 cx: AsyncApp,
245 ) -> Result<Entity<Self>> {
246 Self::from_join_response(
247 client.request(proto::JoinRoom { id: room_id }).await?,
248 client,
249 user_store,
250 cx,
251 )
252 }
253
254 fn released(&mut self, cx: &mut App) {
255 if self.status.is_online() {
256 self.leave_internal(cx).detach_and_log_err(cx);
257 }
258 }
259
260 fn app_will_quit(&mut self, cx: &mut Context<Self>) -> impl Future<Output = ()> + use<> {
261 let task = if self.status.is_online() {
262 let leave = self.leave_internal(cx);
263 Some(cx.background_spawn(async move {
264 leave.await.log_err();
265 }))
266 } else {
267 None
268 };
269
270 async move {
271 if let Some(task) = task {
272 task.await;
273 }
274 }
275 }
276
277 pub fn mute_on_join(cx: &App) -> bool {
278 CallSettings::get_global(cx).mute_on_join || client::IMPERSONATE_LOGIN.is_some()
279 }
280
281 fn from_join_response(
282 response: proto::JoinRoomResponse,
283 client: Arc<Client>,
284 user_store: Entity<UserStore>,
285 mut cx: AsyncApp,
286 ) -> Result<Entity<Self>> {
287 let room_proto = response.room.context("invalid room")?;
288 let room = cx.new(|cx| {
289 Self::new(
290 room_proto.id,
291 response.channel_id.map(ChannelId),
292 response.live_kit_connection_info,
293 client,
294 user_store,
295 cx,
296 )
297 });
298 room.update(&mut cx, |room, cx| {
299 room.leave_when_empty = room.channel_id.is_none();
300 room.apply_room_update(room_proto, cx)?;
301 anyhow::Ok(())
302 })?;
303 Ok(room)
304 }
305
306 fn should_leave(&self) -> bool {
307 self.leave_when_empty
308 && self.pending_room_update.is_none()
309 && self.pending_participants.is_empty()
310 && self.remote_participants.is_empty()
311 && self.pending_call_count == 0
312 }
313
314 pub(crate) fn leave(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
315 cx.notify();
316 self.emit_video_track_unsubscribed_events(cx);
317 self.leave_internal(cx)
318 }
319
320 fn leave_internal(&mut self, cx: &mut App) -> Task<Result<()>> {
321 if self.status.is_offline() {
322 return Task::ready(Err(anyhow!("room is offline")));
323 }
324
325 log::info!("leaving room");
326 Audio::play_sound(Sound::Leave, cx);
327
328 self.clear_state(cx);
329
330 let leave_room = self.client.request(proto::LeaveRoom {});
331 cx.background_spawn(async move {
332 leave_room.await?;
333 anyhow::Ok(())
334 })
335 }
336
337 pub(crate) fn clear_state(&mut self, cx: &mut App) {
338 for project in self.shared_projects.drain() {
339 if let Some(project) = project.upgrade() {
340 project.update(cx, |project, cx| {
341 project.unshare(cx).log_err();
342 });
343 }
344 }
345 for project in self.joined_projects.drain() {
346 if let Some(project) = project.upgrade() {
347 project.update(cx, |project, cx| {
348 project.disconnected_from_host(cx);
349 project.close(cx);
350 });
351 }
352 }
353
354 self.status = RoomStatus::Offline;
355 self.remote_participants.clear();
356 self.pending_participants.clear();
357 self.participant_user_ids.clear();
358 self.client_subscriptions.clear();
359 self.live_kit.take();
360 self.diagnostics.take();
361 self.pending_room_update.take();
362 self.maintain_connection.take();
363 }
364
365 fn emit_video_track_unsubscribed_events(&self, cx: &mut Context<Self>) {
366 for participant in self.remote_participants.values() {
367 for sid in participant.video_tracks.keys() {
368 cx.emit(Event::RemoteVideoTrackUnsubscribed { sid: sid.clone() });
369 }
370 }
371 }
372
373 async fn maintain_connection(
374 this: WeakEntity<Self>,
375 client: Arc<Client>,
376 cx: &mut AsyncApp,
377 ) -> Result<()> {
378 let mut client_status = client.status();
379 loop {
380 let _ = client_status.try_recv();
381 let is_connected = client_status.borrow().is_connected();
382 // Even if we're initially connected, any future change of the status means we momentarily disconnected.
383 if !is_connected || client_status.next().await.is_some() {
384 log::info!("detected client disconnection");
385
386 this.upgrade()
387 .context("room was dropped")?
388 .update(cx, |this, cx| {
389 this.status = RoomStatus::Rejoining;
390 cx.notify();
391 });
392
393 // Wait for client to re-establish a connection to the server.
394 let executor = cx.background_executor().clone();
395 let client_reconnection = async {
396 let mut remaining_attempts = 3;
397 while remaining_attempts > 0 {
398 if client_status.borrow().is_connected() {
399 log::info!("client reconnected, attempting to rejoin room");
400
401 let Some(this) = this.upgrade() else { break };
402 let task = this.update(cx, |this, cx| this.rejoin(cx));
403 if task.await.log_err().is_some() {
404 return true;
405 } else {
406 remaining_attempts -= 1;
407 }
408 } else if client_status.borrow().is_signed_out() {
409 return false;
410 }
411
412 log::info!(
413 "waiting for client status change, remaining attempts {}",
414 remaining_attempts
415 );
416 client_status.next().await;
417 }
418 false
419 };
420
421 match client_reconnection
422 .with_timeout(RECONNECT_TIMEOUT, &executor)
423 .await
424 {
425 Ok(true) => {
426 log::info!("successfully reconnected to room");
427 // If we successfully joined the room, go back around the loop
428 // waiting for future connection status changes.
429 continue;
430 }
431 Ok(false) => break,
432 Err(Timeout) => {
433 log::info!("room reconnection timeout expired");
434 break;
435 }
436 }
437 }
438 }
439
440 // The client failed to re-establish a connection to the server
441 // or an error occurred while trying to re-join the room. Either way
442 // we leave the room and return an error.
443 if let Some(this) = this.upgrade() {
444 log::info!("reconnection failed, leaving room");
445 this.update(cx, |this, cx| this.leave(cx)).await?;
446 }
447 anyhow::bail!("can't reconnect to room: client failed to re-establish connection");
448 }
449
450 fn rejoin(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
451 let mut projects = HashMap::default();
452 let mut reshared_projects = Vec::new();
453 let mut rejoined_projects = Vec::new();
454 self.shared_projects.retain(|project| {
455 if let Some(handle) = project.upgrade() {
456 let project = handle.read(cx);
457 if let Some(project_id) = project.remote_id() {
458 projects.insert(project_id, handle.clone());
459 reshared_projects.push(proto::UpdateProject {
460 project_id,
461 worktrees: project.worktree_metadata_protos(cx),
462 });
463 return true;
464 }
465 }
466 false
467 });
468 self.joined_projects.retain(|project| {
469 if let Some(handle) = project.upgrade() {
470 let project = handle.read(cx);
471 if let Some(project_id) = project.remote_id() {
472 projects.insert(project_id, handle.clone());
473 let mut worktrees = Vec::new();
474 let mut repositories = Vec::new();
475 for worktree in project.worktrees(cx) {
476 let worktree = worktree.read(cx);
477 worktrees.push(proto::RejoinWorktree {
478 id: worktree.id().to_proto(),
479 scan_id: worktree.completed_scan_id() as u64,
480 });
481 }
482 for (entry_id, repository) in project.repositories(cx) {
483 let repository = repository.read(cx);
484 repositories.push(proto::RejoinRepository {
485 id: entry_id.to_proto(),
486 scan_id: repository.scan_id,
487 });
488 }
489
490 rejoined_projects.push(proto::RejoinProject {
491 id: project_id,
492 worktrees,
493 repositories,
494 });
495 }
496 return true;
497 }
498 false
499 });
500
501 let response = self.client.request_envelope(proto::RejoinRoom {
502 id: self.id,
503 reshared_projects,
504 rejoined_projects,
505 });
506
507 cx.spawn(async move |this, cx| {
508 let response = response.await?;
509 let message_id = response.message_id;
510 let response = response.payload;
511 let room_proto = response.room.context("invalid room")?;
512 this.update(cx, |this, cx| {
513 this.status = RoomStatus::Online;
514 this.apply_room_update(room_proto, cx)?;
515
516 for reshared_project in response.reshared_projects {
517 if let Some(project) = projects.get(&reshared_project.id) {
518 project.update(cx, |project, cx| {
519 project.reshared(reshared_project, cx).log_err();
520 });
521 }
522 }
523
524 for rejoined_project in response.rejoined_projects {
525 if let Some(project) = projects.get(&rejoined_project.id) {
526 project.update(cx, |project, cx| {
527 project.rejoined(rejoined_project, message_id, cx).log_err();
528 });
529 }
530 }
531
532 anyhow::Ok(())
533 })?
534 })
535 }
536
537 pub fn id(&self) -> u64 {
538 self.id
539 }
540
541 pub fn room_id(&self) -> impl Future<Output = Option<String>> + 'static {
542 let room = self.live_kit.as_ref().map(|lk| lk.room.clone());
543 async move {
544 let room = room?;
545 let sid = room.sid().await;
546 let name = room.name();
547 Some(format!("{} (sid: {sid})", name))
548 }
549 }
550
551 pub fn get_stats(&self, cx: &App) -> Task<Option<livekit::SessionStats>> {
552 match self.live_kit.as_ref() {
553 Some(lk) => {
554 let task = lk.room.stats_task(cx);
555 cx.background_executor()
556 .spawn(async move { task.await.ok() })
557 }
558 None => Task::ready(None),
559 }
560 }
561
562 pub fn input_lag(&self) -> Option<Duration> {
563 let us = self
564 .live_kit
565 .as_ref()?
566 .input_lag_us
567 .as_ref()?
568 .load(std::sync::atomic::Ordering::Relaxed);
569 if us > 0 {
570 Some(Duration::from_micros(us))
571 } else {
572 None
573 }
574 }
575
576 pub fn diagnostics(&self) -> Option<&Entity<CallDiagnostics>> {
577 self.diagnostics.as_ref()
578 }
579
580 pub fn connection_quality(&self) -> livekit::ConnectionQuality {
581 self.live_kit
582 .as_ref()
583 .map(|lk| lk.room.local_participant().connection_quality())
584 .unwrap_or(livekit::ConnectionQuality::Lost)
585 }
586
587 pub fn status(&self) -> RoomStatus {
588 self.status
589 }
590
591 pub fn local_participant(&self) -> &LocalParticipant {
592 &self.local_participant
593 }
594
595 pub fn local_participant_user(&self, cx: &App) -> Option<Arc<User>> {
596 self.user_store.read(cx).current_user()
597 }
598
599 pub fn remote_participants(&self) -> &BTreeMap<u64, RemoteParticipant> {
600 &self.remote_participants
601 }
602
603 pub fn remote_participant_for_peer_id(&self, peer_id: PeerId) -> Option<&RemoteParticipant> {
604 self.remote_participants
605 .values()
606 .find(|p| p.peer_id == peer_id)
607 }
608
609 pub fn role_for_user(&self, user_id: u64) -> Option<proto::ChannelRole> {
610 self.remote_participants
611 .get(&user_id)
612 .map(|participant| participant.role)
613 }
614
615 pub fn contains_guests(&self) -> bool {
616 self.local_participant.role == proto::ChannelRole::Guest
617 || self
618 .remote_participants
619 .values()
620 .any(|p| p.role == proto::ChannelRole::Guest)
621 }
622
623 pub fn local_participant_is_admin(&self) -> bool {
624 self.local_participant.role == proto::ChannelRole::Admin
625 }
626
627 pub fn local_participant_is_guest(&self) -> bool {
628 self.local_participant.role == proto::ChannelRole::Guest
629 }
630
631 pub fn set_participant_role(
632 &mut self,
633 user_id: u64,
634 role: proto::ChannelRole,
635 cx: &Context<Self>,
636 ) -> Task<Result<()>> {
637 let client = self.client.clone();
638 let room_id = self.id;
639 let role = role.into();
640 cx.spawn(async move |_, _| {
641 client
642 .request(proto::SetRoomParticipantRole {
643 room_id,
644 user_id,
645 role,
646 })
647 .await
648 .map(|_| ())
649 })
650 }
651
652 pub fn pending_participants(&self) -> &[Arc<User>] {
653 &self.pending_participants
654 }
655
656 pub fn contains_participant(&self, user_id: u64) -> bool {
657 self.participant_user_ids.contains(&user_id)
658 }
659
660 pub fn followers_for(&self, leader_id: PeerId, project_id: u64) -> &[PeerId] {
661 self.follows_by_leader_id_project_id
662 .get(&(leader_id, project_id))
663 .map_or(&[], |v| v.as_slice())
664 }
665
666 /// Returns the most 'active' projects, defined as most people in the project
667 pub fn most_active_project(&self, cx: &App) -> Option<(u64, u64)> {
668 let mut project_hosts_and_guest_counts = HashMap::<u64, (Option<u64>, u32)>::default();
669 for participant in self.remote_participants.values() {
670 match participant.location {
671 ParticipantLocation::SharedProject { project_id } => {
672 project_hosts_and_guest_counts
673 .entry(project_id)
674 .or_default()
675 .1 += 1;
676 }
677 ParticipantLocation::External | ParticipantLocation::UnsharedProject => {}
678 }
679 for project in &participant.projects {
680 project_hosts_and_guest_counts
681 .entry(project.id)
682 .or_default()
683 .0 = Some(participant.user.legacy_id);
684 }
685 }
686
687 if let Some(user) = self.user_store.read(cx).current_user() {
688 for project in &self.local_participant.projects {
689 project_hosts_and_guest_counts
690 .entry(project.id)
691 .or_default()
692 .0 = Some(user.legacy_id);
693 }
694 }
695
696 project_hosts_and_guest_counts
697 .into_iter()
698 .filter_map(|(id, (host, guest_count))| Some((id, host?, guest_count)))
699 .max_by_key(|(_, _, guest_count)| *guest_count)
700 .map(|(id, host, _)| (id, host))
701 }
702
703 async fn handle_room_updated(
704 this: Entity<Self>,
705 envelope: TypedEnvelope<proto::RoomUpdated>,
706 mut cx: AsyncApp,
707 ) -> Result<()> {
708 let room = envelope.payload.room.context("invalid room")?;
709 this.update(&mut cx, |this, cx| this.apply_room_update(room, cx))
710 }
711
712 fn apply_room_update(&mut self, room: proto::Room, cx: &mut Context<Self>) -> Result<()> {
713 log::trace!(
714 "client {:?}. room update: {:?}",
715 self.client.user_id(),
716 &room
717 );
718
719 self.pending_room_update = Some(self.start_room_connection(room, cx));
720
721 cx.notify();
722 Ok(())
723 }
724
725 pub fn room_update_completed(&mut self) -> impl Future<Output = ()> + use<> {
726 let mut done_rx = self.room_update_completed_rx.clone();
727 async move {
728 while let Some(result) = done_rx.next().await {
729 if result.is_some() {
730 break;
731 }
732 }
733 }
734 }
735
736 fn start_room_connection(&self, mut room: proto::Room, cx: &mut Context<Self>) -> Task<()> {
737 // Filter ourselves out from the room's participants.
738 let local_participant_ix = room
739 .participants
740 .iter()
741 .position(|participant| Some(participant.user_id) == self.client.user_id());
742 let local_participant = local_participant_ix.map(|ix| room.participants.swap_remove(ix));
743
744 let pending_participant_user_ids = room
745 .pending_participants
746 .iter()
747 .map(|p| p.user_id)
748 .collect::<Vec<_>>();
749
750 let remote_participant_user_ids = room
751 .participants
752 .iter()
753 .map(|p| p.user_id)
754 .collect::<Vec<_>>();
755
756 let (remote_participants, pending_participants) =
757 self.user_store.update(cx, move |user_store, cx| {
758 (
759 user_store.get_users(remote_participant_user_ids, cx),
760 user_store.get_users(pending_participant_user_ids, cx),
761 )
762 });
763 cx.spawn(async move |this, cx| {
764 let (remote_participants, pending_participants) =
765 futures::join!(remote_participants, pending_participants);
766
767 this.update(cx, |this, cx| {
768 this.participant_user_ids.clear();
769
770 if let Some(participant) = local_participant {
771 let role = participant.role();
772 this.local_participant.projects = participant.projects;
773 if this.local_participant.role != role {
774 this.local_participant.role = role;
775
776 if role == proto::ChannelRole::Guest {
777 for project in mem::take(&mut this.shared_projects) {
778 if let Some(project) = project.upgrade() {
779 this.unshare_project(project, cx).log_err();
780 }
781 }
782 this.local_participant.projects.clear();
783 if let Some(livekit_room) = &mut this.live_kit {
784 livekit_room.stop_publishing(cx);
785 }
786 }
787
788 this.joined_projects.retain(|project| {
789 if let Some(project) = project.upgrade() {
790 project.update(cx, |project, cx| project.set_role(role, cx));
791 true
792 } else {
793 false
794 }
795 });
796 }
797 } else {
798 this.local_participant.projects.clear();
799 }
800
801 let livekit_participants = this
802 .live_kit
803 .as_ref()
804 .map(|live_kit| live_kit.room.remote_participants());
805
806 if let Some(participants) = remote_participants.log_err() {
807 for (participant, user) in room.participants.into_iter().zip(participants) {
808 let Some(peer_id) = participant.peer_id else {
809 continue;
810 };
811 let participant_index = ParticipantIndex(participant.participant_index);
812 this.participant_user_ids.insert(participant.user_id);
813
814 let old_projects = this
815 .remote_participants
816 .get(&participant.user_id)
817 .into_iter()
818 .flat_map(|existing| &existing.projects)
819 .map(|project| project.id)
820 .collect::<HashSet<_>>();
821 let new_projects = participant
822 .projects
823 .iter()
824 .map(|project| project.id)
825 .collect::<HashSet<_>>();
826
827 for project in &participant.projects {
828 if !old_projects.contains(&project.id) {
829 cx.emit(Event::RemoteProjectShared {
830 owner: user.clone(),
831 project_id: project.id,
832 worktree_root_names: project.worktree_root_names.clone(),
833 });
834 }
835 }
836
837 for unshared_project_id in old_projects.difference(&new_projects) {
838 this.joined_projects.retain(|project| {
839 if let Some(project) = project.upgrade() {
840 project.update(cx, |project, cx| {
841 if project.remote_id() == Some(*unshared_project_id) {
842 project.disconnected_from_host(cx);
843 false
844 } else {
845 true
846 }
847 })
848 } else {
849 false
850 }
851 });
852 cx.emit(Event::RemoteProjectUnshared {
853 project_id: *unshared_project_id,
854 });
855 }
856
857 let role = participant.role();
858 let location = ParticipantLocation::from_proto(participant.location)
859 .unwrap_or(ParticipantLocation::External);
860 if let Some(remote_participant) =
861 this.remote_participants.get_mut(&participant.user_id)
862 {
863 remote_participant.peer_id = peer_id;
864 remote_participant.projects = participant.projects;
865 remote_participant.participant_index = participant_index;
866 if location != remote_participant.location
867 || role != remote_participant.role
868 {
869 remote_participant.location = location;
870 remote_participant.role = role;
871 cx.emit(Event::ParticipantLocationChanged {
872 participant_id: peer_id,
873 });
874 }
875 } else {
876 this.remote_participants.insert(
877 participant.user_id,
878 RemoteParticipant {
879 user: user.clone(),
880 participant_index,
881 peer_id,
882 projects: participant.projects,
883 location,
884 role,
885 muted: true,
886 speaking: false,
887 video_tracks: Default::default(),
888 audio_tracks: Default::default(),
889 },
890 );
891
892 // When joining a room start_room_connection gets
893 // called but we have already played the join sound.
894 // Dont play extra sounds over that.
895 if this.created.elapsed() > Duration::from_millis(100) {
896 if let proto::ChannelRole::Guest = role {
897 Audio::play_sound(Sound::GuestJoined, cx);
898 // Do not play join sound in large meetings
899 } else if this.remote_participants().len() < 10 {
900 Audio::play_sound(Sound::Joined, cx);
901 }
902 }
903
904 if let Some(livekit_participants) = &livekit_participants
905 && let Some(livekit_participant) = livekit_participants
906 .get(&ParticipantIdentity(user.legacy_id.to_string()))
907 {
908 for publication in
909 livekit_participant.track_publications().into_values()
910 {
911 if let Some(track) = publication.track() {
912 this.livekit_room_updated(
913 RoomEvent::TrackSubscribed {
914 track,
915 publication,
916 participant: livekit_participant.clone(),
917 },
918 cx,
919 )
920 .warn_on_err();
921 }
922 }
923 }
924 }
925 }
926
927 this.remote_participants.retain(|user_id, participant| {
928 if this.participant_user_ids.contains(user_id) {
929 true
930 } else {
931 for project in &participant.projects {
932 cx.emit(Event::RemoteProjectUnshared {
933 project_id: project.id,
934 });
935 }
936 for sid in participant.video_tracks.keys() {
937 cx.emit(Event::RemoteVideoTrackUnsubscribed { sid: sid.clone() });
938 }
939 if !participant.video_tracks.is_empty() {
940 cx.emit(Event::RemoteVideoTracksChanged {
941 participant_id: participant.peer_id,
942 });
943 }
944 false
945 }
946 });
947 }
948
949 if let Some(pending_participants) = pending_participants.log_err() {
950 this.pending_participants = pending_participants;
951 for participant in &this.pending_participants {
952 this.participant_user_ids.insert(participant.legacy_id);
953 }
954 }
955
956 this.follows_by_leader_id_project_id.clear();
957 for follower in room.followers {
958 let project_id = follower.project_id;
959 let (leader, follower) = match (follower.leader_id, follower.follower_id) {
960 (Some(leader), Some(follower)) => (leader, follower),
961
962 _ => {
963 log::error!("Follower message {follower:?} missing some state");
964 continue;
965 }
966 };
967
968 let list = this
969 .follows_by_leader_id_project_id
970 .entry((leader, project_id))
971 .or_default();
972 if !list.contains(&follower) {
973 list.push(follower);
974 }
975 }
976
977 this.pending_room_update.take();
978 if this.should_leave() {
979 log::info!("room is empty, leaving");
980 this.leave(cx).detach();
981 }
982
983 this.user_store.update(cx, |user_store, cx| {
984 let participant_indices_by_user_id = this
985 .remote_participants
986 .iter()
987 .map(|(user_id, participant)| (*user_id, participant.participant_index))
988 .collect();
989 user_store.set_participant_indices(participant_indices_by_user_id, cx);
990 });
991
992 this.check_invariants();
993 this.room_update_completed_tx.try_send(Some(())).ok();
994 cx.notify();
995 })
996 .ok();
997 })
998 }
999
1000 fn livekit_room_updated(&mut self, event: RoomEvent, cx: &mut Context<Self>) -> Result<()> {
1001 log::trace!(
1002 "client {:?}. livekit event: {:?}",
1003 self.client.user_id(),
1004 &event
1005 );
1006
1007 match event {
1008 RoomEvent::TrackSubscribed {
1009 track,
1010 participant,
1011 publication,
1012 } => {
1013 let user_id = participant.identity().0.parse()?;
1014 let track_id = track.sid();
1015 let participant =
1016 self.remote_participants
1017 .get_mut(&user_id)
1018 .with_context(|| {
1019 format!(
1020 "{:?} subscribed to track by unknown participant {user_id}",
1021 self.client.user_id()
1022 )
1023 })?;
1024 if self.live_kit.as_ref().is_none_or(|kit| kit.deafened) && publication.is_audio() {
1025 publication.set_enabled(false, cx);
1026 }
1027 match track {
1028 livekit_client::RemoteTrack::Audio(track) => {
1029 cx.emit(Event::RemoteAudioTracksChanged {
1030 participant_id: participant.peer_id,
1031 });
1032 if let Some(live_kit) = self.live_kit.as_ref() {
1033 let stream = live_kit.room.play_remote_audio_track(&track, cx)?;
1034 participant.audio_tracks.insert(track_id, (track, stream));
1035 participant.muted = publication.is_muted();
1036 }
1037 }
1038 livekit_client::RemoteTrack::Video(track) => {
1039 cx.emit(Event::RemoteVideoTracksChanged {
1040 participant_id: participant.peer_id,
1041 });
1042 participant.video_tracks.insert(track_id, track);
1043 }
1044 }
1045 }
1046
1047 RoomEvent::TrackUnsubscribed {
1048 track, participant, ..
1049 } => {
1050 let user_id = participant.identity().0.parse()?;
1051 let participant =
1052 self.remote_participants
1053 .get_mut(&user_id)
1054 .with_context(|| {
1055 format!(
1056 "{:?}, unsubscribed from track by unknown participant {user_id}",
1057 self.client.user_id()
1058 )
1059 })?;
1060 match track {
1061 livekit_client::RemoteTrack::Audio(track) => {
1062 participant.audio_tracks.remove(&track.sid());
1063 participant.muted = true;
1064 cx.emit(Event::RemoteAudioTracksChanged {
1065 participant_id: participant.peer_id,
1066 });
1067 }
1068 livekit_client::RemoteTrack::Video(track) => {
1069 participant.video_tracks.remove(&track.sid());
1070 cx.emit(Event::RemoteVideoTracksChanged {
1071 participant_id: participant.peer_id,
1072 });
1073 cx.emit(Event::RemoteVideoTrackUnsubscribed { sid: track.sid() });
1074 }
1075 }
1076 }
1077
1078 RoomEvent::ActiveSpeakersChanged { speakers } => {
1079 let mut speaker_ids = speakers
1080 .into_iter()
1081 .filter_map(|speaker| speaker.identity().0.parse().ok())
1082 .collect::<Vec<u64>>();
1083 speaker_ids.sort_unstable();
1084 for (sid, participant) in &mut self.remote_participants {
1085 participant.speaking = speaker_ids.binary_search(sid).is_ok();
1086 }
1087 if let Some(id) = self.client.user_id()
1088 && let Some(room) = &mut self.live_kit
1089 {
1090 room.speaking = speaker_ids.binary_search(&id).is_ok();
1091 }
1092 }
1093
1094 RoomEvent::TrackMuted {
1095 participant,
1096 publication,
1097 }
1098 | RoomEvent::TrackUnmuted {
1099 participant,
1100 publication,
1101 } => {
1102 let mut found = false;
1103 let user_id = participant.identity().0.parse()?;
1104 let track_id = publication.sid();
1105 if let Some(participant) = self.remote_participants.get_mut(&user_id) {
1106 for (track, _) in participant.audio_tracks.values() {
1107 if track.sid() == track_id {
1108 found = true;
1109 break;
1110 }
1111 }
1112 if found {
1113 participant.muted = publication.is_muted();
1114 }
1115 }
1116 }
1117
1118 RoomEvent::LocalTrackUnpublished { publication, .. } => {
1119 log::info!("unpublished track {}", publication.sid());
1120 if let Some(room) = &mut self.live_kit {
1121 if let LocalTrack::Published {
1122 track_publication, ..
1123 } = &room.microphone_track
1124 && track_publication.sid() == publication.sid()
1125 {
1126 room.microphone_track = LocalTrack::None;
1127 }
1128 if let LocalTrack::Published {
1129 track_publication, ..
1130 } = &room.screen_track
1131 && track_publication.sid() == publication.sid()
1132 {
1133 room.screen_track = LocalTrack::None;
1134 }
1135 }
1136 }
1137
1138 RoomEvent::LocalTrackPublished { publication, .. } => {
1139 log::info!("published track {:?}", publication.sid());
1140 }
1141
1142 RoomEvent::Disconnected { reason } => {
1143 log::info!("disconnected from room: {reason:?}");
1144 self.leave(cx).detach_and_log_err(cx);
1145 }
1146 _ => {}
1147 }
1148
1149 cx.notify();
1150 Ok(())
1151 }
1152
1153 fn check_invariants(&self) {
1154 #[cfg(any(test, feature = "test-support"))]
1155 {
1156 for participant in self.remote_participants.values() {
1157 assert!(
1158 self.participant_user_ids
1159 .contains(&participant.user.legacy_id)
1160 );
1161 assert_ne!(participant.user.legacy_id, self.client.user_id().unwrap());
1162 }
1163
1164 for participant in &self.pending_participants {
1165 assert!(self.participant_user_ids.contains(&participant.legacy_id));
1166 assert_ne!(participant.legacy_id, self.client.user_id().unwrap());
1167 }
1168
1169 assert_eq!(
1170 self.participant_user_ids.len(),
1171 self.remote_participants.len() + self.pending_participants.len()
1172 );
1173 }
1174 }
1175
1176 pub(crate) fn call(
1177 &mut self,
1178 called_user_id: u64,
1179 initial_project_id: Option<u64>,
1180 cx: &mut Context<Self>,
1181 ) -> Task<Result<()>> {
1182 if self.status.is_offline() {
1183 return Task::ready(Err(anyhow!("room is offline")));
1184 }
1185
1186 cx.notify();
1187 let client = self.client.clone();
1188 let room_id = self.id;
1189 self.pending_call_count += 1;
1190 cx.spawn(async move |this, cx| {
1191 let result = client
1192 .request(proto::Call {
1193 room_id,
1194 called_user_id,
1195 initial_project_id,
1196 })
1197 .await;
1198 this.update(cx, |this, cx| {
1199 this.pending_call_count -= 1;
1200 if this.should_leave() {
1201 this.leave(cx).detach_and_log_err(cx);
1202 }
1203 })?;
1204 result?;
1205 Ok(())
1206 })
1207 }
1208
1209 pub fn join_project(
1210 &mut self,
1211 id: u64,
1212 language_registry: Arc<LanguageRegistry>,
1213 fs: Arc<dyn Fs>,
1214 cx: &mut Context<Self>,
1215 ) -> Task<Result<Entity<Project>>> {
1216 let client = self.client.clone();
1217 let user_store = self.user_store.clone();
1218 cx.emit(Event::RemoteProjectJoined { project_id: id });
1219 cx.spawn(async move |this, cx| {
1220 let project =
1221 Project::in_room(id, client, user_store, language_registry, fs, cx.clone()).await?;
1222
1223 this.update(cx, |this, cx| {
1224 this.joined_projects.retain(|project| {
1225 if let Some(project) = project.upgrade() {
1226 !project.read(cx).is_disconnected(cx)
1227 } else {
1228 false
1229 }
1230 });
1231 this.joined_projects.insert(project.downgrade());
1232 })?;
1233 Ok(project)
1234 })
1235 }
1236
1237 pub fn share_project(
1238 &mut self,
1239 project: Entity<Project>,
1240 cx: &mut Context<Self>,
1241 ) -> Task<Result<u64>> {
1242 if let Some(project_id) = project.read(cx).remote_id() {
1243 return Task::ready(Ok(project_id));
1244 }
1245
1246 let request = self.client.request(proto::ShareProject {
1247 room_id: self.id(),
1248 worktrees: project.read(cx).worktree_metadata_protos(cx),
1249 is_ssh_project: project.read(cx).is_via_remote_server(),
1250 windows_paths: Some(project.read(cx).path_style(cx) == PathStyle::Windows),
1251 features: CURRENT_PROJECT_FEATURES
1252 .iter()
1253 .map(|s| s.to_string())
1254 .collect(),
1255 });
1256
1257 cx.spawn(async move |this, cx| {
1258 let response = request.await?;
1259
1260 project.update(cx, |project, cx| project.shared(response.project_id, cx))?;
1261
1262 // If the user's location is in this project, it changes from UnsharedProject to SharedProject.
1263 this.update(cx, |this, cx| {
1264 this.shared_projects.insert(project.downgrade());
1265 let active_project = this.local_participant.active_project.as_ref();
1266 if active_project.is_some_and(|location| *location == project) {
1267 this.set_location(Some(&project), cx)
1268 } else {
1269 Task::ready(Ok(()))
1270 }
1271 })?
1272 .await?;
1273
1274 Ok(response.project_id)
1275 })
1276 }
1277
1278 pub(crate) fn unshare_project(
1279 &mut self,
1280 project: Entity<Project>,
1281 cx: &mut Context<Self>,
1282 ) -> Result<()> {
1283 let project_id = match project.read(cx).remote_id() {
1284 Some(project_id) => project_id,
1285 None => return Ok(()),
1286 };
1287
1288 self.client.send(proto::UnshareProject { project_id })?;
1289 project.update(cx, |this, cx| this.unshare(cx))?;
1290
1291 if self.local_participant.active_project == Some(project.downgrade()) {
1292 self.set_location(Some(&project), cx).detach_and_log_err(cx);
1293 }
1294 Ok(())
1295 }
1296
1297 pub(crate) fn set_location(
1298 &mut self,
1299 project: Option<&Entity<Project>>,
1300 cx: &mut Context<Self>,
1301 ) -> Task<Result<()>> {
1302 if self.status.is_offline() {
1303 return Task::ready(Err(anyhow!("room is offline")));
1304 }
1305
1306 let client = self.client.clone();
1307 let room_id = self.id;
1308 let location = if let Some(project) = project {
1309 self.local_participant.active_project = Some(project.downgrade());
1310 if let Some(project_id) = project.read(cx).remote_id() {
1311 proto::participant_location::Variant::SharedProject(
1312 proto::participant_location::SharedProject { id: project_id },
1313 )
1314 } else {
1315 proto::participant_location::Variant::UnsharedProject(
1316 proto::participant_location::UnsharedProject {},
1317 )
1318 }
1319 } else {
1320 self.local_participant.active_project = None;
1321 proto::participant_location::Variant::External(proto::participant_location::External {})
1322 };
1323
1324 cx.notify();
1325 cx.background_spawn(async move {
1326 client
1327 .request(proto::UpdateParticipantLocation {
1328 room_id,
1329 location: Some(proto::ParticipantLocation {
1330 variant: Some(location),
1331 }),
1332 })
1333 .await?;
1334 Ok(())
1335 })
1336 }
1337
1338 pub fn is_sharing_screen(&self) -> bool {
1339 self.live_kit
1340 .as_ref()
1341 .is_some_and(|live_kit| !matches!(live_kit.screen_track, LocalTrack::None))
1342 }
1343
1344 pub fn shared_screen_id(&self) -> Option<u64> {
1345 self.live_kit.as_ref().and_then(|lk| match lk.screen_track {
1346 LocalTrack::Published { ref _stream, .. } => {
1347 _stream.metadata().ok().map(|meta| meta.id)
1348 }
1349 _ => None,
1350 })
1351 }
1352
1353 pub fn is_sharing_mic(&self) -> bool {
1354 self.live_kit
1355 .as_ref()
1356 .is_some_and(|live_kit| !matches!(live_kit.microphone_track, LocalTrack::None))
1357 }
1358
1359 pub fn is_muted(&self) -> bool {
1360 self.live_kit.as_ref().is_some_and(|live_kit| {
1361 matches!(live_kit.microphone_track, LocalTrack::None)
1362 || live_kit.muted_by_user
1363 || live_kit.deafened
1364 })
1365 }
1366
1367 pub fn muted_by_user(&self) -> bool {
1368 self.live_kit
1369 .as_ref()
1370 .is_some_and(|live_kit| live_kit.muted_by_user)
1371 }
1372
1373 pub fn is_speaking(&self) -> bool {
1374 self.live_kit
1375 .as_ref()
1376 .is_some_and(|live_kit| live_kit.speaking)
1377 }
1378
1379 pub fn is_deafened(&self) -> Option<bool> {
1380 self.live_kit.as_ref().map(|live_kit| live_kit.deafened)
1381 }
1382
1383 pub fn can_use_microphone(&self) -> bool {
1384 use proto::ChannelRole::*;
1385
1386 match self.local_participant.role {
1387 Admin | Member | Talker => true,
1388 Guest | Banned => false,
1389 }
1390 }
1391
1392 pub fn can_share_projects(&self) -> bool {
1393 use proto::ChannelRole::*;
1394 match self.local_participant.role {
1395 Admin | Member => true,
1396 Guest | Banned | Talker => false,
1397 }
1398 }
1399
1400 #[track_caller]
1401 pub fn share_microphone(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
1402 if self.status.is_offline() {
1403 return Task::ready(Err(anyhow!("room is offline")));
1404 }
1405
1406 let (room, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
1407 let publish_id = post_inc(&mut live_kit.next_publish_id);
1408 live_kit.microphone_track = LocalTrack::Pending { publish_id };
1409 cx.notify();
1410 (live_kit.room.clone(), publish_id)
1411 } else {
1412 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1413 };
1414
1415 let is_staff = cx.is_staff();
1416 let user_name = self
1417 .user_store
1418 .read(cx)
1419 .current_user()
1420 .and_then(|user| user.name.clone())
1421 .unwrap_or_else(|| "unknown".to_string());
1422
1423 cx.spawn(async move |this, cx| {
1424 let publication = room
1425 .publish_local_microphone_track(user_name, is_staff, cx)
1426 .await;
1427 this.update(cx, |this, cx| {
1428 let live_kit = this
1429 .live_kit
1430 .as_mut()
1431 .context("live-kit was not initialized")?;
1432
1433 let canceled = if let LocalTrack::Pending {
1434 publish_id: cur_publish_id,
1435 } = &live_kit.microphone_track
1436 {
1437 *cur_publish_id != publish_id
1438 } else {
1439 true
1440 };
1441
1442 match publication {
1443 Ok((publication, stream, input_lag_us)) => {
1444 if canceled {
1445 cx.spawn(async move |_, cx| {
1446 room.unpublish_local_track(publication.sid(), cx).await
1447 })
1448 .detach_and_log_err(cx)
1449 } else {
1450 if live_kit.muted_by_user || live_kit.deafened {
1451 publication.mute(cx);
1452 }
1453 live_kit.input_lag_us = Some(input_lag_us);
1454 live_kit.microphone_track = LocalTrack::Published {
1455 track_publication: publication,
1456 _stream: Box::new(stream),
1457 };
1458 cx.notify();
1459 }
1460 Ok(())
1461 }
1462 Err(error) => {
1463 if canceled {
1464 Ok(())
1465 } else {
1466 live_kit.microphone_track = LocalTrack::None;
1467 cx.notify();
1468 Err(error)
1469 }
1470 }
1471 }
1472 })?
1473 })
1474 }
1475
1476 pub fn share_screen(
1477 &mut self,
1478 source: Rc<dyn ScreenCaptureSource>,
1479 cx: &mut Context<Self>,
1480 ) -> Task<Result<()>> {
1481 if self.status.is_offline() {
1482 return Task::ready(Err(anyhow!("room is offline")));
1483 }
1484 if self.is_sharing_screen() {
1485 return Task::ready(Err(anyhow!("screen was already shared")));
1486 }
1487
1488 let (participant, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
1489 let publish_id = post_inc(&mut live_kit.next_publish_id);
1490 live_kit.screen_track = LocalTrack::Pending { publish_id };
1491 cx.notify();
1492 (live_kit.room.local_participant(), publish_id)
1493 } else {
1494 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1495 };
1496
1497 cx.spawn(async move |this, cx| {
1498 let publication = participant.publish_screenshare_track(&*source, cx).await;
1499
1500 this.update(cx, |this, cx| {
1501 let live_kit = this
1502 .live_kit
1503 .as_mut()
1504 .context("live-kit was not initialized")?;
1505
1506 let canceled = if let LocalTrack::Pending {
1507 publish_id: cur_publish_id,
1508 } = &live_kit.screen_track
1509 {
1510 *cur_publish_id != publish_id
1511 } else {
1512 true
1513 };
1514
1515 match publication {
1516 Ok((publication, stream)) => {
1517 if canceled {
1518 cx.spawn(async move |_, cx| {
1519 participant.unpublish_track(publication.sid(), cx).await
1520 })
1521 .detach()
1522 } else {
1523 live_kit.screen_track = LocalTrack::Published {
1524 track_publication: publication,
1525 _stream: stream,
1526 };
1527 cx.emit(Event::LocalScreenShareStarted);
1528 cx.notify();
1529 }
1530
1531 Audio::play_sound(Sound::StartScreenshare, cx);
1532 Ok(())
1533 }
1534 Err(error) => {
1535 if canceled {
1536 Ok(())
1537 } else {
1538 live_kit.screen_track = LocalTrack::None;
1539 cx.notify();
1540 Err(error)
1541 }
1542 }
1543 }
1544 })?
1545 })
1546 }
1547
1548 #[cfg(target_os = "linux")]
1549 pub fn share_screen_wayland(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
1550 log::info!("will screenshare on wayland");
1551 if self.status.is_offline() {
1552 return Task::ready(Err(anyhow!("room is offline")));
1553 }
1554 if self.is_sharing_screen() {
1555 return Task::ready(Err(anyhow!("screen was already shared")));
1556 }
1557
1558 let (participant, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
1559 let publish_id = post_inc(&mut live_kit.next_publish_id);
1560 live_kit.screen_track = LocalTrack::Pending { publish_id };
1561 cx.notify();
1562 (live_kit.room.local_participant(), publish_id)
1563 } else {
1564 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1565 };
1566
1567 cx.spawn(async move |this, cx| {
1568 let publication = participant.publish_screenshare_track_wayland(cx).await;
1569
1570 this.update(cx, |this, cx| {
1571 let live_kit = this
1572 .live_kit
1573 .as_mut()
1574 .context("live-kit was not initialized")?;
1575
1576 let canceled = if let LocalTrack::Pending {
1577 publish_id: cur_publish_id,
1578 } = &live_kit.screen_track
1579 {
1580 *cur_publish_id != publish_id
1581 } else {
1582 true
1583 };
1584
1585 match publication {
1586 Ok((publication, stream, failure_rx)) => {
1587 if canceled {
1588 cx.spawn(async move |_, cx| {
1589 participant.unpublish_track(publication.sid(), cx).await
1590 })
1591 .detach()
1592 } else {
1593 cx.spawn(async move |this, cx| {
1594 if failure_rx.await.is_ok() {
1595 log::warn!("Wayland capture died, auto-unsharing screen");
1596 let _ =
1597 this.update(cx, |this, cx| this.unshare_screen(false, cx));
1598 }
1599 })
1600 .detach();
1601
1602 live_kit.screen_track = LocalTrack::Published {
1603 track_publication: publication,
1604 _stream: stream,
1605 };
1606 cx.notify();
1607 }
1608
1609 Audio::play_sound(Sound::StartScreenshare, cx);
1610 Ok(())
1611 }
1612 Err(error) => {
1613 if canceled {
1614 Ok(())
1615 } else {
1616 live_kit.screen_track = LocalTrack::None;
1617 cx.notify();
1618 Err(error)
1619 }
1620 }
1621 }
1622 })?
1623 })
1624 }
1625
1626 pub fn toggle_mute(&mut self, cx: &mut Context<Self>) {
1627 if let Some(live_kit) = self.live_kit.as_mut() {
1628 // When unmuting, undeafen if the user was deafened before.
1629 let was_deafened = live_kit.deafened;
1630 if live_kit.muted_by_user
1631 || live_kit.deafened
1632 || matches!(live_kit.microphone_track, LocalTrack::None)
1633 {
1634 live_kit.muted_by_user = false;
1635 live_kit.deafened = false;
1636 } else {
1637 live_kit.muted_by_user = true;
1638 }
1639 let muted = live_kit.muted_by_user;
1640 let should_undeafen = was_deafened && !live_kit.deafened;
1641
1642 if let Some(task) = self.set_mute(muted, cx) {
1643 task.detach_and_log_err(cx);
1644 }
1645
1646 if should_undeafen {
1647 self.set_deafened(false, cx);
1648 }
1649 }
1650 }
1651
1652 pub fn toggle_deafen(&mut self, cx: &mut Context<Self>) {
1653 if let Some(live_kit) = self.live_kit.as_mut() {
1654 // When deafening, mute the microphone if it was not already muted.
1655 // When un-deafening, unmute the microphone, unless it was explicitly muted.
1656 let deafened = !live_kit.deafened;
1657 live_kit.deafened = deafened;
1658 let should_change_mute = !live_kit.muted_by_user;
1659
1660 self.set_deafened(deafened, cx);
1661
1662 if should_change_mute && let Some(task) = self.set_mute(deafened, cx) {
1663 task.detach_and_log_err(cx);
1664 }
1665 }
1666 }
1667
1668 pub fn unshare_screen(&mut self, play_sound: bool, cx: &mut Context<Self>) -> Result<()> {
1669 anyhow::ensure!(!self.status.is_offline(), "room is offline");
1670
1671 let live_kit = self
1672 .live_kit
1673 .as_mut()
1674 .context("live-kit was not initialized")?;
1675 match mem::take(&mut live_kit.screen_track) {
1676 LocalTrack::None => anyhow::bail!("screen was not shared"),
1677 LocalTrack::Pending { .. } => {
1678 cx.notify();
1679 Ok(())
1680 }
1681 LocalTrack::Published {
1682 track_publication, ..
1683 } => {
1684 {
1685 let local_participant = live_kit.room.local_participant();
1686 let sid = track_publication.sid();
1687 cx.spawn(async move |_, cx| local_participant.unpublish_track(sid, cx).await)
1688 .detach_and_log_err(cx);
1689 cx.emit(Event::LocalScreenShareStopped);
1690 cx.notify();
1691 }
1692
1693 if play_sound {
1694 Audio::play_sound(Sound::StopScreenshare, cx);
1695 }
1696
1697 Ok(())
1698 }
1699 }
1700 }
1701
1702 fn set_deafened(&mut self, deafened: bool, cx: &mut Context<Self>) -> Option<()> {
1703 {
1704 let live_kit = self.live_kit.as_mut()?;
1705 cx.notify();
1706 for (_, participant) in live_kit.room.remote_participants() {
1707 for (_, publication) in participant.track_publications() {
1708 if publication.is_audio() {
1709 publication.set_enabled(!deafened, cx);
1710 }
1711 }
1712 }
1713 }
1714
1715 None
1716 }
1717
1718 fn set_mute(&mut self, should_mute: bool, cx: &mut Context<Room>) -> Option<Task<Result<()>>> {
1719 let live_kit = self.live_kit.as_mut()?;
1720 cx.notify();
1721
1722 if should_mute {
1723 Audio::play_sound(Sound::Mute, cx);
1724 } else {
1725 Audio::play_sound(Sound::Unmute, cx);
1726 }
1727
1728 match &mut live_kit.microphone_track {
1729 LocalTrack::None => {
1730 if should_mute {
1731 None
1732 } else {
1733 Some(self.share_microphone(cx))
1734 }
1735 }
1736 LocalTrack::Pending { .. } => None,
1737 LocalTrack::Published {
1738 track_publication, ..
1739 } => {
1740 let guard = Tokio::handle(cx);
1741 if should_mute {
1742 track_publication.mute(cx)
1743 } else {
1744 track_publication.unmute(cx)
1745 }
1746 drop(guard);
1747
1748 None
1749 }
1750 }
1751 }
1752}
1753
1754fn spawn_room_connection(
1755 livekit_connection_info: Option<proto::LiveKitConnectionInfo>,
1756 cx: &mut Context<Room>,
1757) {
1758 if let Some(connection_info) = livekit_connection_info {
1759 cx.spawn(async move |this, cx| {
1760 let (room, mut events) =
1761 livekit::Room::connect(connection_info.server_url, connection_info.token, cx)
1762 .await?;
1763
1764 let weak_room = this.clone();
1765 this.update(cx, |this, cx| {
1766 let _handle_updates = cx.spawn(async move |this, cx| {
1767 while let Some(event) = events.next().await {
1768 if this
1769 .update(cx, |this, cx| {
1770 this.livekit_room_updated(event, cx).warn_on_err();
1771 })
1772 .is_err()
1773 {
1774 break;
1775 }
1776 }
1777 });
1778
1779 let muted_by_user = Room::mute_on_join(cx);
1780 this.live_kit = Some(LiveKitRoom {
1781 room: Rc::new(room),
1782 screen_track: LocalTrack::None,
1783 microphone_track: LocalTrack::None,
1784 input_lag_us: None,
1785 next_publish_id: 0,
1786 muted_by_user,
1787 deafened: false,
1788 speaking: false,
1789 _handle_updates,
1790 });
1791 this.diagnostics = Some(cx.new(|cx| CallDiagnostics::new(weak_room, cx)));
1792
1793 // Always open the microphone track on join, even when
1794 // `muted_by_user` is set. Note that the microphone will still
1795 // be muted, as it is still gated in `share_microphone` by
1796 // `muted_by_user`. For users that have `mute_on_join` enabled,
1797 // this moves the Bluetooth profile switch (A2DP -> HFP) (which
1798 // can cause 1-2 seconds of audio silence on some Bluetooth
1799 // headphones) from first unmute to channel join, where
1800 // instability is expected.
1801 if this.can_use_microphone() {
1802 this.share_microphone(cx)
1803 } else {
1804 Task::ready(Ok(()))
1805 }
1806 })?
1807 .await
1808 })
1809 .detach_and_log_err(cx);
1810 }
1811}
1812
1813struct LiveKitRoom {
1814 room: Rc<livekit::Room>,
1815 screen_track: LocalTrack<dyn ScreenCaptureStream>,
1816 microphone_track: LocalTrack<AudioStream>,
1817 /// Shared atomic storing the most recent input lag measurement in microseconds.
1818 /// Written by the audio capture/transmit pipeline, read here for diagnostics.
1819 input_lag_us: Option<Arc<AtomicU64>>,
1820 /// Tracks whether we're currently in a muted state due to auto-mute from deafening or manual mute performed by user.
1821 muted_by_user: bool,
1822 deafened: bool,
1823 speaking: bool,
1824 next_publish_id: usize,
1825 _handle_updates: Task<()>,
1826}
1827
1828impl LiveKitRoom {
1829 fn stop_publishing(&mut self, cx: &mut Context<Room>) {
1830 let mut tracks_to_unpublish = Vec::new();
1831 if let LocalTrack::Published {
1832 track_publication, ..
1833 } = mem::replace(&mut self.microphone_track, LocalTrack::None)
1834 {
1835 tracks_to_unpublish.push(track_publication.sid());
1836 self.input_lag_us = None;
1837 cx.notify();
1838 }
1839
1840 if let LocalTrack::Published {
1841 track_publication, ..
1842 } = mem::replace(&mut self.screen_track, LocalTrack::None)
1843 {
1844 tracks_to_unpublish.push(track_publication.sid());
1845 cx.notify();
1846 }
1847
1848 let participant = self.room.local_participant();
1849 cx.spawn(async move |_, cx| {
1850 for sid in tracks_to_unpublish {
1851 participant.unpublish_track(sid, cx).await.log_err();
1852 }
1853 })
1854 .detach();
1855 }
1856}
1857
1858#[derive(Default)]
1859enum LocalTrack<Stream: ?Sized> {
1860 #[default]
1861 None,
1862 Pending {
1863 publish_id: usize,
1864 },
1865 Published {
1866 track_publication: LocalTrackPublication,
1867 _stream: Box<Stream>,
1868 },
1869}
1870
1871#[derive(Copy, Clone, PartialEq, Eq)]
1872pub enum RoomStatus {
1873 Online,
1874 Rejoining,
1875 Offline,
1876}
1877
1878impl RoomStatus {
1879 pub fn is_offline(&self) -> bool {
1880 matches!(self, RoomStatus::Offline)
1881 }
1882
1883 pub fn is_online(&self) -> bool {
1884 matches!(self, RoomStatus::Online)
1885 }
1886}
1887