1#[cfg(any(test, feature = "test-support"))]
2pub mod test;
3
4pub mod telemetry;
5pub mod user;
6
7use anyhow::{anyhow, Context as _, Result};
8use async_recursion::async_recursion;
9use async_tungstenite::tungstenite::{
10 error::Error as WebsocketError,
11 http::{Request, StatusCode},
12};
13use clock::SystemClock;
14use collections::HashMap;
15use futures::{
16 channel::oneshot, future::LocalBoxFuture, AsyncReadExt, FutureExt, SinkExt, Stream, StreamExt,
17 TryFutureExt as _, TryStreamExt,
18};
19use gpui::{
20 actions, AnyModel, AnyWeakModel, AppContext, AsyncAppContext, Global, Model, Task, WeakModel,
21};
22use lazy_static::lazy_static;
23use parking_lot::RwLock;
24use postage::watch;
25use rand::prelude::*;
26use release_channel::{AppVersion, ReleaseChannel};
27use rpc::proto::{AnyTypedEnvelope, EntityMessage, EnvelopedMessage, PeerId, RequestMessage};
28use schemars::JsonSchema;
29use serde::{Deserialize, Serialize};
30use settings::{Settings, SettingsStore};
31use std::fmt;
32use std::{
33 any::TypeId,
34 convert::TryFrom,
35 fmt::Write as _,
36 future::Future,
37 marker::PhantomData,
38 path::PathBuf,
39 sync::{
40 atomic::{AtomicU64, Ordering},
41 Arc, Weak,
42 },
43 time::{Duration, Instant},
44};
45use telemetry::Telemetry;
46use thiserror::Error;
47use url::Url;
48use util::http::{HttpClient, HttpClientWithUrl};
49use util::{ResultExt, TryFutureExt};
50
51pub use rpc::*;
52pub use telemetry_events::Event;
53pub use user::*;
54
55#[derive(Debug, Clone, Eq, PartialEq)]
56pub struct DevServerToken(pub String);
57
58impl fmt::Display for DevServerToken {
59 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60 write!(f, "{}", self.0)
61 }
62}
63
64lazy_static! {
65 static ref ZED_SERVER_URL: Option<String> = std::env::var("ZED_SERVER_URL").ok();
66 static ref ZED_RPC_URL: Option<String> = std::env::var("ZED_RPC_URL").ok();
67 pub static ref IMPERSONATE_LOGIN: Option<String> = std::env::var("ZED_IMPERSONATE")
68 .ok()
69 .and_then(|s| if s.is_empty() { None } else { Some(s) });
70 pub static ref ADMIN_API_TOKEN: Option<String> = std::env::var("ZED_ADMIN_API_TOKEN")
71 .ok()
72 .and_then(|s| if s.is_empty() { None } else { Some(s) });
73 pub static ref ZED_APP_PATH: Option<PathBuf> =
74 std::env::var("ZED_APP_PATH").ok().map(PathBuf::from);
75 pub static ref ZED_ALWAYS_ACTIVE: bool =
76 std::env::var("ZED_ALWAYS_ACTIVE").map_or(false, |e| !e.is_empty());
77}
78
79pub const INITIAL_RECONNECTION_DELAY: Duration = Duration::from_millis(100);
80pub const CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);
81
82actions!(client, [SignIn, SignOut, Reconnect]);
83
84#[derive(Clone, Default, Serialize, Deserialize, JsonSchema)]
85pub struct ClientSettingsContent {
86 server_url: Option<String>,
87}
88
89#[derive(Deserialize)]
90pub struct ClientSettings {
91 pub server_url: String,
92}
93
94impl Settings for ClientSettings {
95 const KEY: Option<&'static str> = None;
96
97 type FileContent = ClientSettingsContent;
98
99 fn load(
100 default_value: &Self::FileContent,
101 user_values: &[&Self::FileContent],
102 _: &mut AppContext,
103 ) -> Result<Self>
104 where
105 Self: Sized,
106 {
107 let mut result = Self::load_via_json_merge(default_value, user_values)?;
108 if let Some(server_url) = &*ZED_SERVER_URL {
109 result.server_url = server_url.clone()
110 }
111 Ok(result)
112 }
113}
114
115pub fn init_settings(cx: &mut AppContext) {
116 TelemetrySettings::register(cx);
117 cx.update_global(|store: &mut SettingsStore, cx| {
118 store.register_setting::<ClientSettings>(cx);
119 });
120}
121
122pub fn init(client: &Arc<Client>, cx: &mut AppContext) {
123 let client = Arc::downgrade(client);
124 cx.on_action({
125 let client = client.clone();
126 move |_: &SignIn, cx| {
127 if let Some(client) = client.upgrade() {
128 cx.spawn(
129 |cx| async move { client.authenticate_and_connect(true, &cx).log_err().await },
130 )
131 .detach();
132 }
133 }
134 });
135
136 cx.on_action({
137 let client = client.clone();
138 move |_: &SignOut, cx| {
139 if let Some(client) = client.upgrade() {
140 cx.spawn(|cx| async move {
141 client.disconnect(&cx);
142 })
143 .detach();
144 }
145 }
146 });
147
148 cx.on_action({
149 let client = client.clone();
150 move |_: &Reconnect, cx| {
151 if let Some(client) = client.upgrade() {
152 cx.spawn(|cx| async move {
153 client.reconnect(&cx);
154 })
155 .detach();
156 }
157 }
158 });
159}
160
161struct GlobalClient(Arc<Client>);
162
163impl Global for GlobalClient {}
164
165pub struct Client {
166 id: AtomicU64,
167 peer: Arc<Peer>,
168 http: Arc<HttpClientWithUrl>,
169 telemetry: Arc<Telemetry>,
170 state: RwLock<ClientState>,
171
172 #[allow(clippy::type_complexity)]
173 #[cfg(any(test, feature = "test-support"))]
174 authenticate: RwLock<
175 Option<Box<dyn 'static + Send + Sync + Fn(&AsyncAppContext) -> Task<Result<Credentials>>>>,
176 >,
177
178 #[allow(clippy::type_complexity)]
179 #[cfg(any(test, feature = "test-support"))]
180 establish_connection: RwLock<
181 Option<
182 Box<
183 dyn 'static
184 + Send
185 + Sync
186 + Fn(
187 &Credentials,
188 &AsyncAppContext,
189 ) -> Task<Result<Connection, EstablishConnectionError>>,
190 >,
191 >,
192 >,
193}
194
195#[derive(Error, Debug)]
196pub enum EstablishConnectionError {
197 #[error("upgrade required")]
198 UpgradeRequired,
199 #[error("unauthorized")]
200 Unauthorized,
201 #[error("{0}")]
202 Other(#[from] anyhow::Error),
203 #[error("{0}")]
204 Http(#[from] util::http::Error),
205 #[error("{0}")]
206 Io(#[from] std::io::Error),
207 #[error("{0}")]
208 Websocket(#[from] async_tungstenite::tungstenite::http::Error),
209}
210
211impl From<WebsocketError> for EstablishConnectionError {
212 fn from(error: WebsocketError) -> Self {
213 if let WebsocketError::Http(response) = &error {
214 match response.status() {
215 StatusCode::UNAUTHORIZED => return EstablishConnectionError::Unauthorized,
216 StatusCode::UPGRADE_REQUIRED => return EstablishConnectionError::UpgradeRequired,
217 _ => {}
218 }
219 }
220 EstablishConnectionError::Other(error.into())
221 }
222}
223
224impl EstablishConnectionError {
225 pub fn other(error: impl Into<anyhow::Error> + Send + Sync) -> Self {
226 Self::Other(error.into())
227 }
228}
229
230#[derive(Copy, Clone, Debug, PartialEq)]
231pub enum Status {
232 SignedOut,
233 UpgradeRequired,
234 Authenticating,
235 Connecting,
236 ConnectionError,
237 Connected {
238 peer_id: PeerId,
239 connection_id: ConnectionId,
240 },
241 ConnectionLost,
242 Reauthenticating,
243 Reconnecting,
244 ReconnectionError {
245 next_reconnection: Instant,
246 },
247}
248
249impl Status {
250 pub fn is_connected(&self) -> bool {
251 matches!(self, Self::Connected { .. })
252 }
253
254 pub fn is_signed_out(&self) -> bool {
255 matches!(self, Self::SignedOut | Self::UpgradeRequired)
256 }
257}
258
259struct ClientState {
260 credentials: Option<Credentials>,
261 status: (watch::Sender<Status>, watch::Receiver<Status>),
262 entity_id_extractors: HashMap<TypeId, fn(&dyn AnyTypedEnvelope) -> u64>,
263 _reconnect_task: Option<Task<()>>,
264 reconnect_interval: Duration,
265 entities_by_type_and_remote_id: HashMap<(TypeId, u64), WeakSubscriber>,
266 models_by_message_type: HashMap<TypeId, AnyWeakModel>,
267 entity_types_by_message_type: HashMap<TypeId, TypeId>,
268 #[allow(clippy::type_complexity)]
269 message_handlers: HashMap<
270 TypeId,
271 Arc<
272 dyn Send
273 + Sync
274 + Fn(
275 AnyModel,
276 Box<dyn AnyTypedEnvelope>,
277 &Arc<Client>,
278 AsyncAppContext,
279 ) -> LocalBoxFuture<'static, Result<()>>,
280 >,
281 >,
282}
283
284enum WeakSubscriber {
285 Entity { handle: AnyWeakModel },
286 Pending(Vec<Box<dyn AnyTypedEnvelope>>),
287}
288
289#[derive(Clone, Debug, Eq, PartialEq)]
290pub enum Credentials {
291 DevServer { token: DevServerToken },
292 User { user_id: u64, access_token: String },
293}
294
295impl Credentials {
296 pub fn authorization_header(&self) -> String {
297 match self {
298 Credentials::DevServer { token } => format!("dev-server-token {}", token),
299 Credentials::User {
300 user_id,
301 access_token,
302 } => format!("{} {}", user_id, access_token),
303 }
304 }
305}
306
307impl Default for ClientState {
308 fn default() -> Self {
309 Self {
310 credentials: None,
311 status: watch::channel_with(Status::SignedOut),
312 entity_id_extractors: Default::default(),
313 _reconnect_task: None,
314 reconnect_interval: Duration::from_secs(5),
315 models_by_message_type: Default::default(),
316 entities_by_type_and_remote_id: Default::default(),
317 entity_types_by_message_type: Default::default(),
318 message_handlers: Default::default(),
319 }
320 }
321}
322
323pub enum Subscription {
324 Entity {
325 client: Weak<Client>,
326 id: (TypeId, u64),
327 },
328 Message {
329 client: Weak<Client>,
330 id: TypeId,
331 },
332}
333
334impl Drop for Subscription {
335 fn drop(&mut self) {
336 match self {
337 Subscription::Entity { client, id } => {
338 if let Some(client) = client.upgrade() {
339 let mut state = client.state.write();
340 let _ = state.entities_by_type_and_remote_id.remove(id);
341 }
342 }
343 Subscription::Message { client, id } => {
344 if let Some(client) = client.upgrade() {
345 let mut state = client.state.write();
346 let _ = state.entity_types_by_message_type.remove(id);
347 let _ = state.message_handlers.remove(id);
348 }
349 }
350 }
351 }
352}
353
354pub struct PendingEntitySubscription<T: 'static> {
355 client: Arc<Client>,
356 remote_id: u64,
357 _entity_type: PhantomData<T>,
358 consumed: bool,
359}
360
361impl<T: 'static> PendingEntitySubscription<T> {
362 pub fn set_model(mut self, model: &Model<T>, cx: &mut AsyncAppContext) -> Subscription {
363 self.consumed = true;
364 let mut state = self.client.state.write();
365 let id = (TypeId::of::<T>(), self.remote_id);
366 let Some(WeakSubscriber::Pending(messages)) =
367 state.entities_by_type_and_remote_id.remove(&id)
368 else {
369 unreachable!()
370 };
371
372 state.entities_by_type_and_remote_id.insert(
373 id,
374 WeakSubscriber::Entity {
375 handle: model.downgrade().into(),
376 },
377 );
378 drop(state);
379 for message in messages {
380 self.client.handle_message(message, cx);
381 }
382 Subscription::Entity {
383 client: Arc::downgrade(&self.client),
384 id,
385 }
386 }
387}
388
389impl<T: 'static> Drop for PendingEntitySubscription<T> {
390 fn drop(&mut self) {
391 if !self.consumed {
392 let mut state = self.client.state.write();
393 if let Some(WeakSubscriber::Pending(messages)) = state
394 .entities_by_type_and_remote_id
395 .remove(&(TypeId::of::<T>(), self.remote_id))
396 {
397 for message in messages {
398 log::info!("unhandled message {}", message.payload_type_name());
399 }
400 }
401 }
402 }
403}
404
405#[derive(Copy, Clone)]
406pub struct TelemetrySettings {
407 pub diagnostics: bool,
408 pub metrics: bool,
409}
410
411/// Control what info is collected by Zed.
412#[derive(Default, Clone, Serialize, Deserialize, JsonSchema)]
413pub struct TelemetrySettingsContent {
414 /// Send debug info like crash reports.
415 ///
416 /// Default: true
417 pub diagnostics: Option<bool>,
418 /// Send anonymized usage data like what languages you're using Zed with.
419 ///
420 /// Default: true
421 pub metrics: Option<bool>,
422}
423
424impl settings::Settings for TelemetrySettings {
425 const KEY: Option<&'static str> = Some("telemetry");
426
427 type FileContent = TelemetrySettingsContent;
428
429 fn load(
430 default_value: &Self::FileContent,
431 user_values: &[&Self::FileContent],
432 _: &mut AppContext,
433 ) -> Result<Self> {
434 Ok(Self {
435 diagnostics: user_values.first().and_then(|v| v.diagnostics).unwrap_or(
436 default_value
437 .diagnostics
438 .ok_or_else(Self::missing_default)?,
439 ),
440 metrics: user_values
441 .first()
442 .and_then(|v| v.metrics)
443 .unwrap_or(default_value.metrics.ok_or_else(Self::missing_default)?),
444 })
445 }
446}
447
448impl Client {
449 pub fn new(
450 clock: Arc<dyn SystemClock>,
451 http: Arc<HttpClientWithUrl>,
452 cx: &mut AppContext,
453 ) -> Arc<Self> {
454 Arc::new(Self {
455 id: AtomicU64::new(0),
456 peer: Peer::new(0),
457 telemetry: Telemetry::new(clock, http.clone(), cx),
458 http,
459 state: Default::default(),
460
461 #[cfg(any(test, feature = "test-support"))]
462 authenticate: Default::default(),
463 #[cfg(any(test, feature = "test-support"))]
464 establish_connection: Default::default(),
465 })
466 }
467
468 pub fn id(&self) -> u64 {
469 self.id.load(Ordering::SeqCst)
470 }
471
472 pub fn http_client(&self) -> Arc<HttpClientWithUrl> {
473 self.http.clone()
474 }
475
476 pub fn set_id(&self, id: u64) -> &Self {
477 self.id.store(id, Ordering::SeqCst);
478 self
479 }
480
481 #[cfg(any(test, feature = "test-support"))]
482 pub fn teardown(&self) {
483 let mut state = self.state.write();
484 state._reconnect_task.take();
485 state.message_handlers.clear();
486 state.models_by_message_type.clear();
487 state.entities_by_type_and_remote_id.clear();
488 state.entity_id_extractors.clear();
489 self.peer.teardown();
490 }
491
492 #[cfg(any(test, feature = "test-support"))]
493 pub fn override_authenticate<F>(&self, authenticate: F) -> &Self
494 where
495 F: 'static + Send + Sync + Fn(&AsyncAppContext) -> Task<Result<Credentials>>,
496 {
497 *self.authenticate.write() = Some(Box::new(authenticate));
498 self
499 }
500
501 #[cfg(any(test, feature = "test-support"))]
502 pub fn override_establish_connection<F>(&self, connect: F) -> &Self
503 where
504 F: 'static
505 + Send
506 + Sync
507 + Fn(&Credentials, &AsyncAppContext) -> Task<Result<Connection, EstablishConnectionError>>,
508 {
509 *self.establish_connection.write() = Some(Box::new(connect));
510 self
511 }
512
513 pub fn global(cx: &AppContext) -> Arc<Self> {
514 cx.global::<GlobalClient>().0.clone()
515 }
516 pub fn set_global(client: Arc<Client>, cx: &mut AppContext) {
517 cx.set_global(GlobalClient(client))
518 }
519
520 pub fn user_id(&self) -> Option<u64> {
521 if let Some(Credentials::User { user_id, .. }) = self.state.read().credentials.as_ref() {
522 Some(*user_id)
523 } else {
524 None
525 }
526 }
527
528 pub fn peer_id(&self) -> Option<PeerId> {
529 if let Status::Connected { peer_id, .. } = &*self.status().borrow() {
530 Some(*peer_id)
531 } else {
532 None
533 }
534 }
535
536 pub fn status(&self) -> watch::Receiver<Status> {
537 self.state.read().status.1.clone()
538 }
539
540 fn set_status(self: &Arc<Self>, status: Status, cx: &AsyncAppContext) {
541 log::info!("set status on client {}: {:?}", self.id(), status);
542 let mut state = self.state.write();
543 *state.status.0.borrow_mut() = status;
544
545 match status {
546 Status::Connected { .. } => {
547 state._reconnect_task = None;
548 }
549 Status::ConnectionLost => {
550 let this = self.clone();
551 let reconnect_interval = state.reconnect_interval;
552 state._reconnect_task = Some(cx.spawn(move |cx| async move {
553 #[cfg(any(test, feature = "test-support"))]
554 let mut rng = StdRng::seed_from_u64(0);
555 #[cfg(not(any(test, feature = "test-support")))]
556 let mut rng = StdRng::from_entropy();
557
558 let mut delay = INITIAL_RECONNECTION_DELAY;
559 while let Err(error) = this.authenticate_and_connect(true, &cx).await {
560 log::error!("failed to connect {}", error);
561 if matches!(*this.status().borrow(), Status::ConnectionError) {
562 this.set_status(
563 Status::ReconnectionError {
564 next_reconnection: Instant::now() + delay,
565 },
566 &cx,
567 );
568 cx.background_executor().timer(delay).await;
569 delay = delay
570 .mul_f32(rng.gen_range(1.0..=2.0))
571 .min(reconnect_interval);
572 } else {
573 break;
574 }
575 }
576 }));
577 }
578 Status::SignedOut | Status::UpgradeRequired => {
579 self.telemetry.set_authenticated_user_info(None, false);
580 state._reconnect_task.take();
581 }
582 _ => {}
583 }
584 }
585
586 pub fn subscribe_to_entity<T>(
587 self: &Arc<Self>,
588 remote_id: u64,
589 ) -> Result<PendingEntitySubscription<T>>
590 where
591 T: 'static,
592 {
593 let id = (TypeId::of::<T>(), remote_id);
594
595 let mut state = self.state.write();
596 if state.entities_by_type_and_remote_id.contains_key(&id) {
597 return Err(anyhow!("already subscribed to entity"));
598 }
599
600 state
601 .entities_by_type_and_remote_id
602 .insert(id, WeakSubscriber::Pending(Default::default()));
603
604 Ok(PendingEntitySubscription {
605 client: self.clone(),
606 remote_id,
607 consumed: false,
608 _entity_type: PhantomData,
609 })
610 }
611
612 #[track_caller]
613 pub fn add_message_handler<M, E, H, F>(
614 self: &Arc<Self>,
615 entity: WeakModel<E>,
616 handler: H,
617 ) -> Subscription
618 where
619 M: EnvelopedMessage,
620 E: 'static,
621 H: 'static
622 + Sync
623 + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F
624 + Send
625 + Sync,
626 F: 'static + Future<Output = Result<()>>,
627 {
628 let message_type_id = TypeId::of::<M>();
629 let mut state = self.state.write();
630 state
631 .models_by_message_type
632 .insert(message_type_id, entity.into());
633
634 let prev_handler = state.message_handlers.insert(
635 message_type_id,
636 Arc::new(move |subscriber, envelope, client, cx| {
637 let subscriber = subscriber.downcast::<E>().unwrap();
638 let envelope = envelope.into_any().downcast::<TypedEnvelope<M>>().unwrap();
639 handler(subscriber, *envelope, client.clone(), cx).boxed_local()
640 }),
641 );
642 if prev_handler.is_some() {
643 let location = std::panic::Location::caller();
644 panic!(
645 "{}:{} registered handler for the same message {} twice",
646 location.file(),
647 location.line(),
648 std::any::type_name::<M>()
649 );
650 }
651
652 Subscription::Message {
653 client: Arc::downgrade(self),
654 id: message_type_id,
655 }
656 }
657
658 pub fn add_request_handler<M, E, H, F>(
659 self: &Arc<Self>,
660 model: WeakModel<E>,
661 handler: H,
662 ) -> Subscription
663 where
664 M: RequestMessage,
665 E: 'static,
666 H: 'static
667 + Sync
668 + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F
669 + Send
670 + Sync,
671 F: 'static + Future<Output = Result<M::Response>>,
672 {
673 self.add_message_handler(model, move |handle, envelope, this, cx| {
674 Self::respond_to_request(
675 envelope.receipt(),
676 handler(handle, envelope, this.clone(), cx),
677 this,
678 )
679 })
680 }
681
682 pub fn add_model_message_handler<M, E, H, F>(self: &Arc<Self>, handler: H)
683 where
684 M: EntityMessage,
685 E: 'static,
686 H: 'static + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F + Send + Sync,
687 F: 'static + Future<Output = Result<()>>,
688 {
689 self.add_entity_message_handler::<M, E, _, _>(move |subscriber, message, client, cx| {
690 handler(subscriber.downcast::<E>().unwrap(), message, client, cx)
691 })
692 }
693
694 fn add_entity_message_handler<M, E, H, F>(self: &Arc<Self>, handler: H)
695 where
696 M: EntityMessage,
697 E: 'static,
698 H: 'static + Fn(AnyModel, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F + Send + Sync,
699 F: 'static + Future<Output = Result<()>>,
700 {
701 let model_type_id = TypeId::of::<E>();
702 let message_type_id = TypeId::of::<M>();
703
704 let mut state = self.state.write();
705 state
706 .entity_types_by_message_type
707 .insert(message_type_id, model_type_id);
708 state
709 .entity_id_extractors
710 .entry(message_type_id)
711 .or_insert_with(|| {
712 |envelope| {
713 envelope
714 .as_any()
715 .downcast_ref::<TypedEnvelope<M>>()
716 .unwrap()
717 .payload
718 .remote_entity_id()
719 }
720 });
721 let prev_handler = state.message_handlers.insert(
722 message_type_id,
723 Arc::new(move |handle, envelope, client, cx| {
724 let envelope = envelope.into_any().downcast::<TypedEnvelope<M>>().unwrap();
725 handler(handle, *envelope, client.clone(), cx).boxed_local()
726 }),
727 );
728 if prev_handler.is_some() {
729 panic!("registered handler for the same message twice");
730 }
731 }
732
733 pub fn add_model_request_handler<M, E, H, F>(self: &Arc<Self>, handler: H)
734 where
735 M: EntityMessage + RequestMessage,
736 E: 'static,
737 H: 'static + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F + Send + Sync,
738 F: 'static + Future<Output = Result<M::Response>>,
739 {
740 self.add_model_message_handler(move |entity, envelope, client, cx| {
741 Self::respond_to_request::<M, _>(
742 envelope.receipt(),
743 handler(entity, envelope, client.clone(), cx),
744 client,
745 )
746 })
747 }
748
749 async fn respond_to_request<T: RequestMessage, F: Future<Output = Result<T::Response>>>(
750 receipt: Receipt<T>,
751 response: F,
752 client: Arc<Self>,
753 ) -> Result<()> {
754 match response.await {
755 Ok(response) => {
756 client.respond(receipt, response)?;
757 Ok(())
758 }
759 Err(error) => {
760 client.respond_with_error(receipt, error.to_proto())?;
761 Err(error)
762 }
763 }
764 }
765
766 pub async fn has_keychain_credentials(&self, cx: &AsyncAppContext) -> bool {
767 read_credentials_from_keychain(cx).await.is_some()
768 }
769
770 pub fn set_dev_server_token(&self, token: DevServerToken) {
771 self.state.write().credentials = Some(Credentials::DevServer { token });
772 }
773
774 #[async_recursion(?Send)]
775 pub async fn authenticate_and_connect(
776 self: &Arc<Self>,
777 try_keychain: bool,
778 cx: &AsyncAppContext,
779 ) -> anyhow::Result<()> {
780 let was_disconnected = match *self.status().borrow() {
781 Status::SignedOut => true,
782 Status::ConnectionError
783 | Status::ConnectionLost
784 | Status::Authenticating { .. }
785 | Status::Reauthenticating { .. }
786 | Status::ReconnectionError { .. } => false,
787 Status::Connected { .. } | Status::Connecting { .. } | Status::Reconnecting { .. } => {
788 return Ok(())
789 }
790 Status::UpgradeRequired => return Err(EstablishConnectionError::UpgradeRequired)?,
791 };
792
793 if was_disconnected {
794 self.set_status(Status::Authenticating, cx);
795 } else {
796 self.set_status(Status::Reauthenticating, cx)
797 }
798
799 let mut read_from_keychain = false;
800 let mut credentials = self.state.read().credentials.clone();
801 if credentials.is_none() && try_keychain {
802 credentials = read_credentials_from_keychain(cx).await;
803 read_from_keychain = credentials.is_some();
804 }
805 if credentials.is_none() {
806 let mut status_rx = self.status();
807 let _ = status_rx.next().await;
808 futures::select_biased! {
809 authenticate = self.authenticate(cx).fuse() => {
810 match authenticate {
811 Ok(creds) => credentials = Some(creds),
812 Err(err) => {
813 self.set_status(Status::ConnectionError, cx);
814 return Err(err);
815 }
816 }
817 }
818 _ = status_rx.next().fuse() => {
819 return Err(anyhow!("authentication canceled"));
820 }
821 }
822 }
823 let credentials = credentials.unwrap();
824 if let Credentials::User { user_id, .. } = &credentials {
825 self.set_id(*user_id);
826 }
827
828 if was_disconnected {
829 self.set_status(Status::Connecting, cx);
830 } else {
831 self.set_status(Status::Reconnecting, cx);
832 }
833
834 let mut timeout =
835 futures::FutureExt::fuse(cx.background_executor().timer(CONNECTION_TIMEOUT));
836 futures::select_biased! {
837 connection = self.establish_connection(&credentials, cx).fuse() => {
838 match connection {
839 Ok(conn) => {
840 self.state.write().credentials = Some(credentials.clone());
841 if !read_from_keychain && IMPERSONATE_LOGIN.is_none() {
842 if let Credentials::User{user_id, access_token} = credentials {
843 write_credentials_to_keychain(user_id, access_token, cx).await.log_err();
844 }
845 }
846
847 futures::select_biased! {
848 result = self.set_connection(conn, cx).fuse() => result,
849 _ = timeout => {
850 self.set_status(Status::ConnectionError, cx);
851 Err(anyhow!("timed out waiting on hello message from server"))
852 }
853 }
854 }
855 Err(EstablishConnectionError::Unauthorized) => {
856 self.state.write().credentials.take();
857 if read_from_keychain {
858 delete_credentials_from_keychain(cx).await.log_err();
859 self.set_status(Status::SignedOut, cx);
860 self.authenticate_and_connect(false, cx).await
861 } else {
862 self.set_status(Status::ConnectionError, cx);
863 Err(EstablishConnectionError::Unauthorized)?
864 }
865 }
866 Err(EstablishConnectionError::UpgradeRequired) => {
867 self.set_status(Status::UpgradeRequired, cx);
868 Err(EstablishConnectionError::UpgradeRequired)?
869 }
870 Err(error) => {
871 self.set_status(Status::ConnectionError, cx);
872 Err(error)?
873 }
874 }
875 }
876 _ = &mut timeout => {
877 self.set_status(Status::ConnectionError, cx);
878 Err(anyhow!("timed out trying to establish connection"))
879 }
880 }
881 }
882
883 async fn set_connection(
884 self: &Arc<Self>,
885 conn: Connection,
886 cx: &AsyncAppContext,
887 ) -> Result<()> {
888 let executor = cx.background_executor();
889 log::info!("add connection to peer");
890 let (connection_id, handle_io, mut incoming) = self.peer.add_connection(conn, {
891 let executor = executor.clone();
892 move |duration| executor.timer(duration)
893 });
894 let handle_io = executor.spawn(handle_io);
895
896 let peer_id = async {
897 log::info!("waiting for server hello");
898 let message = incoming
899 .next()
900 .await
901 .ok_or_else(|| anyhow!("no hello message received"))?;
902 log::info!("got server hello");
903 let hello_message_type_name = message.payload_type_name().to_string();
904 let hello = message
905 .into_any()
906 .downcast::<TypedEnvelope<proto::Hello>>()
907 .map_err(|_| {
908 anyhow!(
909 "invalid hello message received: {:?}",
910 hello_message_type_name
911 )
912 })?;
913 let peer_id = hello
914 .payload
915 .peer_id
916 .ok_or_else(|| anyhow!("invalid peer id"))?;
917 Ok(peer_id)
918 };
919
920 let peer_id = match peer_id.await {
921 Ok(peer_id) => peer_id,
922 Err(error) => {
923 self.peer.disconnect(connection_id);
924 return Err(error);
925 }
926 };
927
928 log::info!(
929 "set status to connected (connection id: {:?}, peer id: {:?})",
930 connection_id,
931 peer_id
932 );
933 self.set_status(
934 Status::Connected {
935 peer_id,
936 connection_id,
937 },
938 cx,
939 );
940
941 cx.spawn({
942 let this = self.clone();
943 |cx| {
944 async move {
945 while let Some(message) = incoming.next().await {
946 this.handle_message(message, &cx);
947 // Don't starve the main thread when receiving lots of messages at once.
948 smol::future::yield_now().await;
949 }
950 }
951 }
952 })
953 .detach();
954
955 cx.spawn({
956 let this = self.clone();
957 move |cx| async move {
958 match handle_io.await {
959 Ok(()) => {
960 if *this.status().borrow()
961 == (Status::Connected {
962 connection_id,
963 peer_id,
964 })
965 {
966 this.set_status(Status::SignedOut, &cx);
967 }
968 }
969 Err(err) => {
970 log::error!("connection error: {:?}", err);
971 this.set_status(Status::ConnectionLost, &cx);
972 }
973 }
974 }
975 })
976 .detach();
977
978 Ok(())
979 }
980
981 fn authenticate(self: &Arc<Self>, cx: &AsyncAppContext) -> Task<Result<Credentials>> {
982 #[cfg(any(test, feature = "test-support"))]
983 if let Some(callback) = self.authenticate.read().as_ref() {
984 return callback(cx);
985 }
986
987 self.authenticate_with_browser(cx)
988 }
989
990 fn establish_connection(
991 self: &Arc<Self>,
992 credentials: &Credentials,
993 cx: &AsyncAppContext,
994 ) -> Task<Result<Connection, EstablishConnectionError>> {
995 #[cfg(any(test, feature = "test-support"))]
996 if let Some(callback) = self.establish_connection.read().as_ref() {
997 return callback(credentials, cx);
998 }
999
1000 self.establish_websocket_connection(credentials, cx)
1001 }
1002
1003 async fn get_rpc_url(
1004 http: Arc<HttpClientWithUrl>,
1005 release_channel: Option<ReleaseChannel>,
1006 ) -> Result<Url> {
1007 if let Some(url) = &*ZED_RPC_URL {
1008 return Url::parse(url).context("invalid rpc url");
1009 }
1010
1011 let mut url = http.build_url("/rpc");
1012 if let Some(preview_param) =
1013 release_channel.and_then(|channel| channel.release_query_param())
1014 {
1015 url += "?";
1016 url += preview_param;
1017 }
1018 let response = http.get(&url, Default::default(), false).await?;
1019 let collab_url = if response.status().is_redirection() {
1020 response
1021 .headers()
1022 .get("Location")
1023 .ok_or_else(|| anyhow!("missing location header in /rpc response"))?
1024 .to_str()
1025 .map_err(EstablishConnectionError::other)?
1026 .to_string()
1027 } else {
1028 Err(anyhow!(
1029 "unexpected /rpc response status {}",
1030 response.status()
1031 ))?
1032 };
1033
1034 Url::parse(&collab_url).context("invalid rpc url")
1035 }
1036
1037 fn establish_websocket_connection(
1038 self: &Arc<Self>,
1039 credentials: &Credentials,
1040 cx: &AsyncAppContext,
1041 ) -> Task<Result<Connection, EstablishConnectionError>> {
1042 let release_channel = cx
1043 .update(|cx| ReleaseChannel::try_global(cx))
1044 .ok()
1045 .flatten();
1046 let app_version = cx
1047 .update(|cx| AppVersion::global(cx).to_string())
1048 .ok()
1049 .unwrap_or_default();
1050
1051 let request = Request::builder()
1052 .header("Authorization", credentials.authorization_header())
1053 .header("x-zed-protocol-version", rpc::PROTOCOL_VERSION)
1054 .header("x-zed-app-version", app_version)
1055 .header(
1056 "x-zed-release-channel",
1057 release_channel.map(|r| r.dev_name()).unwrap_or("unknown"),
1058 );
1059
1060 let http = self.http.clone();
1061 cx.background_executor().spawn(async move {
1062 let mut rpc_url = Self::get_rpc_url(http, release_channel).await?;
1063 let rpc_host = rpc_url
1064 .host_str()
1065 .zip(rpc_url.port_or_known_default())
1066 .ok_or_else(|| anyhow!("missing host in rpc url"))?;
1067 let stream = smol::net::TcpStream::connect(rpc_host).await?;
1068
1069 log::info!("connected to rpc endpoint {}", rpc_url);
1070
1071 match rpc_url.scheme() {
1072 "https" => {
1073 rpc_url.set_scheme("wss").unwrap();
1074 let request = request.uri(rpc_url.as_str()).body(())?;
1075 let (stream, _) =
1076 async_tungstenite::async_std::client_async_tls(request, stream).await?;
1077 Ok(Connection::new(
1078 stream
1079 .map_err(|error| anyhow!(error))
1080 .sink_map_err(|error| anyhow!(error)),
1081 ))
1082 }
1083 "http" => {
1084 rpc_url.set_scheme("ws").unwrap();
1085 let request = request.uri(rpc_url.as_str()).body(())?;
1086 let (stream, _) = async_tungstenite::client_async(request, stream).await?;
1087 Ok(Connection::new(
1088 stream
1089 .map_err(|error| anyhow!(error))
1090 .sink_map_err(|error| anyhow!(error)),
1091 ))
1092 }
1093 _ => Err(anyhow!("invalid rpc url: {}", rpc_url))?,
1094 }
1095 })
1096 }
1097
1098 pub fn authenticate_with_browser(
1099 self: &Arc<Self>,
1100 cx: &AsyncAppContext,
1101 ) -> Task<Result<Credentials>> {
1102 let http = self.http.clone();
1103 cx.spawn(|cx| async move {
1104 let background = cx.background_executor().clone();
1105
1106 let (open_url_tx, open_url_rx) = oneshot::channel::<String>();
1107 cx.update(|cx| {
1108 cx.spawn(move |cx| async move {
1109 let url = open_url_rx.await?;
1110 cx.update(|cx| cx.open_url(&url))
1111 })
1112 .detach_and_log_err(cx);
1113 })
1114 .log_err();
1115
1116 let credentials = background
1117 .clone()
1118 .spawn(async move {
1119 // Generate a pair of asymmetric encryption keys. The public key will be used by the
1120 // zed server to encrypt the user's access token, so that it can'be intercepted by
1121 // any other app running on the user's device.
1122 let (public_key, private_key) =
1123 rpc::auth::keypair().expect("failed to generate keypair for auth");
1124 let public_key_string = String::try_from(public_key)
1125 .expect("failed to serialize public key for auth");
1126
1127 if let Some((login, token)) =
1128 IMPERSONATE_LOGIN.as_ref().zip(ADMIN_API_TOKEN.as_ref())
1129 {
1130 return Self::authenticate_as_admin(http, login.clone(), token.clone())
1131 .await;
1132 }
1133
1134 // Start an HTTP server to receive the redirect from Zed's sign-in page.
1135 let server =
1136 tiny_http::Server::http("127.0.0.1:0").expect("failed to find open port");
1137 let port = server.server_addr().port();
1138
1139 // Open the Zed sign-in page in the user's browser, with query parameters that indicate
1140 // that the user is signing in from a Zed app running on the same device.
1141 let mut url = http.build_url(&format!(
1142 "/native_app_signin?native_app_port={}&native_app_public_key={}",
1143 port, public_key_string
1144 ));
1145
1146 if let Some(impersonate_login) = IMPERSONATE_LOGIN.as_ref() {
1147 log::info!("impersonating user @{}", impersonate_login);
1148 write!(&mut url, "&impersonate={}", impersonate_login).unwrap();
1149 }
1150
1151 open_url_tx.send(url).log_err();
1152
1153 // Receive the HTTP request from the user's browser. Retrieve the user id and encrypted
1154 // access token from the query params.
1155 //
1156 // TODO - Avoid ever starting more than one HTTP server. Maybe switch to using a
1157 // custom URL scheme instead of this local HTTP server.
1158 let (user_id, access_token) = background
1159 .spawn(async move {
1160 for _ in 0..100 {
1161 if let Some(req) = server.recv_timeout(Duration::from_secs(1))? {
1162 let path = req.url();
1163 let mut user_id = None;
1164 let mut access_token = None;
1165 let url = Url::parse(&format!("http://example.com{}", path))
1166 .context("failed to parse login notification url")?;
1167 for (key, value) in url.query_pairs() {
1168 if key == "access_token" {
1169 access_token = Some(value.to_string());
1170 } else if key == "user_id" {
1171 user_id = Some(value.to_string());
1172 }
1173 }
1174
1175 let post_auth_url =
1176 http.build_url("/native_app_signin_succeeded");
1177 req.respond(
1178 tiny_http::Response::empty(302).with_header(
1179 tiny_http::Header::from_bytes(
1180 &b"Location"[..],
1181 post_auth_url.as_bytes(),
1182 )
1183 .unwrap(),
1184 ),
1185 )
1186 .context("failed to respond to login http request")?;
1187 return Ok((
1188 user_id
1189 .ok_or_else(|| anyhow!("missing user_id parameter"))?,
1190 access_token.ok_or_else(|| {
1191 anyhow!("missing access_token parameter")
1192 })?,
1193 ));
1194 }
1195 }
1196
1197 Err(anyhow!("didn't receive login redirect"))
1198 })
1199 .await?;
1200
1201 let access_token = private_key
1202 .decrypt_string(&access_token)
1203 .context("failed to decrypt access token")?;
1204
1205 Ok(Credentials::User {
1206 user_id: user_id.parse()?,
1207 access_token,
1208 })
1209 })
1210 .await?;
1211
1212 cx.update(|cx| cx.activate(true))?;
1213 Ok(credentials)
1214 })
1215 }
1216
1217 async fn authenticate_as_admin(
1218 http: Arc<HttpClientWithUrl>,
1219 login: String,
1220 mut api_token: String,
1221 ) -> Result<Credentials> {
1222 #[derive(Deserialize)]
1223 struct AuthenticatedUserResponse {
1224 user: User,
1225 }
1226
1227 #[derive(Deserialize)]
1228 struct User {
1229 id: u64,
1230 }
1231
1232 // Use the collab server's admin API to retrieve the id
1233 // of the impersonated user.
1234 let mut url = Self::get_rpc_url(http.clone(), None).await?;
1235 url.set_path("/user");
1236 url.set_query(Some(&format!("github_login={login}")));
1237 let request = Request::get(url.as_str())
1238 .header("Authorization", format!("token {api_token}"))
1239 .body("".into())?;
1240
1241 let mut response = http.send(request).await?;
1242 let mut body = String::new();
1243 response.body_mut().read_to_string(&mut body).await?;
1244 if !response.status().is_success() {
1245 Err(anyhow!(
1246 "admin user request failed {} - {}",
1247 response.status().as_u16(),
1248 body,
1249 ))?;
1250 }
1251 let response: AuthenticatedUserResponse = serde_json::from_str(&body)?;
1252
1253 // Use the admin API token to authenticate as the impersonated user.
1254 api_token.insert_str(0, "ADMIN_TOKEN:");
1255 Ok(Credentials::User {
1256 user_id: response.user.id,
1257 access_token: api_token,
1258 })
1259 }
1260
1261 pub fn disconnect(self: &Arc<Self>, cx: &AsyncAppContext) {
1262 self.peer.teardown();
1263 self.set_status(Status::SignedOut, cx);
1264 }
1265
1266 pub fn reconnect(self: &Arc<Self>, cx: &AsyncAppContext) {
1267 self.peer.teardown();
1268 self.set_status(Status::ConnectionLost, cx);
1269 }
1270
1271 fn connection_id(&self) -> Result<ConnectionId> {
1272 if let Status::Connected { connection_id, .. } = *self.status().borrow() {
1273 Ok(connection_id)
1274 } else {
1275 Err(anyhow!("not connected"))
1276 }
1277 }
1278
1279 pub fn send<T: EnvelopedMessage>(&self, message: T) -> Result<()> {
1280 log::debug!("rpc send. client_id:{}, name:{}", self.id(), T::NAME);
1281 self.peer.send(self.connection_id()?, message)
1282 }
1283
1284 pub fn request<T: RequestMessage>(
1285 &self,
1286 request: T,
1287 ) -> impl Future<Output = Result<T::Response>> {
1288 self.request_envelope(request)
1289 .map_ok(|envelope| envelope.payload)
1290 }
1291
1292 pub fn request_stream<T: RequestMessage>(
1293 &self,
1294 request: T,
1295 ) -> impl Future<Output = Result<impl Stream<Item = Result<T::Response>>>> {
1296 let client_id = self.id.load(Ordering::SeqCst);
1297 log::debug!(
1298 "rpc request start. client_id:{}. name:{}",
1299 client_id,
1300 T::NAME
1301 );
1302 let response = self
1303 .connection_id()
1304 .map(|conn_id| self.peer.request_stream(conn_id, request));
1305 async move {
1306 let response = response?.await;
1307 log::debug!(
1308 "rpc request finish. client_id:{}. name:{}",
1309 client_id,
1310 T::NAME
1311 );
1312 response
1313 }
1314 }
1315
1316 pub fn request_envelope<T: RequestMessage>(
1317 &self,
1318 request: T,
1319 ) -> impl Future<Output = Result<TypedEnvelope<T::Response>>> {
1320 let client_id = self.id();
1321 log::debug!(
1322 "rpc request start. client_id:{}. name:{}",
1323 client_id,
1324 T::NAME
1325 );
1326 let response = self
1327 .connection_id()
1328 .map(|conn_id| self.peer.request_envelope(conn_id, request));
1329 async move {
1330 let response = response?.await;
1331 log::debug!(
1332 "rpc request finish. client_id:{}. name:{}",
1333 client_id,
1334 T::NAME
1335 );
1336 response
1337 }
1338 }
1339
1340 fn respond<T: RequestMessage>(&self, receipt: Receipt<T>, response: T::Response) -> Result<()> {
1341 log::debug!("rpc respond. client_id:{}. name:{}", self.id(), T::NAME);
1342 self.peer.respond(receipt, response)
1343 }
1344
1345 fn respond_with_error<T: RequestMessage>(
1346 &self,
1347 receipt: Receipt<T>,
1348 error: proto::Error,
1349 ) -> Result<()> {
1350 log::debug!("rpc respond. client_id:{}. name:{}", self.id(), T::NAME);
1351 self.peer.respond_with_error(receipt, error)
1352 }
1353
1354 fn handle_message(
1355 self: &Arc<Client>,
1356 message: Box<dyn AnyTypedEnvelope>,
1357 cx: &AsyncAppContext,
1358 ) {
1359 let mut state = self.state.write();
1360 let type_name = message.payload_type_name();
1361 let payload_type_id = message.payload_type_id();
1362 let sender_id = message.original_sender_id();
1363
1364 let mut subscriber = None;
1365
1366 if let Some(handle) = state
1367 .models_by_message_type
1368 .get(&payload_type_id)
1369 .and_then(|handle| handle.upgrade())
1370 {
1371 subscriber = Some(handle);
1372 } else if let Some((extract_entity_id, entity_type_id)) =
1373 state.entity_id_extractors.get(&payload_type_id).zip(
1374 state
1375 .entity_types_by_message_type
1376 .get(&payload_type_id)
1377 .copied(),
1378 )
1379 {
1380 let entity_id = (extract_entity_id)(message.as_ref());
1381
1382 match state
1383 .entities_by_type_and_remote_id
1384 .get_mut(&(entity_type_id, entity_id))
1385 {
1386 Some(WeakSubscriber::Pending(pending)) => {
1387 pending.push(message);
1388 return;
1389 }
1390 Some(weak_subscriber) => match weak_subscriber {
1391 WeakSubscriber::Entity { handle } => {
1392 subscriber = handle.upgrade();
1393 }
1394
1395 WeakSubscriber::Pending(_) => {}
1396 },
1397 _ => {}
1398 }
1399 }
1400
1401 let subscriber = if let Some(subscriber) = subscriber {
1402 subscriber
1403 } else {
1404 log::info!("unhandled message {}", type_name);
1405 self.peer.respond_with_unhandled_message(message).log_err();
1406 return;
1407 };
1408
1409 let handler = state.message_handlers.get(&payload_type_id).cloned();
1410 // Dropping the state prevents deadlocks if the handler interacts with rpc::Client.
1411 // It also ensures we don't hold the lock while yielding back to the executor, as
1412 // that might cause the executor thread driving this future to block indefinitely.
1413 drop(state);
1414
1415 if let Some(handler) = handler {
1416 let future = handler(subscriber, message, self, cx.clone());
1417 let client_id = self.id();
1418 log::debug!(
1419 "rpc message received. client_id:{}, sender_id:{:?}, type:{}",
1420 client_id,
1421 sender_id,
1422 type_name
1423 );
1424 cx.spawn(move |_| async move {
1425 match future.await {
1426 Ok(()) => {
1427 log::debug!(
1428 "rpc message handled. client_id:{}, sender_id:{:?}, type:{}",
1429 client_id,
1430 sender_id,
1431 type_name
1432 );
1433 }
1434 Err(error) => {
1435 log::error!(
1436 "error handling message. client_id:{}, sender_id:{:?}, type:{}, error:{:?}",
1437 client_id,
1438 sender_id,
1439 type_name,
1440 error
1441 );
1442 }
1443 }
1444 })
1445 .detach();
1446 } else {
1447 log::info!("unhandled message {}", type_name);
1448 self.peer.respond_with_unhandled_message(message).log_err();
1449 }
1450 }
1451
1452 pub fn telemetry(&self) -> &Arc<Telemetry> {
1453 &self.telemetry
1454 }
1455}
1456
1457async fn read_credentials_from_keychain(cx: &AsyncAppContext) -> Option<Credentials> {
1458 if IMPERSONATE_LOGIN.is_some() {
1459 return None;
1460 }
1461
1462 let (user_id, access_token) = cx
1463 .update(|cx| cx.read_credentials(&ClientSettings::get_global(cx).server_url))
1464 .log_err()?
1465 .await
1466 .log_err()??;
1467
1468 Some(Credentials::User {
1469 user_id: user_id.parse().ok()?,
1470 access_token: String::from_utf8(access_token).ok()?,
1471 })
1472}
1473
1474async fn write_credentials_to_keychain(
1475 user_id: u64,
1476 access_token: String,
1477 cx: &AsyncAppContext,
1478) -> Result<()> {
1479 cx.update(move |cx| {
1480 cx.write_credentials(
1481 &ClientSettings::get_global(cx).server_url,
1482 &user_id.to_string(),
1483 access_token.as_bytes(),
1484 )
1485 })?
1486 .await
1487}
1488
1489async fn delete_credentials_from_keychain(cx: &AsyncAppContext) -> Result<()> {
1490 cx.update(move |cx| cx.delete_credentials(&ClientSettings::get_global(cx).server_url))?
1491 .await
1492}
1493
1494/// prefix for the zed:// url scheme
1495pub static ZED_URL_SCHEME: &str = "zed";
1496
1497/// Parses the given link into a Zed link.
1498///
1499/// Returns a [`Some`] containing the unprefixed link if the link is a Zed link.
1500/// Returns [`None`] otherwise.
1501pub fn parse_zed_link<'a>(link: &'a str, cx: &AppContext) -> Option<&'a str> {
1502 let server_url = &ClientSettings::get_global(cx).server_url;
1503 if let Some(stripped) = link
1504 .strip_prefix(server_url)
1505 .and_then(|result| result.strip_prefix('/'))
1506 {
1507 return Some(stripped);
1508 }
1509 if let Some(stripped) = link
1510 .strip_prefix(ZED_URL_SCHEME)
1511 .and_then(|result| result.strip_prefix("://"))
1512 {
1513 return Some(stripped);
1514 }
1515
1516 None
1517}
1518
1519#[cfg(test)]
1520mod tests {
1521 use super::*;
1522 use crate::test::FakeServer;
1523
1524 use clock::FakeSystemClock;
1525 use gpui::{BackgroundExecutor, Context, TestAppContext};
1526 use parking_lot::Mutex;
1527 use settings::SettingsStore;
1528 use std::future;
1529 use util::http::FakeHttpClient;
1530
1531 #[gpui::test(iterations = 10)]
1532 async fn test_reconnection(cx: &mut TestAppContext) {
1533 init_test(cx);
1534 let user_id = 5;
1535 let client = cx.update(|cx| {
1536 Client::new(
1537 Arc::new(FakeSystemClock::default()),
1538 FakeHttpClient::with_404_response(),
1539 cx,
1540 )
1541 });
1542 let server = FakeServer::for_client(user_id, &client, cx).await;
1543 let mut status = client.status();
1544 assert!(matches!(
1545 status.next().await,
1546 Some(Status::Connected { .. })
1547 ));
1548 assert_eq!(server.auth_count(), 1);
1549
1550 server.forbid_connections();
1551 server.disconnect();
1552 while !matches!(status.next().await, Some(Status::ReconnectionError { .. })) {}
1553
1554 server.allow_connections();
1555 cx.executor().advance_clock(Duration::from_secs(10));
1556 while !matches!(status.next().await, Some(Status::Connected { .. })) {}
1557 assert_eq!(server.auth_count(), 1); // Client reused the cached credentials when reconnecting
1558
1559 server.forbid_connections();
1560 server.disconnect();
1561 while !matches!(status.next().await, Some(Status::ReconnectionError { .. })) {}
1562
1563 // Clear cached credentials after authentication fails
1564 server.roll_access_token();
1565 server.allow_connections();
1566 cx.executor().run_until_parked();
1567 cx.executor().advance_clock(Duration::from_secs(10));
1568 while !matches!(status.next().await, Some(Status::Connected { .. })) {}
1569 assert_eq!(server.auth_count(), 2); // Client re-authenticated due to an invalid token
1570 }
1571
1572 #[gpui::test(iterations = 10)]
1573 async fn test_connection_timeout(executor: BackgroundExecutor, cx: &mut TestAppContext) {
1574 init_test(cx);
1575 let user_id = 5;
1576 let client = cx.update(|cx| {
1577 Client::new(
1578 Arc::new(FakeSystemClock::default()),
1579 FakeHttpClient::with_404_response(),
1580 cx,
1581 )
1582 });
1583 let mut status = client.status();
1584
1585 // Time out when client tries to connect.
1586 client.override_authenticate(move |cx| {
1587 cx.background_executor().spawn(async move {
1588 Ok(Credentials::User {
1589 user_id,
1590 access_token: "token".into(),
1591 })
1592 })
1593 });
1594 client.override_establish_connection(|_, cx| {
1595 cx.background_executor().spawn(async move {
1596 future::pending::<()>().await;
1597 unreachable!()
1598 })
1599 });
1600 let auth_and_connect = cx.spawn({
1601 let client = client.clone();
1602 |cx| async move { client.authenticate_and_connect(false, &cx).await }
1603 });
1604 executor.run_until_parked();
1605 assert!(matches!(status.next().await, Some(Status::Connecting)));
1606
1607 executor.advance_clock(CONNECTION_TIMEOUT);
1608 assert!(matches!(
1609 status.next().await,
1610 Some(Status::ConnectionError { .. })
1611 ));
1612 auth_and_connect.await.unwrap_err();
1613
1614 // Allow the connection to be established.
1615 let server = FakeServer::for_client(user_id, &client, cx).await;
1616 assert!(matches!(
1617 status.next().await,
1618 Some(Status::Connected { .. })
1619 ));
1620
1621 // Disconnect client.
1622 server.forbid_connections();
1623 server.disconnect();
1624 while !matches!(status.next().await, Some(Status::ReconnectionError { .. })) {}
1625
1626 // Time out when re-establishing the connection.
1627 server.allow_connections();
1628 client.override_establish_connection(|_, cx| {
1629 cx.background_executor().spawn(async move {
1630 future::pending::<()>().await;
1631 unreachable!()
1632 })
1633 });
1634 executor.advance_clock(2 * INITIAL_RECONNECTION_DELAY);
1635 assert!(matches!(
1636 status.next().await,
1637 Some(Status::Reconnecting { .. })
1638 ));
1639
1640 executor.advance_clock(CONNECTION_TIMEOUT);
1641 assert!(matches!(
1642 status.next().await,
1643 Some(Status::ReconnectionError { .. })
1644 ));
1645 }
1646
1647 #[gpui::test(iterations = 10)]
1648 async fn test_authenticating_more_than_once(
1649 cx: &mut TestAppContext,
1650 executor: BackgroundExecutor,
1651 ) {
1652 init_test(cx);
1653 let auth_count = Arc::new(Mutex::new(0));
1654 let dropped_auth_count = Arc::new(Mutex::new(0));
1655 let client = cx.update(|cx| {
1656 Client::new(
1657 Arc::new(FakeSystemClock::default()),
1658 FakeHttpClient::with_404_response(),
1659 cx,
1660 )
1661 });
1662 client.override_authenticate({
1663 let auth_count = auth_count.clone();
1664 let dropped_auth_count = dropped_auth_count.clone();
1665 move |cx| {
1666 let auth_count = auth_count.clone();
1667 let dropped_auth_count = dropped_auth_count.clone();
1668 cx.background_executor().spawn(async move {
1669 *auth_count.lock() += 1;
1670 let _drop = util::defer(move || *dropped_auth_count.lock() += 1);
1671 future::pending::<()>().await;
1672 unreachable!()
1673 })
1674 }
1675 });
1676
1677 let _authenticate = cx.spawn({
1678 let client = client.clone();
1679 move |cx| async move { client.authenticate_and_connect(false, &cx).await }
1680 });
1681 executor.run_until_parked();
1682 assert_eq!(*auth_count.lock(), 1);
1683 assert_eq!(*dropped_auth_count.lock(), 0);
1684
1685 let _authenticate = cx.spawn({
1686 let client = client.clone();
1687 |cx| async move { client.authenticate_and_connect(false, &cx).await }
1688 });
1689 executor.run_until_parked();
1690 assert_eq!(*auth_count.lock(), 2);
1691 assert_eq!(*dropped_auth_count.lock(), 1);
1692 }
1693
1694 #[gpui::test]
1695 async fn test_subscribing_to_entity(cx: &mut TestAppContext) {
1696 init_test(cx);
1697 let user_id = 5;
1698 let client = cx.update(|cx| {
1699 Client::new(
1700 Arc::new(FakeSystemClock::default()),
1701 FakeHttpClient::with_404_response(),
1702 cx,
1703 )
1704 });
1705 let server = FakeServer::for_client(user_id, &client, cx).await;
1706
1707 let (done_tx1, mut done_rx1) = smol::channel::unbounded();
1708 let (done_tx2, mut done_rx2) = smol::channel::unbounded();
1709 client.add_model_message_handler(
1710 move |model: Model<TestModel>, _: TypedEnvelope<proto::JoinProject>, _, mut cx| {
1711 match model.update(&mut cx, |model, _| model.id).unwrap() {
1712 1 => done_tx1.try_send(()).unwrap(),
1713 2 => done_tx2.try_send(()).unwrap(),
1714 _ => unreachable!(),
1715 }
1716 async { Ok(()) }
1717 },
1718 );
1719 let model1 = cx.new_model(|_| TestModel {
1720 id: 1,
1721 subscription: None,
1722 });
1723 let model2 = cx.new_model(|_| TestModel {
1724 id: 2,
1725 subscription: None,
1726 });
1727 let model3 = cx.new_model(|_| TestModel {
1728 id: 3,
1729 subscription: None,
1730 });
1731
1732 let _subscription1 = client
1733 .subscribe_to_entity(1)
1734 .unwrap()
1735 .set_model(&model1, &mut cx.to_async());
1736 let _subscription2 = client
1737 .subscribe_to_entity(2)
1738 .unwrap()
1739 .set_model(&model2, &mut cx.to_async());
1740 // Ensure dropping a subscription for the same entity type still allows receiving of
1741 // messages for other entity IDs of the same type.
1742 let subscription3 = client
1743 .subscribe_to_entity(3)
1744 .unwrap()
1745 .set_model(&model3, &mut cx.to_async());
1746 drop(subscription3);
1747
1748 server.send(proto::JoinProject { project_id: 1 });
1749 server.send(proto::JoinProject { project_id: 2 });
1750 done_rx1.next().await.unwrap();
1751 done_rx2.next().await.unwrap();
1752 }
1753
1754 #[gpui::test]
1755 async fn test_subscribing_after_dropping_subscription(cx: &mut TestAppContext) {
1756 init_test(cx);
1757 let user_id = 5;
1758 let client = cx.update(|cx| {
1759 Client::new(
1760 Arc::new(FakeSystemClock::default()),
1761 FakeHttpClient::with_404_response(),
1762 cx,
1763 )
1764 });
1765 let server = FakeServer::for_client(user_id, &client, cx).await;
1766
1767 let model = cx.new_model(|_| TestModel::default());
1768 let (done_tx1, _done_rx1) = smol::channel::unbounded();
1769 let (done_tx2, mut done_rx2) = smol::channel::unbounded();
1770 let subscription1 = client.add_message_handler(
1771 model.downgrade(),
1772 move |_, _: TypedEnvelope<proto::Ping>, _, _| {
1773 done_tx1.try_send(()).unwrap();
1774 async { Ok(()) }
1775 },
1776 );
1777 drop(subscription1);
1778 let _subscription2 = client.add_message_handler(
1779 model.downgrade(),
1780 move |_, _: TypedEnvelope<proto::Ping>, _, _| {
1781 done_tx2.try_send(()).unwrap();
1782 async { Ok(()) }
1783 },
1784 );
1785 server.send(proto::Ping {});
1786 done_rx2.next().await.unwrap();
1787 }
1788
1789 #[gpui::test]
1790 async fn test_dropping_subscription_in_handler(cx: &mut TestAppContext) {
1791 init_test(cx);
1792 let user_id = 5;
1793 let client = cx.update(|cx| {
1794 Client::new(
1795 Arc::new(FakeSystemClock::default()),
1796 FakeHttpClient::with_404_response(),
1797 cx,
1798 )
1799 });
1800 let server = FakeServer::for_client(user_id, &client, cx).await;
1801
1802 let model = cx.new_model(|_| TestModel::default());
1803 let (done_tx, mut done_rx) = smol::channel::unbounded();
1804 let subscription = client.add_message_handler(
1805 model.clone().downgrade(),
1806 move |model: Model<TestModel>, _: TypedEnvelope<proto::Ping>, _, mut cx| {
1807 model
1808 .update(&mut cx, |model, _| model.subscription.take())
1809 .unwrap();
1810 done_tx.try_send(()).unwrap();
1811 async { Ok(()) }
1812 },
1813 );
1814 model.update(cx, |model, _| {
1815 model.subscription = Some(subscription);
1816 });
1817 server.send(proto::Ping {});
1818 done_rx.next().await.unwrap();
1819 }
1820
1821 #[derive(Default)]
1822 struct TestModel {
1823 id: usize,
1824 subscription: Option<Subscription>,
1825 }
1826
1827 fn init_test(cx: &mut TestAppContext) {
1828 cx.update(|cx| {
1829 let settings_store = SettingsStore::test(cx);
1830 cx.set_global(settings_store);
1831 init_settings(cx);
1832 });
1833 }
1834}