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