#![cfg_attr(target_os = "windows", allow(unused))] use crate::{ call_settings::CallSettings, participant::{LocalParticipant, ParticipantLocation, RemoteParticipant},};use anyhow::{anyhow, Result};use audio::{Audio, Sound};use client::{ proto::{self, PeerId}, ChannelId, Client, ParticipantIndex, TypedEnvelope, User, UserStore,};use collections::{BTreeMap, HashMap, HashSet};use fs::Fs;use futures::{FutureExt, StreamExt};use gpui::{ AppContext, AsyncAppContext, Context, EventEmitter, Model, ModelContext, Task, WeakModel,};use language::LanguageRegistry;#[cfg(not(target_os = "windows"))]use livekit::{ capture_local_audio_track, capture_local_video_track, id::ParticipantIdentity, options::{TrackPublishOptions, VideoCodec}, play_remote_audio_track, publication::LocalTrackPublication, track::{TrackKind, TrackSource}, RoomEvent, RoomOptions,};#[cfg(target_os = "windows")]use livekit::{publication::LocalTrackPublication, RoomEvent};use livekit_client as livekit;use postage::{sink::Sink, stream::Stream, watch};use project::Project;use settings::Settings as _;use std::{any::Any, future::Future, mem, sync::Arc, time::Duration};use util::{post_inc, ResultExt, TryFutureExt}; pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30); #[derive(Clone, Debug, PartialEq, Eq)]pub enum Event { RoomJoined { channel_id: Option<ChannelId>, }, ParticipantLocationChanged { participant_id: proto::PeerId, }, RemoteVideoTracksChanged { participant_id: proto::PeerId, }, RemoteAudioTracksChanged { participant_id: proto::PeerId, }, RemoteProjectShared { owner: Arc<User>, project_id: u64, worktree_root_names: Vec<String>, }, RemoteProjectUnshared { project_id: u64, }, RemoteProjectJoined { project_id: u64, }, RemoteProjectInvitationDiscarded { project_id: u64, }, RoomLeft { channel_id: Option<ChannelId>, },} pub struct Room { id: u64, channel_id: Option<ChannelId>, live_kit: Option<LiveKitRoom>, status: RoomStatus, shared_projects: HashSet<WeakModel<Project>>, joined_projects: HashSet<WeakModel<Project>>, local_participant: LocalParticipant, remote_participants: BTreeMap<u64, RemoteParticipant>, pending_participants: Vec<Arc<User>>, participant_user_ids: HashSet<u64>, pending_call_count: usize, leave_when_empty: bool, client: Arc<Client>, user_store: Model<UserStore>, follows_by_leader_id_project_id: HashMap<(PeerId, u64), Vec<PeerId>>, client_subscriptions: Vec<client::Subscription>, _subscriptions: Vec<gpui::Subscription>, room_update_completed_tx: watch::Sender<Option<()>>, room_update_completed_rx: watch::Receiver<Option<()>>, pending_room_update: Option<Task<()>>, maintain_connection: Option<Task<Option<()>>>,} impl EventEmitter<Event> for Room {} impl Room { pub fn channel_id(&self) -> Option<ChannelId> { self.channel_id } pub fn is_sharing_project(&self) -> bool { !self.shared_projects.is_empty() } #[cfg(all(any(test, feature = "test-support"), not(target_os = "windows")))] pub fn is_connected(&self) -> bool { if let Some(live_kit) = self.live_kit.as_ref() { live_kit.room.connection_state() == livekit::ConnectionState::Connected } else { false } } fn new( id: u64, channel_id: Option<ChannelId>, livekit_connection_info: Option<proto::LiveKitConnectionInfo>, client: Arc<Client>, user_store: Model<UserStore>, cx: &mut ModelContext<Self>, ) -> Self { spawn_room_connection(livekit_connection_info, cx); let maintain_connection = cx.spawn({ let client = client.clone(); move |this, cx| Self::maintain_connection(this, client.clone(), cx).log_err() }); Audio::play_sound(Sound::Joined, cx); let (room_update_completed_tx, room_update_completed_rx) = watch::channel(); Self { id, channel_id, live_kit: None, status: RoomStatus::Online, shared_projects: Default::default(), joined_projects: Default::default(), participant_user_ids: Default::default(), local_participant: Default::default(), remote_participants: Default::default(), pending_participants: Default::default(), pending_call_count: 0, client_subscriptions: vec![ client.add_message_handler(cx.weak_model(), Self::handle_room_updated) ], _subscriptions: vec![ cx.on_release(Self::released), cx.on_app_quit(Self::app_will_quit), ], leave_when_empty: false, pending_room_update: None, client, user_store, follows_by_leader_id_project_id: Default::default(), maintain_connection: Some(maintain_connection), room_update_completed_tx, room_update_completed_rx, } } pub(crate) fn create( called_user_id: u64, initial_project: Option<Model<Project>>, client: Arc<Client>, user_store: Model<UserStore>, cx: &mut AppContext, ) -> Task<Result<Model<Self>>> { cx.spawn(move |mut cx| async move { let response = client.request(proto::CreateRoom {}).await?; let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?; let room = cx.new_model(|cx| { let mut room = Self::new( room_proto.id, None, response.live_kit_connection_info, client, user_store, cx, ); if let Some(participant) = room_proto.participants.first() { room.local_participant.role = participant.role() } room })?; let initial_project_id = if let Some(initial_project) = initial_project { let initial_project_id = room .update(&mut cx, |room, cx| { room.share_project(initial_project.clone(), cx) })? .await?; Some(initial_project_id) } else { None }; let did_join = room .update(&mut cx, |room, cx| { room.leave_when_empty = true; room.call(called_user_id, initial_project_id, cx) })? .await; match did_join { Ok(()) => Ok(room), Err(error) => Err(error.context("room creation failed")), } }) } pub(crate) async fn join_channel( channel_id: ChannelId, client: Arc<Client>, user_store: Model<UserStore>, cx: AsyncAppContext, ) -> Result<Model<Self>> { Self::from_join_response( client .request(proto::JoinChannel { channel_id: channel_id.0, }) .await?, client, user_store, cx, ) } pub(crate) async fn join( room_id: u64, client: Arc<Client>, user_store: Model<UserStore>, cx: AsyncAppContext, ) -> Result<Model<Self>> { Self::from_join_response( client.request(proto::JoinRoom { id: room_id }).await?, client, user_store, cx, ) } fn released(&mut self, cx: &mut AppContext) { if self.status.is_online() { self.leave_internal(cx).detach_and_log_err(cx); } } fn app_will_quit(&mut self, cx: &mut ModelContext<Self>) -> impl Future<Output = ()> { let task = if self.status.is_online() { let leave = self.leave_internal(cx); Some(cx.background_executor().spawn(async move { leave.await.log_err(); })) } else { None }; async move { if let Some(task) = task { task.await; } } } pub fn mute_on_join(cx: &AppContext) -> bool { CallSettings::get_global(cx).mute_on_join || client::IMPERSONATE_LOGIN.is_some() } fn from_join_response( response: proto::JoinRoomResponse, client: Arc<Client>, user_store: Model<UserStore>, mut cx: AsyncAppContext, ) -> Result<Model<Self>> { let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?; let room = cx.new_model(|cx| { Self::new( room_proto.id, response.channel_id.map(ChannelId), response.live_kit_connection_info, client, user_store, cx, ) })?; room.update(&mut cx, |room, cx| { room.leave_when_empty = room.channel_id.is_none(); room.apply_room_update(room_proto, cx)?; anyhow::Ok(()) })??; Ok(room) } fn should_leave(&self) -> bool { self.leave_when_empty && self.pending_room_update.is_none() && self.pending_participants.is_empty() && self.remote_participants.is_empty() && self.pending_call_count == 0 } pub(crate) fn leave(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> { cx.notify(); self.leave_internal(cx) } fn leave_internal(&mut self, cx: &mut AppContext) -> Task<Result<()>> { if self.status.is_offline() { return Task::ready(Err(anyhow!("room is offline"))); } log::info!("leaving room"); Audio::play_sound(Sound::Leave, cx); self.clear_state(cx); let leave_room = self.client.request(proto::LeaveRoom {}); cx.background_executor().spawn(async move { leave_room.await?; anyhow::Ok(()) }) } pub(crate) fn clear_state(&mut self, cx: &mut AppContext) { for project in self.shared_projects.drain() { if let Some(project) = project.upgrade() { project.update(cx, |project, cx| { project.unshare(cx).log_err(); }); } } for project in self.joined_projects.drain() { if let Some(project) = project.upgrade() { project.update(cx, |project, cx| { project.disconnected_from_host(cx); project.close(cx); }); } } self.status = RoomStatus::Offline; self.remote_participants.clear(); self.pending_participants.clear(); self.participant_user_ids.clear(); self.client_subscriptions.clear(); self.live_kit.take(); self.pending_room_update.take(); self.maintain_connection.take(); } async fn maintain_connection( this: WeakModel<Self>, client: Arc<Client>, mut cx: AsyncAppContext, ) -> Result<()> { let mut client_status = client.status(); loop { let _ = client_status.try_recv(); let is_connected = client_status.borrow().is_connected(); // Even if we're initially connected, any future change of the status means we momentarily disconnected. if !is_connected || client_status.next().await.is_some() { log::info!("detected client disconnection"); this.upgrade() .ok_or_else(|| anyhow!("room was dropped"))? .update(&mut cx, |this, cx| { this.status = RoomStatus::Rejoining; cx.notify(); })?; // Wait for client to re-establish a connection to the server. { let mut reconnection_timeout = cx.background_executor().timer(RECONNECT_TIMEOUT).fuse(); let client_reconnection = async { let mut remaining_attempts = 3; while remaining_attempts > 0 { if client_status.borrow().is_connected() { log::info!("client reconnected, attempting to rejoin room"); let Some(this) = this.upgrade() else { break }; match this.update(&mut cx, |this, cx| this.rejoin(cx)) { Ok(task) => { if task.await.log_err().is_some() { return true; } else { remaining_attempts -= 1; } } Err(_app_dropped) => return false, } } else if client_status.borrow().is_signed_out() { return false; } log::info!( "waiting for client status change, remaining attempts {}", remaining_attempts ); client_status.next().await; } false } .fuse(); futures::pin_mut!(client_reconnection); futures::select_biased! { reconnected = client_reconnection => { if reconnected { log::info!("successfully reconnected to room"); // If we successfully joined the room, go back around the loop // waiting for future connection status changes. continue; } } _ = reconnection_timeout => { log::info!("room reconnection timeout expired"); } } } break; } } // The client failed to re-establish a connection to the server // or an error occurred while trying to re-join the room. Either way // we leave the room and return an error. if let Some(this) = this.upgrade() { log::info!("reconnection failed, leaving room"); this.update(&mut cx, |this, cx| this.leave(cx))?.await?; } Err(anyhow!( "can't reconnect to room: client failed to re-establish connection" )) } fn rejoin(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> { let mut projects = HashMap::default(); let mut reshared_projects = Vec::new(); let mut rejoined_projects = Vec::new(); self.shared_projects.retain(|project| { if let Some(handle) = project.upgrade() { let project = handle.read(cx); if let Some(project_id) = project.remote_id() { projects.insert(project_id, handle.clone()); reshared_projects.push(proto::UpdateProject { project_id, worktrees: project.worktree_metadata_protos(cx), }); return true; } } false }); self.joined_projects.retain(|project| { if let Some(handle) = project.upgrade() { let project = handle.read(cx); if let Some(project_id) = project.remote_id() { projects.insert(project_id, handle.clone()); rejoined_projects.push(proto::RejoinProject { id: project_id, worktrees: project .worktrees(cx) .map(|worktree| { let worktree = worktree.read(cx); proto::RejoinWorktree { id: worktree.id().to_proto(), scan_id: worktree.completed_scan_id() as u64, } }) .collect(), }); } return true; } false }); let response = self.client.request_envelope(proto::RejoinRoom { id: self.id, reshared_projects, rejoined_projects, }); cx.spawn(|this, mut cx| async move { let response = response.await?; let message_id = response.message_id; let response = response.payload; let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?; this.update(&mut cx, |this, cx| { this.status = RoomStatus::Online; this.apply_room_update(room_proto, cx)?; Lines 1–500 of 1777. Load more
#![cfg_attr(target_os = "windows", allow(unused))] use crate::{ call_settings::CallSettings, participant::{LocalParticipant, ParticipantLocation, RemoteParticipant}, }; use anyhow::{anyhow, Result}; use audio::{Audio, Sound}; use client::{ proto::{self, PeerId}, ChannelId, Client, ParticipantIndex, TypedEnvelope, User, UserStore, }; use collections::{BTreeMap, HashMap, HashSet}; use fs::Fs; use futures::{FutureExt, StreamExt}; use gpui::{ AppContext, AsyncAppContext, Context, EventEmitter, Model, ModelContext, Task, WeakModel, }; use language::LanguageRegistry; #[cfg(not(target_os = "windows"))] use livekit::{ capture_local_audio_track, capture_local_video_track, id::ParticipantIdentity, options::{TrackPublishOptions, VideoCodec}, play_remote_audio_track, publication::LocalTrackPublication, track::{TrackKind, TrackSource}, RoomEvent, RoomOptions, }; #[cfg(target_os = "windows")] use livekit::{publication::LocalTrackPublication, RoomEvent}; use livekit_client as livekit; use postage::{sink::Sink, stream::Stream, watch}; use project::Project; use settings::Settings as _; use std::{any::Any, future::Future, mem, sync::Arc, time::Duration}; use util::{post_inc, ResultExt, TryFutureExt}; pub const RECONNECT_TIMEOUT: Duration = Duration::from_secs(30); #[derive(Clone, Debug, PartialEq, Eq)] pub enum Event { RoomJoined { channel_id: Option<ChannelId>, }, ParticipantLocationChanged { participant_id: proto::PeerId, }, RemoteVideoTracksChanged { participant_id: proto::PeerId, }, RemoteAudioTracksChanged { participant_id: proto::PeerId, }, RemoteProjectShared { owner: Arc<User>, project_id: u64, worktree_root_names: Vec<String>, }, RemoteProjectUnshared { project_id: u64, }, RemoteProjectJoined { project_id: u64, }, RemoteProjectInvitationDiscarded { project_id: u64, }, RoomLeft { channel_id: Option<ChannelId>, }, } pub struct Room { id: u64, channel_id: Option<ChannelId>, live_kit: Option<LiveKitRoom>, status: RoomStatus, shared_projects: HashSet<WeakModel<Project>>, joined_projects: HashSet<WeakModel<Project>>, local_participant: LocalParticipant, remote_participants: BTreeMap<u64, RemoteParticipant>, pending_participants: Vec<Arc<User>>, participant_user_ids: HashSet<u64>, pending_call_count: usize, leave_when_empty: bool, client: Arc<Client>, user_store: Model<UserStore>, follows_by_leader_id_project_id: HashMap<(PeerId, u64), Vec<PeerId>>, client_subscriptions: Vec<client::Subscription>, _subscriptions: Vec<gpui::Subscription>, room_update_completed_tx: watch::Sender<Option<()>>, room_update_completed_rx: watch::Receiver<Option<()>>, pending_room_update: Option<Task<()>>, maintain_connection: Option<Task<Option<()>>>, } impl EventEmitter<Event> for Room {} impl Room { pub fn channel_id(&self) -> Option<ChannelId> { self.channel_id } pub fn is_sharing_project(&self) -> bool { !self.shared_projects.is_empty() } #[cfg(all(any(test, feature = "test-support"), not(target_os = "windows")))] pub fn is_connected(&self) -> bool { if let Some(live_kit) = self.live_kit.as_ref() { live_kit.room.connection_state() == livekit::ConnectionState::Connected } else { false } } fn new( id: u64, channel_id: Option<ChannelId>, livekit_connection_info: Option<proto::LiveKitConnectionInfo>, client: Arc<Client>, user_store: Model<UserStore>, cx: &mut ModelContext<Self>, ) -> Self { spawn_room_connection(livekit_connection_info, cx); let maintain_connection = cx.spawn({ let client = client.clone(); move |this, cx| Self::maintain_connection(this, client.clone(), cx).log_err() }); Audio::play_sound(Sound::Joined, cx); let (room_update_completed_tx, room_update_completed_rx) = watch::channel(); Self { id, channel_id, live_kit: None, status: RoomStatus::Online, shared_projects: Default::default(), joined_projects: Default::default(), participant_user_ids: Default::default(), local_participant: Default::default(), remote_participants: Default::default(), pending_participants: Default::default(), pending_call_count: 0, client_subscriptions: vec![ client.add_message_handler(cx.weak_model(), Self::handle_room_updated) ], _subscriptions: vec![ cx.on_release(Self::released), cx.on_app_quit(Self::app_will_quit), ], leave_when_empty: false, pending_room_update: None, client, user_store, follows_by_leader_id_project_id: Default::default(), maintain_connection: Some(maintain_connection), room_update_completed_tx, room_update_completed_rx, } } pub(crate) fn create( called_user_id: u64, initial_project: Option<Model<Project>>, client: Arc<Client>, user_store: Model<UserStore>, cx: &mut AppContext, ) -> Task<Result<Model<Self>>> { cx.spawn(move |mut cx| async move { let response = client.request(proto::CreateRoom {}).await?; let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?; let room = cx.new_model(|cx| { let mut room = Self::new( room_proto.id, None, response.live_kit_connection_info, client, user_store, cx, ); if let Some(participant) = room_proto.participants.first() { room.local_participant.role = participant.role() } room })?; let initial_project_id = if let Some(initial_project) = initial_project { let initial_project_id = room .update(&mut cx, |room, cx| { room.share_project(initial_project.clone(), cx) })? .await?; Some(initial_project_id) } else { None }; let did_join = room .update(&mut cx, |room, cx| { room.leave_when_empty = true; room.call(called_user_id, initial_project_id, cx) })? .await; match did_join { Ok(()) => Ok(room), Err(error) => Err(error.context("room creation failed")), } }) } pub(crate) async fn join_channel( channel_id: ChannelId, client: Arc<Client>, user_store: Model<UserStore>, cx: AsyncAppContext, ) -> Result<Model<Self>> { Self::from_join_response( client .request(proto::JoinChannel { channel_id: channel_id.0, }) .await?, client, user_store, cx, ) } pub(crate) async fn join( room_id: u64, client: Arc<Client>, user_store: Model<UserStore>, cx: AsyncAppContext, ) -> Result<Model<Self>> { Self::from_join_response( client.request(proto::JoinRoom { id: room_id }).await?, client, user_store, cx, ) } fn released(&mut self, cx: &mut AppContext) { if self.status.is_online() { self.leave_internal(cx).detach_and_log_err(cx); } } fn app_will_quit(&mut self, cx: &mut ModelContext<Self>) -> impl Future<Output = ()> { let task = if self.status.is_online() { let leave = self.leave_internal(cx); Some(cx.background_executor().spawn(async move { leave.await.log_err(); })) } else { None }; async move { if let Some(task) = task { task.await; } } } pub fn mute_on_join(cx: &AppContext) -> bool { CallSettings::get_global(cx).mute_on_join || client::IMPERSONATE_LOGIN.is_some() } fn from_join_response( response: proto::JoinRoomResponse, client: Arc<Client>, user_store: Model<UserStore>, mut cx: AsyncAppContext, ) -> Result<Model<Self>> { let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?; let room = cx.new_model(|cx| { Self::new( room_proto.id, response.channel_id.map(ChannelId), response.live_kit_connection_info, client, user_store, cx, ) })?; room.update(&mut cx, |room, cx| { room.leave_when_empty = room.channel_id.is_none(); room.apply_room_update(room_proto, cx)?; anyhow::Ok(()) })??; Ok(room) } fn should_leave(&self) -> bool { self.leave_when_empty && self.pending_room_update.is_none() && self.pending_participants.is_empty() && self.remote_participants.is_empty() && self.pending_call_count == 0 } pub(crate) fn leave(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> { cx.notify(); self.leave_internal(cx) } fn leave_internal(&mut self, cx: &mut AppContext) -> Task<Result<()>> { if self.status.is_offline() { return Task::ready(Err(anyhow!("room is offline"))); } log::info!("leaving room"); Audio::play_sound(Sound::Leave, cx); self.clear_state(cx); let leave_room = self.client.request(proto::LeaveRoom {}); cx.background_executor().spawn(async move { leave_room.await?; anyhow::Ok(()) }) } pub(crate) fn clear_state(&mut self, cx: &mut AppContext) { for project in self.shared_projects.drain() { if let Some(project) = project.upgrade() { project.update(cx, |project, cx| { project.unshare(cx).log_err(); }); } } for project in self.joined_projects.drain() { if let Some(project) = project.upgrade() { project.update(cx, |project, cx| { project.disconnected_from_host(cx); project.close(cx); }); } } self.status = RoomStatus::Offline; self.remote_participants.clear(); self.pending_participants.clear(); self.participant_user_ids.clear(); self.client_subscriptions.clear(); self.live_kit.take(); self.pending_room_update.take(); self.maintain_connection.take(); } async fn maintain_connection( this: WeakModel<Self>, client: Arc<Client>, mut cx: AsyncAppContext, ) -> Result<()> { let mut client_status = client.status(); loop { let _ = client_status.try_recv(); let is_connected = client_status.borrow().is_connected(); // Even if we're initially connected, any future change of the status means we momentarily disconnected. if !is_connected || client_status.next().await.is_some() { log::info!("detected client disconnection"); this.upgrade() .ok_or_else(|| anyhow!("room was dropped"))? .update(&mut cx, |this, cx| { this.status = RoomStatus::Rejoining; cx.notify(); })?; // Wait for client to re-establish a connection to the server. { let mut reconnection_timeout = cx.background_executor().timer(RECONNECT_TIMEOUT).fuse(); let client_reconnection = async { let mut remaining_attempts = 3; while remaining_attempts > 0 { if client_status.borrow().is_connected() { log::info!("client reconnected, attempting to rejoin room"); let Some(this) = this.upgrade() else { break }; match this.update(&mut cx, |this, cx| this.rejoin(cx)) { Ok(task) => { if task.await.log_err().is_some() { return true; } else { remaining_attempts -= 1; } } Err(_app_dropped) => return false, } } else if client_status.borrow().is_signed_out() { return false; } log::info!( "waiting for client status change, remaining attempts {}", remaining_attempts ); client_status.next().await; } false } .fuse(); futures::pin_mut!(client_reconnection); futures::select_biased! { reconnected = client_reconnection => { if reconnected { log::info!("successfully reconnected to room"); // If we successfully joined the room, go back around the loop // waiting for future connection status changes. continue; } } _ = reconnection_timeout => { log::info!("room reconnection timeout expired"); } } } break; } } // The client failed to re-establish a connection to the server // or an error occurred while trying to re-join the room. Either way // we leave the room and return an error. if let Some(this) = this.upgrade() { log::info!("reconnection failed, leaving room"); this.update(&mut cx, |this, cx| this.leave(cx))?.await?; } Err(anyhow!( "can't reconnect to room: client failed to re-establish connection" )) } fn rejoin(&mut self, cx: &mut ModelContext<Self>) -> Task<Result<()>> { let mut projects = HashMap::default(); let mut reshared_projects = Vec::new(); let mut rejoined_projects = Vec::new(); self.shared_projects.retain(|project| { if let Some(handle) = project.upgrade() { let project = handle.read(cx); if let Some(project_id) = project.remote_id() { projects.insert(project_id, handle.clone()); reshared_projects.push(proto::UpdateProject { project_id, worktrees: project.worktree_metadata_protos(cx), }); return true; } } false }); self.joined_projects.retain(|project| { if let Some(handle) = project.upgrade() { let project = handle.read(cx); if let Some(project_id) = project.remote_id() { projects.insert(project_id, handle.clone()); rejoined_projects.push(proto::RejoinProject { id: project_id, worktrees: project .worktrees(cx) .map(|worktree| { let worktree = worktree.read(cx); proto::RejoinWorktree { id: worktree.id().to_proto(), scan_id: worktree.completed_scan_id() as u64, } }) .collect(), }); } return true; } false }); let response = self.client.request_envelope(proto::RejoinRoom { id: self.id, reshared_projects, rejoined_projects, }); cx.spawn(|this, mut cx| async move { let response = response.await?; let message_id = response.message_id; let response = response.payload; let room_proto = response.room.ok_or_else(|| anyhow!("invalid room"))?; this.update(&mut cx, |this, cx| { this.status = RoomStatus::Online; this.apply_room_update(room_proto, cx)?;