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