Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Checkpoint

Browse source
This commit is contained in:
Antonio Scandurra 2023-10-22 18:25:24 +02:00
parent db6a3e1783
commit ce75be91e1
9 changed files with 111 additions and 244 deletions
Download patch file Download diff file Expand all files Collapse all files

90
crates/client2/src/client2.rs
View file

@ -14,8 +14,8 @@ use futures::{
future::BoxFuture, AsyncReadExt, FutureExt, SinkExt, StreamExt, TryFutureExt as _, TryStreamExt,
};
use gpui2::{
serde_json, AnyHandle, AnyWeakHandle, AnyWindowHandle, AppContext, AsyncAppContext, Handle,
SemanticVersion, Task, ViewContext, WeakHandle,
serde_json, AnyHandle, AnyWeakHandle, AppContext, AsyncAppContext, Handle, SemanticVersion,
Task, WeakHandle,
};
use lazy_static::lazy_static;
use parking_lot::RwLock;
@ -235,7 +235,7 @@ struct ClientState {
dyn Send
+ Sync
+ Fn(
Subscriber,
AnyHandle,
Box<dyn AnyTypedEnvelope>,
&Arc<Client>,
AsyncAppContext,
@ -245,18 +245,10 @@ struct ClientState {
}
enum WeakSubscriber {
Entity {
handle: AnyWeakHandle,
window_handle: Option<AnyWindowHandle>,
},
Entity { handle: AnyWeakHandle },
Pending(Vec<Box<dyn AnyTypedEnvelope>>),
}
struct Subscriber {
handle: AnyHandle,
window_handle: Option<AnyWindowHandle>,
}
#[derive(Clone, Debug)]
pub struct Credentials {
pub user_id: u64,
@ -310,7 +302,7 @@ impl Drop for Subscription {
}
}
pub struct PendingEntitySubscription<T> {
pub struct PendingEntitySubscription<T: 'static> {
client: Arc<Client>,
remote_id: u64,
_entity_type: PhantomData<T>,
@ -335,7 +327,6 @@ where
id,
WeakSubscriber::Entity {
handle: model.downgrade().into(),
window_handle: None,
},
);
drop(state);
@ -349,7 +340,10 @@ where
}
}
impl<T> Drop for PendingEntitySubscription<T> {
impl<T> Drop for PendingEntitySubscription<T>
where
T: 'static,
{
fn drop(&mut self) {
if !self.consumed {
let mut state = self.client.state.write();
@ -494,7 +488,7 @@ impl Client {
Status::ConnectionLost => {
let this = self.clone();
let reconnect_interval = state.reconnect_interval;
state._reconnect_task = Some(cx.spawn(|cx| async move {
state._reconnect_task = Some(cx.spawn(move |cx| async move {
#[cfg(any(test, feature = "test-support"))]
let mut rng = StdRng::seed_from_u64(0);
#[cfg(not(any(test, feature = "test-support")))]
@ -521,39 +515,21 @@ impl Client {
}));
}
Status::SignedOut | Status::UpgradeRequired => {
cx.update(|cx| self.telemetry.set_authenticated_user_info(None, false, cx));
cx.update(|cx| self.telemetry.set_authenticated_user_info(None, false, cx))
.log_err();
state._reconnect_task.take();
}
_ => {}
}
}
pub fn add_view_for_remote_entity<T>(
self: &Arc<Self>,
remote_id: u64,
cx: &mut ViewContext<T>,
) -> Subscription
where
T: 'static + Send + Sync,
{
let id = (TypeId::of::<T>(), remote_id);
self.state.write().entities_by_type_and_remote_id.insert(
id,
WeakSubscriber::Entity {
handle: cx.handle().into(),
window_handle: Some(cx.window_handle()),
},
);
Subscription::Entity {
client: Arc::downgrade(self),
id,
}
}
pub fn subscribe_to_entity<T>(
self: &Arc<Self>,
remote_id: u64,
) -> Result<PendingEntitySubscription<T>> {
) -> Result<PendingEntitySubscription<T>>
where
T: 'static + Send + Sync,
{
let id = (TypeId::of::<T>(), remote_id);
let mut state = self.state.write();
@ -594,7 +570,7 @@ impl Client {
let prev_handler = state.message_handlers.insert(
message_type_id,
Arc::new(move |subscriber, envelope, client, cx| {
let subscriber = subscriber.handle.downcast::<E>().unwrap();
let subscriber = subscriber.downcast::<E>().unwrap();
let envelope = envelope.into_any().downcast::<TypedEnvelope<M>>().unwrap();
handler(subscriber, *envelope, client.clone(), cx).boxed()
}),
@ -643,22 +619,15 @@ impl Client {
F: 'static + Future<Output = Result<()>> + Send,
{
self.add_entity_message_handler::<M, E, _, _>(move |subscriber, message, client, cx| {
handler(
subscriber.handle.downcast::<E>().unwrap(),
message,
client,
cx,
)
handler(subscriber.downcast::<E>().unwrap(), message, client, cx)
})
}
fn add_entity_message_handler<M, E, H, F>(self: &Arc<Self>, handler: H)
where
M: EntityMessage,
H: 'static
+ Send
+ Sync
+ Fn(Subscriber, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F,
E: 'static + Send + Sync,
H: 'static + Send + Sync + Fn(AnyHandle, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F,
F: 'static + Future<Output = Result<()>> + Send,
{
let model_type_id = TypeId::of::<E>();
@ -1079,7 +1048,7 @@ impl Client {
write!(&mut url, "&impersonate={}", impersonate_login).unwrap();
}
cx.run_on_main(|cx| cx.open_url(&url))?.await;
cx.run_on_main(move |cx| cx.open_url(&url))?.await;
// Receive the HTTP request from the user's browser. Retrieve the user id and encrypted
// access token from the query params.
@ -1269,10 +1238,7 @@ impl Client {
.get(&payload_type_id)
.and_then(|handle| handle.upgrade())
{
subscriber = Some(Subscriber {
handle,
window_handle: None,
});
subscriber = Some(handle);
} else if let Some((extract_entity_id, entity_type_id)) =
state.entity_id_extractors.get(&payload_type_id).zip(
state
@ -1292,14 +1258,8 @@ impl Client {
return;
}
Some(weak_subscriber @ _) => match weak_subscriber {
WeakSubscriber::Entity {
handle,
window_handle,
} => {
subscriber = handle.upgrade().map(|handle| Subscriber {
handle,
window_handle: window_handle.clone(),
});
WeakSubscriber::Entity { handle } => {
subscriber = handle.upgrade();
}
WeakSubscriber::Pending(_) => {}
@ -1331,7 +1291,7 @@ impl Client {
sender_id,
type_name
);
cx.spawn_on_main(|_| async move {
cx.spawn_on_main(move |_| async move {
match future.await {
Ok(()) => {
log::debug!(

2
crates/client2/src/telemetry.rs
View file

@ -159,7 +159,7 @@ impl Telemetry {
}
let this = self.clone();
cx.spawn(|mut cx| async move {
cx.spawn(|cx| async move {
let mut system = System::new_all();
system.refresh_all();

70
crates/client2/src/user.rs
View file

@ -1,7 +1,7 @@
use super::{proto, Client, Status, TypedEnvelope};
use anyhow::{anyhow, Context, Result};
use collections::{hash_map::Entry, HashMap, HashSet};
use feature_flags::FeatureFlagAppExt;
use feature_flags2::FeatureFlagAppExt;
use futures::{channel::mpsc, future, AsyncReadExt, Future, StreamExt};
use gpui2::{AsyncAppContext, EventEmitter, Handle, ImageData, ModelContext, Task};
use postage::{sink::Sink, watch};
@ -78,7 +78,7 @@ pub struct UserStore {
client: Weak<Client>,
http: Arc<dyn HttpClient>,
_maintain_contacts: Task<()>,
_maintain_current_user: Task<()>,
_maintain_current_user: Task<Result<()>>,
}
#[derive(Clone)]
@ -167,28 +167,20 @@ impl UserStore {
client.request(proto::GetPrivateUserInfo {}).log_err();
let (user, info) = futures::join!(fetch_user, fetch_metrics_id);
if let Some(info) = info {
cx.update(|cx| {
cx.update(|cx| {
if let Some(info) = info {
cx.update_flags(info.staff, info.flags);
client.telemetry.set_authenticated_user_info(
Some(info.metrics_id.clone()),
info.staff,
cx,
)
});
} else {
cx.read(|cx| {
client
.telemetry
.set_authenticated_user_info(None, false, cx)
});
}
}
})?;
current_user_tx.send(user).await.ok();
this.update(&mut cx, |_, cx| {
cx.notify();
});
this.update(&mut cx, |_, cx| cx.notify())?;
}
}
Status::SignedOut => {
@ -196,21 +188,20 @@ impl UserStore {
this.update(&mut cx, |this, cx| {
cx.notify();
this.clear_contacts()
})
})?
.await;
}
Status::ConnectionLost => {
if let Some(this) = this.upgrade() {
this.update(&mut cx, |this, cx| {
cx.notify();
this.clear_contacts()
})
.await;
}
this.update(&mut cx, |this, cx| {
cx.notify();
this.clear_contacts()
})?
.await;
}
_ => {}
}
}
Ok(())
}),
pending_contact_requests: Default::default(),
}
@ -233,7 +224,7 @@ impl UserStore {
count: message.payload.count,
});
cx.notify();
});
})?;
Ok(())
}
@ -243,7 +234,7 @@ impl UserStore {
_: Arc<Client>,
mut cx: AsyncAppContext,
) -> Result<()> {
this.update(&mut cx, |_, cx| cx.emit(Event::ShowContacts));
this.update(&mut cx, |_, cx| cx.emit(Event::ShowContacts))?;
Ok(())
}
@ -261,7 +252,7 @@ impl UserStore {
this.update_contacts_tx
.unbounded_send(UpdateContacts::Update(message.payload))
.unwrap();
});
})?;
Ok(())
}
@ -297,6 +288,9 @@ impl UserStore {
// Users are fetched in parallel above and cached in call to get_users
// No need to paralellize here
let mut updated_contacts = Vec::new();
let this = this
.upgrade()
.ok_or_else(|| anyhow!("can't upgrade user store handle"))?;
for contact in message.contacts {
let should_notify = contact.should_notify;
updated_contacts.push((
@ -309,7 +303,9 @@ impl UserStore {
for request in message.incoming_requests {
incoming_requests.push({
let user = this
.update(&mut cx, |this, cx| this.get_user(request.requester_id, cx))
.update(&mut cx, |this, cx| {
this.get_user(request.requester_id, cx)
})?
.await?;
(user, request.should_notify)
});
@ -318,7 +314,7 @@ impl UserStore {
let mut outgoing_requests = Vec::new();
for requested_user_id in message.outgoing_requests {
outgoing_requests.push(
this.update(&mut cx, |this, cx| this.get_user(requested_user_id, cx))
this.update(&mut cx, |this, cx| this.get_user(requested_user_id, cx))?
.await?,
);
}
@ -398,7 +394,7 @@ impl UserStore {
}
cx.notify();
});
})?;
Ok(())
})
@ -494,7 +490,7 @@ impl UserStore {
cx: &mut ModelContext<Self>,
) -> Task<Result<()>> {
let client = self.client.upgrade();
cx.spawn_weak(|_, _| async move {
cx.spawn(move |_, _| async move {
client
.ok_or_else(|| anyhow!("can't upgrade client reference"))?
.request(proto::RespondToContactRequest {
@ -516,7 +512,7 @@ impl UserStore {
*self.pending_contact_requests.entry(user_id).or_insert(0) += 1;
cx.notify();
cx.spawn(|this, mut cx| async move {
cx.spawn(move |this, mut cx| async move {
let response = client
.ok_or_else(|| anyhow!("can't upgrade client reference"))?
.request(request)
@ -531,7 +527,7 @@ impl UserStore {
}
}
cx.notify();
});
})?;
response?;
Ok(())
})
@ -574,11 +570,11 @@ impl UserStore {
},
cx,
)
})
})?
.await?;
}
this.read_with(&cx, |this, _| {
this.update(&mut cx, |this, _| {
user_ids
.iter()
.map(|user_id| {
@ -588,7 +584,7 @@ impl UserStore {
.ok_or_else(|| anyhow!("user {} not found", user_id))
})
.collect()
})
})?
})
}
@ -614,7 +610,7 @@ impl UserStore {
}
let load_users = self.get_users(vec![user_id], cx);
cx.spawn(|this, mut cx| async move {
cx.spawn(move |this, mut cx| async move {
load_users.await?;
this.update(&mut cx, |this, _| {
this.users
@ -700,7 +696,7 @@ impl Contact {
let user = user_store
.update(cx, |user_store, cx| {
user_store.get_user(contact.user_id, cx)
})
})?
.await?;
Ok(Self {
user,