1use crate::{
2 call_settings::CallSettings,
3 participant::{LocalParticipant, ParticipantLocation, RemoteParticipant},
4};
5use anyhow::{anyhow, Result};
6use audio::{Audio, Sound};
7use client::{
8 proto::{self, PeerId},
9 ChannelId, Client, ParticipantIndex, TypedEnvelope, User, UserStore,
10};
11use collections::{BTreeMap, HashMap, HashSet};
12use fs::Fs;
13use futures::{FutureExt, StreamExt};
14use gpui::{
15 AppContext, AsyncAppContext, Context, EventEmitter, Model, ModelContext, Task, WeakModel,
16};
17use language::LanguageRegistry;
18use livekit::{
19 capture_local_audio_track, capture_local_video_track,
20 id::ParticipantIdentity,
21 options::{TrackPublishOptions, VideoCodec},
22 play_remote_audio_track,
23 publication::LocalTrackPublication,
24 track::{TrackKind, TrackSource},
25 RoomEvent, RoomOptions,
26};
27use livekit_client as livekit;
28use postage::{sink::Sink, stream::Stream, watch};
29use project::Project;
30use settings::Settings as _;
31use std::{any::Any, future::Future, mem, sync::Arc, time::Duration};
32use util::{post_inc, ResultExt, TryFutureExt};
33
34pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30);
35
36#[derive(Clone, Debug, PartialEq, Eq)]
37pub enum Event {
38 RoomJoined {
39 channel_id: Option<ChannelId>,
40 },
41 ParticipantLocationChanged {
42 participant_id: proto::PeerId,
43 },
44 RemoteVideoTracksChanged {
45 participant_id: proto::PeerId,
46 },
47 RemoteAudioTracksChanged {
48 participant_id: proto::PeerId,
49 },
50 RemoteProjectShared {
51 owner: Arc<User>,
52 project_id: u64,
53 worktree_root_names: Vec<String>,
54 },
55 RemoteProjectUnshared {
56 project_id: u64,
57 },
58 RemoteProjectJoined {
59 project_id: u64,
60 },
61 RemoteProjectInvitationDiscarded {
62 project_id: u64,
63 },
64 RoomLeft {
65 channel_id: Option<ChannelId>,
66 },
67}
68
69pub struct Room {
70 id: u64,
71 channel_id: Option<ChannelId>,
72 live_kit: Option<LiveKitRoom>,
73 status: RoomStatus,
74 shared_projects: HashSet<WeakModel<Project>>,
75 joined_projects: HashSet<WeakModel<Project>>,
76 local_participant: LocalParticipant,
77 remote_participants: BTreeMap<u64, RemoteParticipant>,
78 pending_participants: Vec<Arc<User>>,
79 participant_user_ids: HashSet<u64>,
80 pending_call_count: usize,
81 leave_when_empty: bool,
82 client: Arc<Client>,
83 user_store: Model<UserStore>,
84 follows_by_leader_id_project_id: HashMap<(PeerId, u64), Vec<PeerId>>,
85 client_subscriptions: Vec<client::Subscription>,
86 _subscriptions: Vec<gpui::Subscription>,
87 room_update_completed_tx: watch::Sender<Option<()>>,
88 room_update_completed_rx: watch::Receiver<Option<()>>,
89 pending_room_update: Option<Task<()>>,
90 maintain_connection: Option<Task<Option<()>>>,
91}
92
93impl EventEmitter<Event> for Room {}
94
95impl Room {
96 pub fn channel_id(&self) -> Option<ChannelId> {
97 self.channel_id
98 }
99
100 pub fn is_sharing_project(&self) -> bool {
101 !self.shared_projects.is_empty()
102 }
103
104 #[cfg(any(test, feature = "test-support"))]
105 pub fn is_connected(&self) -> bool {
106 if let Some(live_kit) = self.live_kit.as_ref() {
107 live_kit.room.connection_state() == livekit::ConnectionState::Connected
108 } else {
109 false
110 }
111 }
112
113 fn new(
114 id: u64,
115 channel_id: Option<ChannelId>,
116 livekit_connection_info: Option<proto::LiveKitConnectionInfo>,
117 client: Arc<Client>,
118 user_store: Model<UserStore>,
119 cx: &mut ModelContext<Self>,
120 ) -> Self {
121 spawn_room_connection(livekit_connection_info, cx);
122
123 let maintain_connection = cx.spawn({
124 let client = client.clone();
125 move |this, cx| Self::maintain_connection(this, client.clone(), cx).log_err()
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_model(), 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<Model<Project>>,
165 client: Arc<Client>,
166 user_store: Model<UserStore>,
167 cx: &mut AppContext,
168 ) -> Task<Result<Model<Self>>> {
169 cx.spawn(move |mut cx| async move {
170 let response = client.request(proto::CreateRoom {}).await?;
171 let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
172 let room = cx.new_model(|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(&mut 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(&mut 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: Model<UserStore>,
215 cx: AsyncAppContext,
216 ) -> Result<Model<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: Model<UserStore>,
233 cx: AsyncAppContext,
234 ) -> Result<Model<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 AppContext) {
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 ModelContext<Self>) -> impl Future<Output = ()> {
250 let task = if self.status.is_online() {
251 let leave = self.leave_internal(cx);
252 Some(cx.background_executor().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: &AppContext) -> 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: Model<UserStore>,
274 mut cx: AsyncAppContext,
275 ) -> Result<Model<Self>> {
276 let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
277 let room = cx.new_model(|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 ModelContext<Self>) -> Task<Result<()>> {
304 cx.notify();
305 self.leave_internal(cx)
306 }
307
308 fn leave_internal(&mut self, cx: &mut AppContext) -> 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_executor().spawn(async move {
320 leave_room.await?;
321 anyhow::Ok(())
322 })
323 }
324
325 pub(crate) fn clear_state(&mut self, cx: &mut AppContext) {
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: WeakModel<Self>,
354 client: Arc<Client>,
355 mut cx: AsyncAppContext,
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 .ok_or_else(|| anyhow!("room was dropped"))?
367 .update(&mut 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(&mut 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(&mut cx, |this, cx| this.leave(cx))?.await?;
433 }
434 Err(anyhow!(
435 "can't reconnect to room: client failed to re-establish connection"
436 ))
437 }
438
439 fn rejoin(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
440 let mut projects = HashMap::default();
441 let mut reshared_projects = Vec::new();
442 let mut rejoined_projects = Vec::new();
443 self.shared_projects.retain(|project| {
444 if let Some(handle) = project.upgrade() {
445 let project = handle.read(cx);
446 if let Some(project_id) = project.remote_id() {
447 projects.insert(project_id, handle.clone());
448 reshared_projects.push(proto::UpdateProject {
449 project_id,
450 worktrees: project.worktree_metadata_protos(cx),
451 });
452 return true;
453 }
454 }
455 false
456 });
457 self.joined_projects.retain(|project| {
458 if let Some(handle) = project.upgrade() {
459 let project = handle.read(cx);
460 if let Some(project_id) = project.remote_id() {
461 projects.insert(project_id, handle.clone());
462 rejoined_projects.push(proto::RejoinProject {
463 id: project_id,
464 worktrees: project
465 .worktrees(cx)
466 .map(|worktree| {
467 let worktree = worktree.read(cx);
468 proto::RejoinWorktree {
469 id: worktree.id().to_proto(),
470 scan_id: worktree.completed_scan_id() as u64,
471 }
472 })
473 .collect(),
474 });
475 }
476 return true;
477 }
478 false
479 });
480
481 let response = self.client.request_envelope(proto::RejoinRoom {
482 id: self.id,
483 reshared_projects,
484 rejoined_projects,
485 });
486
487 cx.spawn(|this, mut cx| async move {
488 let response = response.await?;
489 let message_id = response.message_id;
490 let response = response.payload;
491 let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?;
492 this.update(&mut cx, |this, cx| {
493 this.status = RoomStatus::Online;
494 this.apply_room_update(room_proto, cx)?;
495
496 for reshared_project in response.reshared_projects {
497 if let Some(project) = projects.get(&reshared_project.id) {
498 project.update(cx, |project, cx| {
499 project.reshared(reshared_project, cx).log_err();
500 });
501 }
502 }
503
504 for rejoined_project in response.rejoined_projects {
505 if let Some(project) = projects.get(&rejoined_project.id) {
506 project.update(cx, |project, cx| {
507 project.rejoined(rejoined_project, message_id, cx).log_err();
508 });
509 }
510 }
511
512 anyhow::Ok(())
513 })?
514 })
515 }
516
517 pub fn id(&self) -> u64 {
518 self.id
519 }
520
521 pub fn status(&self) -> RoomStatus {
522 self.status
523 }
524
525 pub fn local_participant(&self) -> &LocalParticipant {
526 &self.local_participant
527 }
528
529 pub fn remote_participants(&self) -> &BTreeMap<u64, RemoteParticipant> {
530 &self.remote_participants
531 }
532
533 pub fn remote_participant_for_peer_id(&self, peer_id: PeerId) -> Option<&RemoteParticipant> {
534 self.remote_participants
535 .values()
536 .find(|p| p.peer_id == peer_id)
537 }
538
539 pub fn role_for_user(&self, user_id: u64) -> Option<proto::ChannelRole> {
540 self.remote_participants
541 .get(&user_id)
542 .map(|participant| participant.role)
543 }
544
545 pub fn contains_guests(&self) -> bool {
546 self.local_participant.role == proto::ChannelRole::Guest
547 || self
548 .remote_participants
549 .values()
550 .any(|p| p.role == proto::ChannelRole::Guest)
551 }
552
553 pub fn local_participant_is_admin(&self) -> bool {
554 self.local_participant.role == proto::ChannelRole::Admin
555 }
556
557 pub fn local_participant_is_guest(&self) -> bool {
558 self.local_participant.role == proto::ChannelRole::Guest
559 }
560
561 pub fn set_participant_role(
562 &mut self,
563 user_id: u64,
564 role: proto::ChannelRole,
565 cx: &ModelContext<Self>,
566 ) -> Task<Result<()>> {
567 let client = self.client.clone();
568 let room_id = self.id;
569 let role = role.into();
570 cx.spawn(|_, _| async move {
571 client
572 .request(proto::SetRoomParticipantRole {
573 room_id,
574 user_id,
575 role,
576 })
577 .await
578 .map(|_| ())
579 })
580 }
581
582 pub fn pending_participants(&self) -> &[Arc<User>] {
583 &self.pending_participants
584 }
585
586 pub fn contains_participant(&self, user_id: u64) -> bool {
587 self.participant_user_ids.contains(&user_id)
588 }
589
590 pub fn followers_for(&self, leader_id: PeerId, project_id: u64) -> &[PeerId] {
591 self.follows_by_leader_id_project_id
592 .get(&(leader_id, project_id))
593 .map_or(&[], |v| v.as_slice())
594 }
595
596 /// Returns the most 'active' projects, defined as most people in the project
597 pub fn most_active_project(&self, cx: &AppContext) -> Option<(u64, u64)> {
598 let mut project_hosts_and_guest_counts = HashMap::<u64, (Option<u64>, u32)>::default();
599 for participant in self.remote_participants.values() {
600 match participant.location {
601 ParticipantLocation::SharedProject { project_id } => {
602 project_hosts_and_guest_counts
603 .entry(project_id)
604 .or_default()
605 .1 += 1;
606 }
607 ParticipantLocation::External | ParticipantLocation::UnsharedProject => {}
608 }
609 for project in &participant.projects {
610 project_hosts_and_guest_counts
611 .entry(project.id)
612 .or_default()
613 .0 = Some(participant.user.id);
614 }
615 }
616
617 if let Some(user) = self.user_store.read(cx).current_user() {
618 for project in &self.local_participant.projects {
619 project_hosts_and_guest_counts
620 .entry(project.id)
621 .or_default()
622 .0 = Some(user.id);
623 }
624 }
625
626 project_hosts_and_guest_counts
627 .into_iter()
628 .filter_map(|(id, (host, guest_count))| Some((id, host?, guest_count)))
629 .max_by_key(|(_, _, guest_count)| *guest_count)
630 .map(|(id, host, _)| (id, host))
631 }
632
633 async fn handle_room_updated(
634 this: Model<Self>,
635 envelope: TypedEnvelope<proto::RoomUpdated>,
636 mut cx: AsyncAppContext,
637 ) -> Result<()> {
638 let room = envelope
639 .payload
640 .room
641 .ok_or_else(|| anyhow!("invalid room"))?;
642 this.update(&mut cx, |this, cx| this.apply_room_update(room, cx))?
643 }
644
645 fn apply_room_update(&mut self, room: proto::Room, cx: &mut ModelContext<Self>) -> Result<()> {
646 log::trace!(
647 "client {:?}. room update: {:?}",
648 self.client.user_id(),
649 &room
650 );
651
652 self.pending_room_update = Some(self.start_room_connection(room, cx));
653
654 cx.notify();
655 Ok(())
656 }
657
658 pub fn room_update_completed(&mut self) -> impl Future<Output = ()> {
659 let mut done_rx = self.room_update_completed_rx.clone();
660 async move {
661 while let Some(result) = done_rx.next().await {
662 if result.is_some() {
663 break;
664 }
665 }
666 }
667 }
668
669 fn start_room_connection(
670 &self,
671 mut room: proto::Room,
672 cx: &mut ModelContext<Self>,
673 ) -> 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(|this, mut cx| async move {
701 let (remote_participants, pending_participants) =
702 futures::join!(remote_participants, pending_participants);
703
704 this.update(&mut 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(
920 &mut self,
921 event: RoomEvent,
922 cx: &mut ModelContext<Self>,
923 ) -> 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 = self.remote_participants.get_mut(&user_id).ok_or_else(|| {
939 anyhow!(
940 "{:?} subscribed to track by unknown participant {user_id}",
941 self.client.user_id()
942 )
943 })?;
944 if self.live_kit.as_ref().map_or(true, |kit| kit.deafened) {
945 track.rtc_track().set_enabled(false);
946 }
947 match track {
948 livekit::track::RemoteTrack::Audio(track) => {
949 cx.emit(Event::RemoteAudioTracksChanged {
950 participant_id: participant.peer_id,
951 });
952 let stream = play_remote_audio_track(&track, cx.background_executor())?;
953 participant.audio_tracks.insert(track_id, (track, stream));
954 participant.muted = publication.is_muted();
955 }
956 livekit::track::RemoteTrack::Video(track) => {
957 cx.emit(Event::RemoteVideoTracksChanged {
958 participant_id: participant.peer_id,
959 });
960 participant.video_tracks.insert(track_id, track);
961 }
962 }
963 }
964
965 RoomEvent::TrackUnsubscribed {
966 track, participant, ..
967 } => {
968 let user_id = participant.identity().0.parse()?;
969 let participant = self.remote_participants.get_mut(&user_id).ok_or_else(|| {
970 anyhow!(
971 "{:?}, unsubscribed from track by unknown participant {user_id}",
972 self.client.user_id()
973 )
974 })?;
975 match track {
976 livekit::track::RemoteTrack::Audio(track) => {
977 participant.audio_tracks.remove(&track.sid());
978 participant.muted = true;
979 cx.emit(Event::RemoteAudioTracksChanged {
980 participant_id: participant.peer_id,
981 });
982 }
983 livekit::track::RemoteTrack::Video(track) => {
984 participant.video_tracks.remove(&track.sid());
985 cx.emit(Event::RemoteVideoTracksChanged {
986 participant_id: participant.peer_id,
987 });
988 }
989 }
990 }
991
992 RoomEvent::ActiveSpeakersChanged { speakers } => {
993 let mut speaker_ids = speakers
994 .into_iter()
995 .filter_map(|speaker| speaker.identity().0.parse().ok())
996 .collect::<Vec<u64>>();
997 speaker_ids.sort_unstable();
998 for (sid, participant) in &mut self.remote_participants {
999 participant.speaking = speaker_ids.binary_search(sid).is_ok();
1000 }
1001 if let Some(id) = self.client.user_id() {
1002 if let Some(room) = &mut self.live_kit {
1003 room.speaking = speaker_ids.binary_search(&id).is_ok();
1004 }
1005 }
1006 }
1007
1008 RoomEvent::TrackMuted {
1009 participant,
1010 publication,
1011 }
1012 | RoomEvent::TrackUnmuted {
1013 participant,
1014 publication,
1015 } => {
1016 let mut found = false;
1017 let user_id = participant.identity().0.parse()?;
1018 let track_id = publication.sid();
1019 if let Some(participant) = self.remote_participants.get_mut(&user_id) {
1020 for (track, _) in participant.audio_tracks.values() {
1021 if track.sid() == track_id {
1022 found = true;
1023 break;
1024 }
1025 }
1026 if found {
1027 participant.muted = publication.is_muted();
1028 }
1029 }
1030 }
1031
1032 RoomEvent::LocalTrackUnpublished { publication, .. } => {
1033 log::info!("unpublished track {}", publication.sid());
1034 if let Some(room) = &mut self.live_kit {
1035 if let LocalTrack::Published {
1036 track_publication, ..
1037 } = &room.microphone_track
1038 {
1039 if track_publication.sid() == publication.sid() {
1040 room.microphone_track = LocalTrack::None;
1041 }
1042 }
1043 if let LocalTrack::Published {
1044 track_publication, ..
1045 } = &room.screen_track
1046 {
1047 if track_publication.sid() == publication.sid() {
1048 room.screen_track = LocalTrack::None;
1049 }
1050 }
1051 }
1052 }
1053
1054 RoomEvent::LocalTrackPublished { publication, .. } => {
1055 log::info!("published track {:?}", publication.sid());
1056 }
1057
1058 RoomEvent::Disconnected { reason } => {
1059 log::info!("disconnected from room: {reason:?}");
1060 self.leave(cx).detach_and_log_err(cx);
1061 }
1062 _ => {}
1063 }
1064
1065 cx.notify();
1066 Ok(())
1067 }
1068
1069 fn check_invariants(&self) {
1070 #[cfg(any(test, feature = "test-support"))]
1071 {
1072 for participant in self.remote_participants.values() {
1073 assert!(self.participant_user_ids.contains(&participant.user.id));
1074 assert_ne!(participant.user.id, self.client.user_id().unwrap());
1075 }
1076
1077 for participant in &self.pending_participants {
1078 assert!(self.participant_user_ids.contains(&participant.id));
1079 assert_ne!(participant.id, self.client.user_id().unwrap());
1080 }
1081
1082 assert_eq!(
1083 self.participant_user_ids.len(),
1084 self.remote_participants.len() + self.pending_participants.len()
1085 );
1086 }
1087 }
1088
1089 pub(crate) fn call(
1090 &mut self,
1091 called_user_id: u64,
1092 initial_project_id: Option<u64>,
1093 cx: &mut ModelContext<Self>,
1094 ) -> Task<Result<()>> {
1095 if self.status.is_offline() {
1096 return Task::ready(Err(anyhow!("room is offline")));
1097 }
1098
1099 cx.notify();
1100 let client = self.client.clone();
1101 let room_id = self.id;
1102 self.pending_call_count += 1;
1103 cx.spawn(move |this, mut cx| async move {
1104 let result = client
1105 .request(proto::Call {
1106 room_id,
1107 called_user_id,
1108 initial_project_id,
1109 })
1110 .await;
1111 this.update(&mut cx, |this, cx| {
1112 this.pending_call_count -= 1;
1113 if this.should_leave() {
1114 this.leave(cx).detach_and_log_err(cx);
1115 }
1116 })?;
1117 result?;
1118 Ok(())
1119 })
1120 }
1121
1122 pub fn join_project(
1123 &mut self,
1124 id: u64,
1125 language_registry: Arc<LanguageRegistry>,
1126 fs: Arc<dyn Fs>,
1127 cx: &mut ModelContext<Self>,
1128 ) -> Task<Result<Model<Project>>> {
1129 let client = self.client.clone();
1130 let user_store = self.user_store.clone();
1131 cx.emit(Event::RemoteProjectJoined { project_id: id });
1132 cx.spawn(move |this, mut cx| async move {
1133 let project =
1134 Project::in_room(id, client, user_store, language_registry, fs, cx.clone()).await?;
1135
1136 this.update(&mut cx, |this, cx| {
1137 this.joined_projects.retain(|project| {
1138 if let Some(project) = project.upgrade() {
1139 !project.read(cx).is_disconnected(cx)
1140 } else {
1141 false
1142 }
1143 });
1144 this.joined_projects.insert(project.downgrade());
1145 })?;
1146 Ok(project)
1147 })
1148 }
1149
1150 pub fn share_project(
1151 &mut self,
1152 project: Model<Project>,
1153 cx: &mut ModelContext<Self>,
1154 ) -> Task<Result<u64>> {
1155 if let Some(project_id) = project.read(cx).remote_id() {
1156 return Task::ready(Ok(project_id));
1157 }
1158
1159 let request = self.client.request(proto::ShareProject {
1160 room_id: self.id(),
1161 worktrees: project.read(cx).worktree_metadata_protos(cx),
1162 is_ssh_project: project.read(cx).is_via_ssh(),
1163 });
1164
1165 cx.spawn(|this, mut cx| async move {
1166 let response = request.await?;
1167
1168 project.update(&mut cx, |project, cx| {
1169 project.shared(response.project_id, cx)
1170 })??;
1171
1172 // If the user's location is in this project, it changes from UnsharedProject to SharedProject.
1173 this.update(&mut cx, |this, cx| {
1174 this.shared_projects.insert(project.downgrade());
1175 let active_project = this.local_participant.active_project.as_ref();
1176 if active_project.map_or(false, |location| *location == project) {
1177 this.set_location(Some(&project), cx)
1178 } else {
1179 Task::ready(Ok(()))
1180 }
1181 })?
1182 .await?;
1183
1184 Ok(response.project_id)
1185 })
1186 }
1187
1188 pub(crate) fn unshare_project(
1189 &mut self,
1190 project: Model<Project>,
1191 cx: &mut ModelContext<Self>,
1192 ) -> Result<()> {
1193 let project_id = match project.read(cx).remote_id() {
1194 Some(project_id) => project_id,
1195 None => return Ok(()),
1196 };
1197
1198 self.client.send(proto::UnshareProject { project_id })?;
1199 project.update(cx, |this, cx| this.unshare(cx))?;
1200
1201 if self.local_participant.active_project == Some(project.downgrade()) {
1202 self.set_location(Some(&project), cx).detach_and_log_err(cx);
1203 }
1204 Ok(())
1205 }
1206
1207 pub(crate) fn set_location(
1208 &mut self,
1209 project: Option<&Model<Project>>,
1210 cx: &mut ModelContext<Self>,
1211 ) -> Task<Result<()>> {
1212 if self.status.is_offline() {
1213 return Task::ready(Err(anyhow!("room is offline")));
1214 }
1215
1216 let client = self.client.clone();
1217 let room_id = self.id;
1218 let location = if let Some(project) = project {
1219 self.local_participant.active_project = Some(project.downgrade());
1220 if let Some(project_id) = project.read(cx).remote_id() {
1221 proto::participant_location::Variant::SharedProject(
1222 proto::participant_location::SharedProject { id: project_id },
1223 )
1224 } else {
1225 proto::participant_location::Variant::UnsharedProject(
1226 proto::participant_location::UnsharedProject {},
1227 )
1228 }
1229 } else {
1230 self.local_participant.active_project = None;
1231 proto::participant_location::Variant::External(proto::participant_location::External {})
1232 };
1233
1234 cx.notify();
1235 cx.background_executor().spawn(async move {
1236 client
1237 .request(proto::UpdateParticipantLocation {
1238 room_id,
1239 location: Some(proto::ParticipantLocation {
1240 variant: Some(location),
1241 }),
1242 })
1243 .await?;
1244 Ok(())
1245 })
1246 }
1247
1248 pub fn is_screen_sharing(&self) -> bool {
1249 self.live_kit.as_ref().map_or(false, |live_kit| {
1250 !matches!(live_kit.screen_track, LocalTrack::None)
1251 })
1252 }
1253
1254 pub fn is_sharing_mic(&self) -> bool {
1255 self.live_kit.as_ref().map_or(false, |live_kit| {
1256 !matches!(live_kit.microphone_track, LocalTrack::None)
1257 })
1258 }
1259
1260 pub fn is_muted(&self) -> bool {
1261 self.live_kit.as_ref().map_or(false, |live_kit| {
1262 matches!(live_kit.microphone_track, LocalTrack::None)
1263 || live_kit.muted_by_user
1264 || live_kit.deafened
1265 })
1266 }
1267
1268 pub fn muted_by_user(&self) -> bool {
1269 self.live_kit
1270 .as_ref()
1271 .map_or(false, |live_kit| live_kit.muted_by_user)
1272 }
1273
1274 pub fn is_speaking(&self) -> bool {
1275 self.live_kit
1276 .as_ref()
1277 .map_or(false, |live_kit| live_kit.speaking)
1278 }
1279
1280 pub fn is_deafened(&self) -> Option<bool> {
1281 self.live_kit.as_ref().map(|live_kit| live_kit.deafened)
1282 }
1283
1284 pub fn can_use_microphone(&self) -> bool {
1285 use proto::ChannelRole::*;
1286
1287 match self.local_participant.role {
1288 Admin | Member | Talker => true,
1289 Guest | Banned => false,
1290 }
1291 }
1292
1293 pub fn can_share_projects(&self) -> bool {
1294 use proto::ChannelRole::*;
1295 match self.local_participant.role {
1296 Admin | Member => true,
1297 Guest | Banned | Talker => false,
1298 }
1299 }
1300
1301 #[track_caller]
1302 pub fn share_microphone(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
1303 if self.status.is_offline() {
1304 return Task::ready(Err(anyhow!("room is offline")));
1305 }
1306
1307 let (participant, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
1308 let publish_id = post_inc(&mut live_kit.next_publish_id);
1309 live_kit.microphone_track = LocalTrack::Pending { publish_id };
1310 cx.notify();
1311 (live_kit.room.local_participant(), publish_id)
1312 } else {
1313 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1314 };
1315
1316 cx.spawn(move |this, mut cx| async move {
1317 let (track, stream) = capture_local_audio_track(cx.background_executor())?.await;
1318
1319 let publication = participant
1320 .publish_track(
1321 livekit::track::LocalTrack::Audio(track),
1322 TrackPublishOptions {
1323 source: TrackSource::Microphone,
1324 ..Default::default()
1325 },
1326 )
1327 .await
1328 .map_err(|error| anyhow!("failed to publish track: {error}"));
1329 this.update(&mut cx, |this, cx| {
1330 let live_kit = this
1331 .live_kit
1332 .as_mut()
1333 .ok_or_else(|| anyhow!("live-kit was not initialized"))?;
1334
1335 let canceled = if let LocalTrack::Pending {
1336 publish_id: cur_publish_id,
1337 } = &live_kit.microphone_track
1338 {
1339 *cur_publish_id != publish_id
1340 } else {
1341 true
1342 };
1343
1344 match publication {
1345 Ok(publication) => {
1346 if canceled {
1347 cx.background_executor()
1348 .spawn(async move {
1349 participant.unpublish_track(&publication.sid()).await
1350 })
1351 .detach_and_log_err(cx)
1352 } else {
1353 if live_kit.muted_by_user || live_kit.deafened {
1354 publication.mute();
1355 }
1356 live_kit.microphone_track = LocalTrack::Published {
1357 track_publication: publication,
1358 _stream: Box::new(stream),
1359 };
1360 cx.notify();
1361 }
1362 Ok(())
1363 }
1364 Err(error) => {
1365 if canceled {
1366 Ok(())
1367 } else {
1368 live_kit.microphone_track = LocalTrack::None;
1369 cx.notify();
1370 Err(error)
1371 }
1372 }
1373 }
1374 })?
1375 })
1376 }
1377
1378 pub fn share_screen(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> {
1379 if self.status.is_offline() {
1380 return Task::ready(Err(anyhow!("room is offline")));
1381 }
1382 if self.is_screen_sharing() {
1383 return Task::ready(Err(anyhow!("screen was already shared")));
1384 }
1385
1386 let (participant, publish_id) = if let Some(live_kit) = self.live_kit.as_mut() {
1387 let publish_id = post_inc(&mut live_kit.next_publish_id);
1388 live_kit.screen_track = LocalTrack::Pending { publish_id };
1389 cx.notify();
1390 (live_kit.room.local_participant(), publish_id)
1391 } else {
1392 return Task::ready(Err(anyhow!("live-kit was not initialized")));
1393 };
1394
1395 let sources = cx.screen_capture_sources();
1396
1397 cx.spawn(move |this, mut cx| async move {
1398 let sources = sources.await??;
1399 let source = sources.first().ok_or_else(|| anyhow!("no display found"))?;
1400
1401 let (track, stream) = capture_local_video_track(&**source).await?;
1402
1403 let publication = participant
1404 .publish_track(
1405 livekit::track::LocalTrack::Video(track),
1406 TrackPublishOptions {
1407 source: TrackSource::Screenshare,
1408 video_codec: VideoCodec::H264,
1409 ..Default::default()
1410 },
1411 )
1412 .await
1413 .map_err(|error| anyhow!("error publishing screen track {error:?}"));
1414
1415 this.update(&mut cx, |this, cx| {
1416 let live_kit = this
1417 .live_kit
1418 .as_mut()
1419 .ok_or_else(|| anyhow!("live-kit was not initialized"))?;
1420
1421 let canceled = if let LocalTrack::Pending {
1422 publish_id: cur_publish_id,
1423 } = &live_kit.screen_track
1424 {
1425 *cur_publish_id != publish_id
1426 } else {
1427 true
1428 };
1429
1430 match publication {
1431 Ok(publication) => {
1432 if canceled {
1433 cx.background_executor()
1434 .spawn(async move {
1435 participant.unpublish_track(&publication.sid()).await
1436 })
1437 .detach()
1438 } else {
1439 live_kit.screen_track = LocalTrack::Published {
1440 track_publication: publication,
1441 _stream: Box::new(stream),
1442 };
1443 cx.notify();
1444 }
1445
1446 Audio::play_sound(Sound::StartScreenshare, cx);
1447 Ok(())
1448 }
1449 Err(error) => {
1450 if canceled {
1451 Ok(())
1452 } else {
1453 live_kit.screen_track = LocalTrack::None;
1454 cx.notify();
1455 Err(error)
1456 }
1457 }
1458 }
1459 })?
1460 })
1461 }
1462
1463 pub fn toggle_mute(&mut self, cx: &mut ModelContext<Self>) {
1464 if let Some(live_kit) = self.live_kit.as_mut() {
1465 // When unmuting, undeafen if the user was deafened before.
1466 let was_deafened = live_kit.deafened;
1467 if live_kit.muted_by_user
1468 || live_kit.deafened
1469 || matches!(live_kit.microphone_track, LocalTrack::None)
1470 {
1471 live_kit.muted_by_user = false;
1472 live_kit.deafened = false;
1473 } else {
1474 live_kit.muted_by_user = true;
1475 }
1476 let muted = live_kit.muted_by_user;
1477 let should_undeafen = was_deafened && !live_kit.deafened;
1478
1479 if let Some(task) = self.set_mute(muted, cx) {
1480 task.detach_and_log_err(cx);
1481 }
1482
1483 if should_undeafen {
1484 self.set_deafened(false, cx);
1485 }
1486 }
1487 }
1488
1489 pub fn toggle_deafen(&mut self, cx: &mut ModelContext<Self>) {
1490 if let Some(live_kit) = self.live_kit.as_mut() {
1491 // When deafening, mute the microphone if it was not already muted.
1492 // When un-deafening, unmute the microphone, unless it was explicitly muted.
1493 let deafened = !live_kit.deafened;
1494 live_kit.deafened = deafened;
1495 let should_change_mute = !live_kit.muted_by_user;
1496
1497 self.set_deafened(deafened, cx);
1498
1499 if should_change_mute {
1500 if let Some(task) = self.set_mute(deafened, cx) {
1501 task.detach_and_log_err(cx);
1502 }
1503 }
1504 }
1505 }
1506
1507 pub fn unshare_screen(&mut self, cx: &mut ModelContext<Self>) -> Result<()> {
1508 if self.status.is_offline() {
1509 return Err(anyhow!("room is offline"));
1510 }
1511
1512 let live_kit = self
1513 .live_kit
1514 .as_mut()
1515 .ok_or_else(|| anyhow!("live-kit was not initialized"))?;
1516 match mem::take(&mut live_kit.screen_track) {
1517 LocalTrack::None => Err(anyhow!("screen was not shared")),
1518 LocalTrack::Pending { .. } => {
1519 cx.notify();
1520 Ok(())
1521 }
1522 LocalTrack::Published {
1523 track_publication, ..
1524 } => {
1525 let local_participant = live_kit.room.local_participant();
1526 let sid = track_publication.sid();
1527 cx.background_executor()
1528 .spawn(async move { local_participant.unpublish_track(&sid).await })
1529 .detach_and_log_err(cx);
1530 cx.notify();
1531
1532 Audio::play_sound(Sound::StopScreenshare, cx);
1533 Ok(())
1534 }
1535 }
1536 }
1537
1538 fn set_deafened(&mut self, deafened: bool, cx: &mut ModelContext<Self>) -> Option<()> {
1539 let live_kit = self.live_kit.as_mut()?;
1540 cx.notify();
1541 for (_, participant) in live_kit.room.remote_participants() {
1542 for (_, publication) in participant.track_publications() {
1543 if publication.kind() == TrackKind::Audio {
1544 publication.set_enabled(!deafened);
1545 }
1546 }
1547 }
1548
1549 None
1550 }
1551
1552 fn set_mute(
1553 &mut self,
1554 should_mute: bool,
1555 cx: &mut ModelContext<Room>,
1556 ) -> Option<Task<Result<()>>> {
1557 let live_kit = self.live_kit.as_mut()?;
1558 cx.notify();
1559
1560 if should_mute {
1561 Audio::play_sound(Sound::Mute, cx);
1562 } else {
1563 Audio::play_sound(Sound::Unmute, cx);
1564 }
1565
1566 match &mut live_kit.microphone_track {
1567 LocalTrack::None => {
1568 if should_mute {
1569 None
1570 } else {
1571 Some(self.share_microphone(cx))
1572 }
1573 }
1574 LocalTrack::Pending { .. } => None,
1575 LocalTrack::Published {
1576 track_publication, ..
1577 } => {
1578 if should_mute {
1579 track_publication.mute()
1580 } else {
1581 track_publication.unmute()
1582 }
1583
1584 None
1585 }
1586 }
1587 }
1588}
1589
1590fn spawn_room_connection(
1591 livekit_connection_info: Option<proto::LiveKitConnectionInfo>,
1592 cx: &mut ModelContext<'_, Room>,
1593) {
1594 if let Some(connection_info) = livekit_connection_info {
1595 cx.spawn(|this, mut cx| async move {
1596 let (room, mut events) = livekit::Room::connect(
1597 &connection_info.server_url,
1598 &connection_info.token,
1599 RoomOptions::default(),
1600 )
1601 .await?;
1602
1603 this.update(&mut cx, |this, cx| {
1604 let _handle_updates = cx.spawn(|this, mut cx| async move {
1605 while let Some(event) = events.recv().await {
1606 if this
1607 .update(&mut cx, |this, cx| {
1608 this.livekit_room_updated(event, cx).warn_on_err();
1609 })
1610 .is_err()
1611 {
1612 break;
1613 }
1614 }
1615 });
1616
1617 let muted_by_user = Room::mute_on_join(cx);
1618 this.live_kit = Some(LiveKitRoom {
1619 room: Arc::new(room),
1620 screen_track: LocalTrack::None,
1621 microphone_track: LocalTrack::None,
1622 next_publish_id: 0,
1623 muted_by_user,
1624 deafened: false,
1625 speaking: false,
1626 _handle_updates,
1627 });
1628
1629 if !muted_by_user && this.can_use_microphone() {
1630 this.share_microphone(cx)
1631 } else {
1632 Task::ready(Ok(()))
1633 }
1634 })?
1635 .await
1636 })
1637 .detach_and_log_err(cx);
1638 }
1639}
1640
1641struct LiveKitRoom {
1642 room: Arc<livekit::Room>,
1643 screen_track: LocalTrack,
1644 microphone_track: LocalTrack,
1645 /// Tracks whether we're currently in a muted state due to auto-mute from deafening or manual mute performed by user.
1646 muted_by_user: bool,
1647 deafened: bool,
1648 speaking: bool,
1649 next_publish_id: usize,
1650 _handle_updates: Task<()>,
1651}
1652
1653impl LiveKitRoom {
1654 fn stop_publishing(&mut self, cx: &mut ModelContext<Room>) {
1655 let mut tracks_to_unpublish = Vec::new();
1656 if let LocalTrack::Published {
1657 track_publication, ..
1658 } = mem::replace(&mut self.microphone_track, LocalTrack::None)
1659 {
1660 tracks_to_unpublish.push(track_publication.sid());
1661 cx.notify();
1662 }
1663
1664 if let LocalTrack::Published {
1665 track_publication, ..
1666 } = mem::replace(&mut self.screen_track, LocalTrack::None)
1667 {
1668 tracks_to_unpublish.push(track_publication.sid());
1669 cx.notify();
1670 }
1671
1672 let participant = self.room.local_participant();
1673 cx.background_executor()
1674 .spawn(async move {
1675 for sid in tracks_to_unpublish {
1676 participant.unpublish_track(&sid).await.log_err();
1677 }
1678 })
1679 .detach();
1680 }
1681}
1682
1683enum LocalTrack {
1684 None,
1685 Pending {
1686 publish_id: usize,
1687 },
1688 Published {
1689 track_publication: LocalTrackPublication,
1690 _stream: Box<dyn Any>,
1691 },
1692}
1693
1694impl Default for LocalTrack {
1695 fn default() -> Self {
1696 Self::None
1697 }
1698}
1699
1700#[derive(Copy, Clone, PartialEq, Eq)]
1701pub enum RoomStatus {
1702 Online,
1703 Rejoining,
1704 Offline,
1705}
1706
1707impl RoomStatus {
1708 pub fn is_offline(&self) -> bool {
1709 matches!(self, RoomStatus::Offline)
1710 }
1711
1712 pub fn is_online(&self) -> bool {
1713 matches!(self, RoomStatus::Online)
1714 }
1715}