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