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