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 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.context("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.context("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 .context("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 anyhow::bail!("can't reconnect to room: client failed to re-establish connection");
432 }
433
434 fn rejoin(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
435 let mut projects = HashMap::default();
436 let mut reshared_projects = Vec::new();
437 let mut rejoined_projects = Vec::new();
438 self.shared_projects.retain(|project| {
439 if let Some(handle) = project.upgrade() {
440 let project = handle.read(cx);
441 if let Some(project_id) = project.remote_id() {
442 projects.insert(project_id, handle.clone());
443 reshared_projects.push(proto::UpdateProject {
444 project_id,
445 worktrees: project.worktree_metadata_protos(cx),
446 });
447 return true;
448 }
449 }
450 false
451 });
452 self.joined_projects.retain(|project| {
453 if let Some(handle) = project.upgrade() {
454 let project = handle.read(cx);
455 if let Some(project_id) = project.remote_id() {
456 projects.insert(project_id, handle.clone());
457 let mut worktrees = Vec::new();
458 let mut repositories = Vec::new();
459 for worktree in project.worktrees(cx) {
460 let worktree = worktree.read(cx);
461 worktrees.push(proto::RejoinWorktree {
462 id: worktree.id().to_proto(),
463 scan_id: worktree.completed_scan_id() as u64,
464 });
465 }
466 for (entry_id, repository) in project.repositories(cx) {
467 let repository = repository.read(cx);
468 repositories.push(proto::RejoinRepository {
469 id: entry_id.to_proto(),
470 scan_id: repository.scan_id,
471 });
472 }
473
474 rejoined_projects.push(proto::RejoinProject {
475 id: project_id,
476 worktrees,
477 repositories,
478 });
479 }
480 return true;
481 }
482 false
483 });
484
485 let response = self.client.request_envelope(proto::RejoinRoom {
486 id: self.id,
487 reshared_projects,
488 rejoined_projects,
489 });
490
491 cx.spawn(async move |this, cx| {
492 let response = response.await?;
493 let message_id = response.message_id;
494 let response = response.payload;
495 let room_proto = response.room.context("invalid room")?;
496 this.update(cx, |this, cx| {
497 this.status = RoomStatus::Online;
498 this.apply_room_update(room_proto, cx)?;
499
500 for reshared_project in response.reshared_projects {
501 if let Some(project) = projects.get(&reshared_project.id) {
502 project.update(cx, |project, cx| {
503 project.reshared(reshared_project, cx).log_err();
504 });
505 }
506 }
507
508 for rejoined_project in response.rejoined_projects {
509 if let Some(project) = projects.get(&rejoined_project.id) {
510 project.update(cx, |project, cx| {
511 project.rejoined(rejoined_project, message_id, cx).log_err();
512 });
513 }
514 }
515
516 anyhow::Ok(())
517 })?
518 })
519 }
520
521 pub fn id(&self) -> u64 {
522 self.id
523 }
524
525 pub fn status(&self) -> RoomStatus {
526 self.status
527 }
528
529 pub fn local_participant(&self) -> &LocalParticipant {
530 &self.local_participant
531 }
532
533 pub fn local_participant_user(&self, cx: &App) -> Option<Arc<User>> {
534 self.user_store.read(cx).current_user()
535 }
536
537 pub fn remote_participants(&self) -> &BTreeMap<u64, RemoteParticipant> {
538 &self.remote_participants
539 }
540
541 pub fn remote_participant_for_peer_id(&self, peer_id: PeerId) -> Option<&RemoteParticipant> {
542 self.remote_participants
543 .values()
544 .find(|p| p.peer_id == peer_id)
545 }
546
547 pub fn role_for_user(&self, user_id: u64) -> Option<proto::ChannelRole> {
548 self.remote_participants
549 .get(&user_id)
550 .map(|participant| participant.role)
551 }
552
553 pub fn contains_guests(&self) -> bool {
554 self.local_participant.role == proto::ChannelRole::Guest
555 || self
556 .remote_participants
557 .values()
558 .any(|p| p.role == proto::ChannelRole::Guest)
559 }
560
561 pub fn local_participant_is_admin(&self) -> bool {
562 self.local_participant.role == proto::ChannelRole::Admin
563 }
564
565 pub fn local_participant_is_guest(&self) -> bool {
566 self.local_participant.role == proto::ChannelRole::Guest
567 }
568
569 pub fn set_participant_role(
570 &mut self,
571 user_id: u64,
572 role: proto::ChannelRole,
573 cx: &Context<Self>,
574 ) -> Task<Result<()>> {
575 let client = self.client.clone();
576 let room_id = self.id;
577 let role = role.into();
578 cx.spawn(async move |_, _| {
579 client
580 .request(proto::SetRoomParticipantRole {
581 room_id,
582 user_id,
583 role,
584 })
585 .await
586 .map(|_| ())
587 })
588 }
589
590 pub fn pending_participants(&self) -> &[Arc<User>] {
591 &self.pending_participants
592 }
593
594 pub fn contains_participant(&self, user_id: u64) -> bool {
595 self.participant_user_ids.contains(&user_id)
596 }
597
598 pub fn followers_for(&self, leader_id: PeerId, project_id: u64) -> &[PeerId] {
599 self.follows_by_leader_id_project_id
600 .get(&(leader_id, project_id))
601 .map_or(&[], |v| v.as_slice())
602 }
603
604 /// Returns the most 'active' projects, defined as most people in the project
605 pub fn most_active_project(&self, cx: &App) -> Option<(u64, u64)> {
606 let mut project_hosts_and_guest_counts = HashMap::<u64, (Option<u64>, u32)>::default();
607 for participant in self.remote_participants.values() {
608 match participant.location {
609 ParticipantLocation::SharedProject { project_id } => {
610 project_hosts_and_guest_counts
611 .entry(project_id)
612 .or_default()
613 .1 += 1;
614 }
615 ParticipantLocation::External | ParticipantLocation::UnsharedProject => {}
616 }
617 for project in &participant.projects {
618 project_hosts_and_guest_counts
619 .entry(project.id)
620 .or_default()
621 .0 = Some(participant.user.id);
622 }
623 }
624
625 if let Some(user) = self.user_store.read(cx).current_user() {
626 for project in &self.local_participant.projects {
627 project_hosts_and_guest_counts
628 .entry(project.id)
629 .or_default()
630 .0 = Some(user.id);
631 }
632 }
633
634 project_hosts_and_guest_counts
635 .into_iter()
636 .filter_map(|(id, (host, guest_count))| Some((id, host?, guest_count)))
637 .max_by_key(|(_, _, guest_count)| *guest_count)
638 .map(|(id, host, _)| (id, host))
639 }
640
641 async fn handle_room_updated(
642 this: Entity<Self>,
643 envelope: TypedEnvelope<proto::RoomUpdated>,
644 mut cx: AsyncApp,
645 ) -> Result<()> {
646 let room = envelope.payload.room.context("invalid room")?;
647 this.update(&mut cx, |this, cx| this.apply_room_update(room, cx))?
648 }
649
650 fn apply_room_update(&mut self, room: proto::Room, cx: &mut Context<Self>) -> Result<()> {
651 log::trace!(
652 "client {:?}. room update: {:?}",
653 self.client.user_id(),
654 &room
655 );
656
657 self.pending_room_update = Some(self.start_room_connection(room, cx));
658
659 cx.notify();
660 Ok(())
661 }
662
663 pub fn room_update_completed(&mut self) -> impl Future<Output = ()> + use<> {
664 let mut done_rx = self.room_update_completed_rx.clone();
665 async move {
666 while let Some(result) = done_rx.next().await {
667 if result.is_some() {
668 break;
669 }
670 }
671 }
672 }
673
674 fn start_room_connection(&self, mut room: proto::Room, cx: &mut Context<Self>) -> Task<()> {
675 // Filter ourselves out from the room's participants.
676 let local_participant_ix = room
677 .participants
678 .iter()
679 .position(|participant| Some(participant.user_id) == self.client.user_id());
680 let local_participant = local_participant_ix.map(|ix| room.participants.swap_remove(ix));
681
682 let pending_participant_user_ids = room
683 .pending_participants
684 .iter()
685 .map(|p| p.user_id)
686 .collect::<Vec<_>>();
687
688 let remote_participant_user_ids = room
689 .participants
690 .iter()
691 .map(|p| p.user_id)
692 .collect::<Vec<_>>();
693
694 let (remote_participants, pending_participants) =
695 self.user_store.update(cx, move |user_store, cx| {
696 (
697 user_store.get_users(remote_participant_user_ids, cx),
698 user_store.get_users(pending_participant_user_ids, cx),
699 )
700 });
701 cx.spawn(async move |this, cx| {
702 let (remote_participants, pending_participants) =
703 futures::join!(remote_participants, pending_participants);
704
705 this.update(cx, |this, cx| {
706 this.participant_user_ids.clear();
707
708 if let Some(participant) = local_participant {
709 let role = participant.role();
710 this.local_participant.projects = participant.projects;
711 if this.local_participant.role != role {
712 this.local_participant.role = role;
713
714 if role == proto::ChannelRole::Guest {
715 for project in mem::take(&mut this.shared_projects) {
716 if let Some(project) = project.upgrade() {
717 this.unshare_project(project, cx).log_err();
718 }
719 }
720 this.local_participant.projects.clear();
721 if let Some(livekit_room) = &mut this.live_kit {
722 livekit_room.stop_publishing(cx);
723 }
724 }
725
726 this.joined_projects.retain(|project| {
727 if let Some(project) = project.upgrade() {
728 project.update(cx, |project, cx| project.set_role(role, cx));
729 true
730 } else {
731 false
732 }
733 });
734 }
735 } else {
736 this.local_participant.projects.clear();
737 }
738
739 let livekit_participants = this
740 .live_kit
741 .as_ref()
742 .map(|live_kit| live_kit.room.remote_participants());
743
744 if let Some(participants) = remote_participants.log_err() {
745 for (participant, user) in room.participants.into_iter().zip(participants) {
746 let Some(peer_id) = participant.peer_id else {
747 continue;
748 };
749 let participant_index = ParticipantIndex(participant.participant_index);
750 this.participant_user_ids.insert(participant.user_id);
751
752 let old_projects = this
753 .remote_participants
754 .get(&participant.user_id)
755 .into_iter()
756 .flat_map(|existing| &existing.projects)
757 .map(|project| project.id)
758 .collect::<HashSet<_>>();
759 let new_projects = participant
760 .projects
761 .iter()
762 .map(|project| project.id)
763 .collect::<HashSet<_>>();
764
765 for project in &participant.projects {
766 if !old_projects.contains(&project.id) {
767 cx.emit(Event::RemoteProjectShared {
768 owner: user.clone(),
769 project_id: project.id,
770 worktree_root_names: project.worktree_root_names.clone(),
771 });
772 }
773 }
774
775 for unshared_project_id in old_projects.difference(&new_projects) {
776 this.joined_projects.retain(|project| {
777 if let Some(project) = project.upgrade() {
778 project.update(cx, |project, cx| {
779 if project.remote_id() == Some(*unshared_project_id) {
780 project.disconnected_from_host(cx);
781 false
782 } else {
783 true
784 }
785 })
786 } else {
787 false
788 }
789 });
790 cx.emit(Event::RemoteProjectUnshared {
791 project_id: *unshared_project_id,
792 });
793 }
794
795 let role = participant.role();
796 let location = ParticipantLocation::from_proto(participant.location)
797 .unwrap_or(ParticipantLocation::External);
798 if let Some(remote_participant) =
799 this.remote_participants.get_mut(&participant.user_id)
800 {
801 remote_participant.peer_id = peer_id;
802 remote_participant.projects = participant.projects;
803 remote_participant.participant_index = participant_index;
804 if location != remote_participant.location
805 || role != remote_participant.role
806 {
807 remote_participant.location = location;
808 remote_participant.role = role;
809 cx.emit(Event::ParticipantLocationChanged {
810 participant_id: peer_id,
811 });
812 }
813 } else {
814 this.remote_participants.insert(
815 participant.user_id,
816 RemoteParticipant {
817 user: user.clone(),
818 participant_index,
819 peer_id,
820 projects: participant.projects,
821 location,
822 role,
823 muted: true,
824 speaking: false,
825 video_tracks: Default::default(),
826 audio_tracks: Default::default(),
827 },
828 );
829
830 Audio::play_sound(Sound::Joined, cx);
831 if let Some(livekit_participants) = &livekit_participants {
832 if let Some(livekit_participant) = livekit_participants
833 .get(&ParticipantIdentity(user.id.to_string()))
834 {
835 for publication in
836 livekit_participant.track_publications().into_values()
837 {
838 if let Some(track) = publication.track() {
839 this.livekit_room_updated(
840 RoomEvent::TrackSubscribed {
841 track,
842 publication,
843 participant: livekit_participant.clone(),
844 },
845 cx,
846 )
847 .warn_on_err();
848 }
849 }
850 }
851 }
852 }
853 }
854
855 this.remote_participants.retain(|user_id, participant| {
856 if this.participant_user_ids.contains(user_id) {
857 true
858 } else {
859 for project in &participant.projects {
860 cx.emit(Event::RemoteProjectUnshared {
861 project_id: project.id,
862 });
863 }
864 false
865 }
866 });
867 }
868
869 if let Some(pending_participants) = pending_participants.log_err() {
870 this.pending_participants = pending_participants;
871 for participant in &this.pending_participants {
872 this.participant_user_ids.insert(participant.id);
873 }
874 }
875
876 this.follows_by_leader_id_project_id.clear();
877 for follower in room.followers {
878 let project_id = follower.project_id;
879 let (leader, follower) = match (follower.leader_id, follower.follower_id) {
880 (Some(leader), Some(follower)) => (leader, follower),
881
882 _ => {
883 log::error!("Follower message {follower:?} missing some state");
884 continue;
885 }
886 };
887
888 let list = this
889 .follows_by_leader_id_project_id
890 .entry((leader, project_id))
891 .or_default();
892 if !list.contains(&follower) {
893 list.push(follower);
894 }
895 }
896
897 this.pending_room_update.take();
898 if this.should_leave() {
899 log::info!("room is empty, leaving");
900 this.leave(cx).detach();
901 }
902
903 this.user_store.update(cx, |user_store, cx| {
904 let participant_indices_by_user_id = this
905 .remote_participants
906 .iter()
907 .map(|(user_id, participant)| (*user_id, participant.participant_index))
908 .collect();
909 user_store.set_participant_indices(participant_indices_by_user_id, cx);
910 });
911
912 this.check_invariants();
913 this.room_update_completed_tx.try_send(Some(())).ok();
914 cx.notify();
915 })
916 .ok();
917 })
918 }
919
920 fn livekit_room_updated(&mut self, event: RoomEvent, cx: &mut Context<Self>) -> Result<()> {
921 log::trace!(
922 "client {:?}. livekit event: {:?}",
923 self.client.user_id(),
924 &event
925 );
926
927 match event {
928 RoomEvent::TrackSubscribed {
929 track,
930 participant,
931 publication,
932 } => {
933 let user_id = participant.identity().0.parse()?;
934 let track_id = track.sid();
935 let participant =
936 self.remote_participants
937 .get_mut(&user_id)
938 .with_context(|| {
939 format!(
940 "{:?} subscribed to track by unknown participant {user_id}",
941 self.client.user_id()
942 )
943 })?;
944 if self.live_kit.as_ref().map_or(true, |kit| kit.deafened) {
945 if publication.is_audio() {
946 publication.set_enabled(false, cx);
947 }
948 }
949 match track {
950 livekit_client::RemoteTrack::Audio(track) => {
951 cx.emit(Event::RemoteAudioTracksChanged {
952 participant_id: participant.peer_id,
953 });
954 if let Some(live_kit) = self.live_kit.as_ref() {
955 let stream = live_kit.room.play_remote_audio_track(&track, cx)?;
956 participant.audio_tracks.insert(track_id, (track, stream));
957 participant.muted = publication.is_muted();
958 }
959 }
960 livekit_client::RemoteTrack::Video(track) => {
961 cx.emit(Event::RemoteVideoTracksChanged {
962 participant_id: participant.peer_id,
963 });
964 participant.video_tracks.insert(track_id, track);
965 }
966 }
967 }
968
969 RoomEvent::TrackUnsubscribed {
970 track, participant, ..
971 } => {
972 let user_id = participant.identity().0.parse()?;
973 let participant =
974 self.remote_participants
975 .get_mut(&user_id)
976 .with_context(|| {
977 format!(
978 "{:?}, unsubscribed from track by unknown participant {user_id}",
979 self.client.user_id()
980 )
981 })?;
982 match track {
983 livekit_client::RemoteTrack::Audio(track) => {
984 participant.audio_tracks.remove(&track.sid());
985 participant.muted = true;
986 cx.emit(Event::RemoteAudioTracksChanged {
987 participant_id: participant.peer_id,
988 });
989 }
990 livekit_client::RemoteTrack::Video(track) => {
991 participant.video_tracks.remove(&track.sid());
992 cx.emit(Event::RemoteVideoTracksChanged {
993 participant_id: participant.peer_id,
994 });
995 cx.emit(Event::RemoteVideoTrackUnsubscribed { sid: track.sid() });
996 }
997 }
998 }
999
1000 RoomEvent::ActiveSpeakersChanged { speakers } => {
1001 let mut speaker_ids = speakers
1002 .into_iter()
1003 .filter_map(|speaker| speaker.identity().0.parse().ok())
1004 .collect::<Vec<u64>>();
1005 speaker_ids.sort_unstable();
1006 for (sid, participant) in &mut self.remote_participants {
1007 participant.speaking = speaker_ids.binary_search(sid).is_ok();
1008 }
1009 if let Some(id) = self.client.user_id() {
1010 if let Some(room) = &mut self.live_kit {
1011 room.speaking = speaker_ids.binary_search(&id).is_ok();
1012 }
1013 }
1014 }
1015
1016 RoomEvent::TrackMuted {
1017 participant,
1018 publication,
1019 }
1020 | RoomEvent::TrackUnmuted {
1021 participant,
1022 publication,
1023 } => {
1024 let mut found = false;
1025 let user_id = participant.identity().0.parse()?;
1026 let track_id = publication.sid();
1027 if let Some(participant) = self.remote_participants.get_mut(&user_id) {
1028 for (track, _) in participant.audio_tracks.values() {
1029 if track.sid() == track_id {
1030 found = true;
1031 break;
1032 }
1033 }
1034 if found {
1035 participant.muted = publication.is_muted();
1036 }
1037 }
1038 }
1039
1040 RoomEvent::LocalTrackUnpublished { publication, .. } => {
1041 log::info!("unpublished track {}", publication.sid());
1042 if let Some(room) = &mut self.live_kit {
1043 if let LocalTrack::Published {
1044 track_publication, ..
1045 } = &room.microphone_track
1046 {
1047 if track_publication.sid() == publication.sid() {
1048 room.microphone_track = LocalTrack::None;
1049 }
1050 }
1051 if let LocalTrack::Published {
1052 track_publication, ..
1053 } = &room.screen_track
1054 {
1055 if track_publication.sid() == publication.sid() {
1056 room.screen_track = LocalTrack::None;
1057 }
1058 }
1059 }
1060 }
1061
1062 RoomEvent::LocalTrackPublished { publication, .. } => {
1063 log::info!("published track {:?}", publication.sid());
1064 }
1065
1066 RoomEvent::Disconnected { reason } => {
1067 log::info!("disconnected from room: {reason:?}");
1068 self.leave(cx).detach_and_log_err(cx);
1069 }
1070 _ => {}
1071 }
1072
1073 cx.notify();
1074 Ok(())
1075 }
1076
1077 fn check_invariants(&self) {
1078 #[cfg(any(test, feature = "test-support"))]
1079 {
1080 for participant in self.remote_participants.values() {
1081 assert!(self.participant_user_ids.contains(&participant.user.id));
1082 assert_ne!(participant.user.id, self.client.user_id().unwrap());
1083 }
1084
1085 for participant in &self.pending_participants {
1086 assert!(self.participant_user_ids.contains(&participant.id));
1087 assert_ne!(participant.id, self.client.user_id().unwrap());
1088 }
1089
1090 assert_eq!(
1091 self.participant_user_ids.len(),
1092 self.remote_participants.len() + self.pending_participants.len()
1093 );
1094 }
1095 }
1096
1097 pub(crate) fn call(
1098 &mut self,
1099 called_user_id: u64,
1100 initial_project_id: Option<u64>,
1101 cx: &mut Context<Self>,
1102 ) -> Task<Result<()>> {
1103 if self.status.is_offline() {
1104 return Task::ready(Err(anyhow!("room is offline")));
1105 }
1106
1107 cx.notify();
1108 let client = self.client.clone();
1109 let room_id = self.id;
1110 self.pending_call_count += 1;
1111 cx.spawn(async move |this, cx| {
1112 let result = client
1113 .request(proto::Call {
1114 room_id,
1115 called_user_id,
1116 initial_project_id,
1117 })
1118 .await;
1119 this.update(cx, |this, cx| {
1120 this.pending_call_count -= 1;
1121 if this.should_leave() {
1122 this.leave(cx).detach_and_log_err(cx);
1123 }
1124 })?;
1125 result?;
1126 Ok(())
1127 })
1128 }
1129
1130 pub fn join_project(
1131 &mut self,
1132 id: u64,
1133 language_registry: Arc<LanguageRegistry>,
1134 fs: Arc<dyn Fs>,
1135 cx: &mut Context<Self>,
1136 ) -> Task<Result<Entity<Project>>> {
1137 let client = self.client.clone();
1138 let user_store = self.user_store.clone();
1139 cx.emit(Event::RemoteProjectJoined { project_id: id });
1140 cx.spawn(async move |this, cx| {
1141 let project =
1142 Project::in_room(id, client, user_store, language_registry, fs, cx.clone()).await?;
1143
1144 this.update(cx, |this, cx| {
1145 this.joined_projects.retain(|project| {
1146 if let Some(project) = project.upgrade() {
1147 !project.read(cx).is_disconnected(cx)
1148 } else {
1149 false
1150 }
1151 });
1152 this.joined_projects.insert(project.downgrade());
1153 })?;
1154 Ok(project)
1155 })
1156 }
1157
1158 pub fn share_project(
1159 &mut self,
1160 project: Entity<Project>,
1161 cx: &mut Context<Self>,
1162 ) -> Task<Result<u64>> {
1163 if let Some(project_id) = project.read(cx).remote_id() {
1164 return Task::ready(Ok(project_id));
1165 }
1166
1167 let request = self.client.request(proto::ShareProject {
1168 room_id: self.id(),
1169 worktrees: project.read(cx).worktree_metadata_protos(cx),
1170 is_ssh_project: project.read(cx).is_via_ssh(),
1171 });
1172
1173 cx.spawn(async move |this, cx| {
1174 let response = request.await?;
1175
1176 project.update(cx, |project, cx| project.shared(response.project_id, cx))??;
1177
1178 // If the user's location is in this project, it changes from UnsharedProject to SharedProject.
1179 this.update(cx, |this, cx| {
1180 this.shared_projects.insert(project.downgrade());
1181 let active_project = this.local_participant.active_project.as_ref();
1182 if active_project.map_or(false, |location| *location == project) {
1183 this.set_location(Some(&project), cx)
1184 } else {
1185 Task::ready(Ok(()))
1186 }
1187 })?
1188 .await?;
1189
1190 Ok(response.project_id)
1191 })
1192 }
1193
1194 pub(crate) fn unshare_project(
1195 &mut self,
1196 project: Entity<Project>,
1197 cx: &mut Context<Self>,
1198 ) -> Result<()> {
1199 let project_id = match project.read(cx).remote_id() {
1200 Some(project_id) => project_id,
1201 None => return Ok(()),
1202 };
1203
1204 self.client.send(proto::UnshareProject { project_id })?;
1205 project.update(cx, |this, cx| this.unshare(cx))?;
1206
1207 if self.local_participant.active_project == Some(project.downgrade()) {
1208 self.set_location(Some(&project), cx).detach_and_log_err(cx);
1209 }
1210 Ok(())
1211 }
1212
1213 pub(crate) fn set_location(
1214 &mut self,
1215 project: Option<&Entity<Project>>,
1216 cx: &mut Context<Self>,
1217 ) -> Task<Result<()>> {
1218 if self.status.is_offline() {
1219 return Task::ready(Err(anyhow!("room is offline")));
1220 }
1221
1222 let client = self.client.clone();
1223 let room_id = self.id;
1224 let location = if let Some(project) = project {
1225 self.local_participant.active_project = Some(project.downgrade());
1226 if let Some(project_id) = project.read(cx).remote_id() {
1227 proto::participant_location::Variant::SharedProject(
1228 proto::participant_location::SharedProject { id: project_id },
1229 )
1230 } else {
1231 proto::participant_location::Variant::UnsharedProject(
1232 proto::participant_location::UnsharedProject {},
1233 )
1234 }
1235 } else {
1236 self.local_participant.active_project = None;
1237 proto::participant_location::Variant::External(proto::participant_location::External {})
1238 };
1239
1240 cx.notify();
1241 cx.background_spawn(async move {
1242 client
1243 .request(proto::UpdateParticipantLocation {
1244 room_id,
1245 location: Some(proto::ParticipantLocation {
1246 variant: Some(location),
1247 }),
1248 })
1249 .await?;
1250 Ok(())
1251 })
1252 }
1253
1254 pub fn is_screen_sharing(&self) -> bool {
1255 self.live_kit.as_ref().map_or(false, |live_kit| {
1256 !matches!(live_kit.screen_track, LocalTrack::None)
1257 })
1258 }
1259
1260 pub fn is_sharing_mic(&self) -> bool {
1261 self.live_kit.as_ref().map_or(false, |live_kit| {
1262 !matches!(live_kit.microphone_track, LocalTrack::None)
1263 })
1264 }
1265
1266 pub fn is_muted(&self) -> bool {
1267 self.live_kit.as_ref().map_or(false, |live_kit| {
1268 matches!(live_kit.microphone_track, LocalTrack::None)
1269 || live_kit.muted_by_user
1270 || live_kit.deafened
1271 })
1272 }
1273
1274 pub fn muted_by_user(&self) -> bool {
1275 self.live_kit
1276 .as_ref()
1277 .map_or(false, |live_kit| live_kit.muted_by_user)
1278 }
1279
1280 pub fn is_speaking(&self) -> bool {
1281 self.live_kit
1282 .as_ref()
1283 .map_or(false, |live_kit| live_kit.speaking)
1284 }
1285
1286 pub fn is_deafened(&self) -> Option<bool> {
1287 self.live_kit.as_ref().map(|live_kit| live_kit.deafened)
1288 }
1289
1290 pub fn can_use_microphone(&self) -> bool {
1291 use proto::ChannelRole::*;
1292
1293 match self.local_participant.role {
1294 Admin | Member | Talker => true,
1295 Guest | Banned => false,
1296 }
1297 }
1298
1299 pub fn can_share_projects(&self) -> bool {
1300 use proto::ChannelRole::*;
1301 match self.local_participant.role {
1302 Admin | Member => true,
1303 Guest | Banned | Talker => false,
1304 }
1305 }
1306
1307 #[track_caller]
1308 pub fn share_microphone(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
1309 if self.status.is_offline() {
1310 return Task::ready(Err(anyhow!("room is offline")));
1311 }
1312
1313 let (room, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
1314 let publish_id = post_inc(&mut live_kit.next_publish_id);
1315 live_kit.microphone_track = LocalTrack::Pending { publish_id };
1316 cx.notify();
1317 (live_kit.room.clone(), publish_id)
1318 } else {
1319 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1320 };
1321
1322 cx.spawn(async move |this, cx| {
1323 let publication = room.publish_local_microphone_track(cx).await;
1324 this.update(cx, |this, cx| {
1325 let live_kit = this
1326 .live_kit
1327 .as_mut()
1328 .context("live-kit was not initialized")?;
1329
1330 let canceled = if let LocalTrack::Pending {
1331 publish_id: cur_publish_id,
1332 } = &live_kit.microphone_track
1333 {
1334 *cur_publish_id != publish_id
1335 } else {
1336 true
1337 };
1338
1339 match publication {
1340 Ok((publication, stream)) => {
1341 if canceled {
1342 cx.spawn(async move |_, cx| {
1343 room.unpublish_local_track(publication.sid(), cx).await
1344 })
1345 .detach_and_log_err(cx)
1346 } else {
1347 if live_kit.muted_by_user || live_kit.deafened {
1348 publication.mute(cx);
1349 }
1350 live_kit.microphone_track = LocalTrack::Published {
1351 track_publication: publication,
1352 _stream: Box::new(stream),
1353 };
1354 cx.notify();
1355 }
1356 Ok(())
1357 }
1358 Err(error) => {
1359 if canceled {
1360 Ok(())
1361 } else {
1362 live_kit.microphone_track = LocalTrack::None;
1363 cx.notify();
1364 Err(error)
1365 }
1366 }
1367 }
1368 })?
1369 })
1370 }
1371
1372 pub fn share_screen(&mut self, cx: &mut Context<Self>) -> Task<Result<()>> {
1373 if self.status.is_offline() {
1374 return Task::ready(Err(anyhow!("room is offline")));
1375 }
1376 if self.is_screen_sharing() {
1377 return Task::ready(Err(anyhow!("screen was already shared")));
1378 }
1379
1380 let (participant, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
1381 let publish_id = post_inc(&mut live_kit.next_publish_id);
1382 live_kit.screen_track = LocalTrack::Pending { publish_id };
1383 cx.notify();
1384 (live_kit.room.local_participant(), publish_id)
1385 } else {
1386 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1387 };
1388
1389 let sources = cx.screen_capture_sources();
1390
1391 cx.spawn(async move |this, cx| {
1392 let sources = sources
1393 .await
1394 .map_err(|error| error.into())
1395 .and_then(|sources| sources);
1396 let source =
1397 sources.and_then(|sources| sources.into_iter().next().context("no display found"));
1398
1399 let publication = match source {
1400 Ok(source) => participant.publish_screenshare_track(&*source, cx).await,
1401 Err(error) => Err(error),
1402 };
1403
1404 this.update(cx, |this, cx| {
1405 let live_kit = this
1406 .live_kit
1407 .as_mut()
1408 .context("live-kit was not initialized")?;
1409
1410 let canceled = if let LocalTrack::Pending {
1411 publish_id: cur_publish_id,
1412 } = &live_kit.screen_track
1413 {
1414 *cur_publish_id != publish_id
1415 } else {
1416 true
1417 };
1418
1419 match publication {
1420 Ok((publication, stream)) => {
1421 if canceled {
1422 cx.spawn(async move |_, cx| {
1423 participant.unpublish_track(publication.sid(), cx).await
1424 })
1425 .detach()
1426 } else {
1427 live_kit.screen_track = LocalTrack::Published {
1428 track_publication: publication,
1429 _stream: Box::new(stream),
1430 };
1431 cx.notify();
1432 }
1433
1434 Audio::play_sound(Sound::StartScreenshare, cx);
1435 Ok(())
1436 }
1437 Err(error) => {
1438 if canceled {
1439 Ok(())
1440 } else {
1441 live_kit.screen_track = LocalTrack::None;
1442 cx.notify();
1443 Err(error)
1444 }
1445 }
1446 }
1447 })?
1448 })
1449 }
1450
1451 pub fn toggle_mute(&mut self, cx: &mut Context<Self>) {
1452 if let Some(live_kit) = self.live_kit.as_mut() {
1453 // When unmuting, undeafen if the user was deafened before.
1454 let was_deafened = live_kit.deafened;
1455 if live_kit.muted_by_user
1456 || live_kit.deafened
1457 || matches!(live_kit.microphone_track, LocalTrack::None)
1458 {
1459 live_kit.muted_by_user = false;
1460 live_kit.deafened = false;
1461 } else {
1462 live_kit.muted_by_user = true;
1463 }
1464 let muted = live_kit.muted_by_user;
1465 let should_undeafen = was_deafened && !live_kit.deafened;
1466
1467 if let Some(task) = self.set_mute(muted, cx) {
1468 task.detach_and_log_err(cx);
1469 }
1470
1471 if should_undeafen {
1472 self.set_deafened(false, cx);
1473 }
1474 }
1475 }
1476
1477 pub fn toggle_deafen(&mut self, cx: &mut Context<Self>) {
1478 if let Some(live_kit) = self.live_kit.as_mut() {
1479 // When deafening, mute the microphone if it was not already muted.
1480 // When un-deafening, unmute the microphone, unless it was explicitly muted.
1481 let deafened = !live_kit.deafened;
1482 live_kit.deafened = deafened;
1483 let should_change_mute = !live_kit.muted_by_user;
1484
1485 self.set_deafened(deafened, cx);
1486
1487 if should_change_mute {
1488 if let Some(task) = self.set_mute(deafened, cx) {
1489 task.detach_and_log_err(cx);
1490 }
1491 }
1492 }
1493 }
1494
1495 pub fn unshare_screen(&mut self, cx: &mut Context<Self>) -> Result<()> {
1496 anyhow::ensure!(!self.status.is_offline(), "room is offline");
1497
1498 let live_kit = self
1499 .live_kit
1500 .as_mut()
1501 .context("live-kit was not initialized")?;
1502 match mem::take(&mut live_kit.screen_track) {
1503 LocalTrack::None => anyhow::bail!("screen was not shared"),
1504 LocalTrack::Pending { .. } => {
1505 cx.notify();
1506 Ok(())
1507 }
1508 LocalTrack::Published {
1509 track_publication, ..
1510 } => {
1511 {
1512 let local_participant = live_kit.room.local_participant();
1513 let sid = track_publication.sid();
1514 cx.spawn(async move |_, cx| local_participant.unpublish_track(sid, cx).await)
1515 .detach_and_log_err(cx);
1516 cx.notify();
1517 }
1518
1519 Audio::play_sound(Sound::StopScreenshare, cx);
1520 Ok(())
1521 }
1522 }
1523 }
1524
1525 fn set_deafened(&mut self, deafened: bool, cx: &mut Context<Self>) -> Option<()> {
1526 {
1527 let live_kit = self.live_kit.as_mut()?;
1528 cx.notify();
1529 for (_, participant) in live_kit.room.remote_participants() {
1530 for (_, publication) in participant.track_publications() {
1531 if publication.is_audio() {
1532 publication.set_enabled(!deafened, cx);
1533 }
1534 }
1535 }
1536 }
1537
1538 None
1539 }
1540
1541 fn set_mute(&mut self, should_mute: bool, cx: &mut Context<Room>) -> Option<Task<Result<()>>> {
1542 let live_kit = self.live_kit.as_mut()?;
1543 cx.notify();
1544
1545 if should_mute {
1546 Audio::play_sound(Sound::Mute, cx);
1547 } else {
1548 Audio::play_sound(Sound::Unmute, cx);
1549 }
1550
1551 match &mut live_kit.microphone_track {
1552 LocalTrack::None => {
1553 if should_mute {
1554 None
1555 } else {
1556 Some(self.share_microphone(cx))
1557 }
1558 }
1559 LocalTrack::Pending { .. } => None,
1560 LocalTrack::Published {
1561 track_publication, ..
1562 } => {
1563 let guard = Tokio::handle(cx);
1564 if should_mute {
1565 track_publication.mute(cx)
1566 } else {
1567 track_publication.unmute(cx)
1568 }
1569 drop(guard);
1570
1571 None
1572 }
1573 }
1574 }
1575}
1576
1577fn spawn_room_connection(
1578 livekit_connection_info: Option<proto::LiveKitConnectionInfo>,
1579 cx: &mut Context<Room>,
1580) {
1581 if let Some(connection_info) = livekit_connection_info {
1582 cx.spawn(async move |this, cx| {
1583 let (room, mut events) =
1584 livekit::Room::connect(connection_info.server_url, connection_info.token, cx)
1585 .await?;
1586
1587 this.update(cx, |this, cx| {
1588 let _handle_updates = cx.spawn(async move |this, cx| {
1589 while let Some(event) = events.next().await {
1590 if this
1591 .update(cx, |this, cx| {
1592 this.livekit_room_updated(event, cx).warn_on_err();
1593 })
1594 .is_err()
1595 {
1596 break;
1597 }
1598 }
1599 });
1600
1601 let muted_by_user = Room::mute_on_join(cx);
1602 this.live_kit = Some(LiveKitRoom {
1603 room: Rc::new(room),
1604 screen_track: LocalTrack::None,
1605 microphone_track: LocalTrack::None,
1606 next_publish_id: 0,
1607 muted_by_user,
1608 deafened: false,
1609 speaking: false,
1610 _handle_updates,
1611 });
1612
1613 if !muted_by_user && this.can_use_microphone() {
1614 this.share_microphone(cx)
1615 } else {
1616 Task::ready(Ok(()))
1617 }
1618 })?
1619 .await
1620 })
1621 .detach_and_log_err(cx);
1622 }
1623}
1624
1625struct LiveKitRoom {
1626 room: Rc<livekit::Room>,
1627 screen_track: LocalTrack,
1628 microphone_track: LocalTrack,
1629 /// Tracks whether we're currently in a muted state due to auto-mute from deafening or manual mute performed by user.
1630 muted_by_user: bool,
1631 deafened: bool,
1632 speaking: bool,
1633 next_publish_id: usize,
1634 _handle_updates: Task<()>,
1635}
1636
1637impl LiveKitRoom {
1638 fn stop_publishing(&mut self, cx: &mut Context<Room>) {
1639 let mut tracks_to_unpublish = Vec::new();
1640 if let LocalTrack::Published {
1641 track_publication, ..
1642 } = mem::replace(&mut self.microphone_track, LocalTrack::None)
1643 {
1644 tracks_to_unpublish.push(track_publication.sid());
1645 cx.notify();
1646 }
1647
1648 if let LocalTrack::Published {
1649 track_publication, ..
1650 } = mem::replace(&mut self.screen_track, LocalTrack::None)
1651 {
1652 tracks_to_unpublish.push(track_publication.sid());
1653 cx.notify();
1654 }
1655
1656 let participant = self.room.local_participant();
1657 cx.spawn(async move |_, cx| {
1658 for sid in tracks_to_unpublish {
1659 participant.unpublish_track(sid, cx).await.log_err();
1660 }
1661 })
1662 .detach();
1663 }
1664}
1665
1666enum LocalTrack {
1667 None,
1668 Pending {
1669 publish_id: usize,
1670 },
1671 Published {
1672 track_publication: LocalTrackPublication,
1673 _stream: Box<dyn Any>,
1674 },
1675}
1676
1677impl Default for LocalTrack {
1678 fn default() -> Self {
1679 Self::None
1680 }
1681}
1682
1683#[derive(Copy, Clone, PartialEq, Eq)]
1684pub enum RoomStatus {
1685 Online,
1686 Rejoining,
1687 Offline,
1688}
1689
1690impl RoomStatus {
1691 pub fn is_offline(&self) -> bool {
1692 matches!(self, RoomStatus::Offline)
1693 }
1694
1695 pub fn is_online(&self) -> bool {
1696 matches!(self, RoomStatus::Online)
1697 }
1698}