1mod store;
2
3use crate::{
4 auth,
5 db::{self, ChannelId, MessageId, UserId},
6 AppState, Result,
7};
8use anyhow::anyhow;
9use async_tungstenite::tungstenite::{
10 protocol::CloseFrame as TungsteniteCloseFrame, Message as TungsteniteMessage,
11};
12use axum::{
13 body::Body,
14 extract::{
15 ws::{CloseFrame as AxumCloseFrame, Message as AxumMessage},
16 ConnectInfo, WebSocketUpgrade,
17 },
18 headers::{Header, HeaderName},
19 http::StatusCode,
20 middleware,
21 response::IntoResponse,
22 routing::get,
23 Extension, Router, TypedHeader,
24};
25use collections::HashMap;
26use futures::{channel::mpsc, future::BoxFuture, FutureExt, SinkExt, StreamExt, TryStreamExt};
27use lazy_static::lazy_static;
28use rpc::{
29 proto::{self, AnyTypedEnvelope, EntityMessage, EnvelopedMessage, RequestMessage},
30 Connection, ConnectionId, Peer, Receipt, TypedEnvelope,
31};
32use std::{
33 any::TypeId,
34 future::Future,
35 marker::PhantomData,
36 net::SocketAddr,
37 ops::{Deref, DerefMut},
38 rc::Rc,
39 sync::{
40 atomic::{AtomicBool, Ordering::SeqCst},
41 Arc,
42 },
43 time::Duration,
44};
45use store::{Store, Worktree};
46use time::OffsetDateTime;
47use tokio::{
48 sync::{RwLock, RwLockReadGuard, RwLockWriteGuard},
49 time::Sleep,
50};
51use tower::ServiceBuilder;
52use tracing::{info_span, Instrument};
53
54type MessageHandler =
55 Box<dyn Send + Sync + Fn(Arc<Server>, Box<dyn AnyTypedEnvelope>) -> BoxFuture<'static, ()>>;
56
57struct Response<R> {
58 server: Arc<Server>,
59 receipt: Receipt<R>,
60 responded: Arc<AtomicBool>,
61}
62
63impl<R: RequestMessage> Response<R> {
64 fn send(self, payload: R::Response) -> Result<()> {
65 self.responded.store(true, SeqCst);
66 self.server.peer.respond(self.receipt, payload)?;
67 Ok(())
68 }
69
70 fn into_receipt(self) -> Receipt<R> {
71 self.responded.store(true, SeqCst);
72 self.receipt
73 }
74}
75
76pub struct Server {
77 peer: Arc<Peer>,
78 store: RwLock<Store>,
79 app_state: Arc<AppState>,
80 handlers: HashMap<TypeId, MessageHandler>,
81 notifications: Option<mpsc::UnboundedSender<()>>,
82}
83
84pub trait Executor: Send + Clone {
85 type Sleep: Send + Future;
86 fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F);
87 fn sleep(&self, duration: Duration) -> Self::Sleep;
88}
89
90#[derive(Clone)]
91pub struct RealExecutor;
92
93const MESSAGE_COUNT_PER_PAGE: usize = 100;
94const MAX_MESSAGE_LEN: usize = 1024;
95
96struct StoreReadGuard<'a> {
97 guard: RwLockReadGuard<'a, Store>,
98 _not_send: PhantomData<Rc<()>>,
99}
100
101struct StoreWriteGuard<'a> {
102 guard: RwLockWriteGuard<'a, Store>,
103 _not_send: PhantomData<Rc<()>>,
104}
105
106impl Server {
107 pub fn new(
108 app_state: Arc<AppState>,
109 notifications: Option<mpsc::UnboundedSender<()>>,
110 ) -> Arc<Self> {
111 let mut server = Self {
112 peer: Peer::new(),
113 app_state,
114 store: Default::default(),
115 handlers: Default::default(),
116 notifications,
117 };
118
119 server
120 .add_request_handler(Server::ping)
121 .add_request_handler(Server::register_project)
122 .add_message_handler(Server::unregister_project)
123 .add_request_handler(Server::join_project)
124 .add_message_handler(Server::leave_project)
125 .add_message_handler(Server::respond_to_join_project_request)
126 .add_request_handler(Server::register_worktree)
127 .add_message_handler(Server::unregister_worktree)
128 .add_request_handler(Server::update_worktree)
129 .add_message_handler(Server::start_language_server)
130 .add_message_handler(Server::update_language_server)
131 .add_message_handler(Server::update_diagnostic_summary)
132 .add_request_handler(Server::forward_project_request::<proto::GetDefinition>)
133 .add_request_handler(Server::forward_project_request::<proto::GetReferences>)
134 .add_request_handler(Server::forward_project_request::<proto::SearchProject>)
135 .add_request_handler(Server::forward_project_request::<proto::GetDocumentHighlights>)
136 .add_request_handler(Server::forward_project_request::<proto::GetProjectSymbols>)
137 .add_request_handler(Server::forward_project_request::<proto::OpenBufferForSymbol>)
138 .add_request_handler(Server::forward_project_request::<proto::OpenBufferById>)
139 .add_request_handler(Server::forward_project_request::<proto::OpenBufferByPath>)
140 .add_request_handler(Server::forward_project_request::<proto::GetCompletions>)
141 .add_request_handler(
142 Server::forward_project_request::<proto::ApplyCompletionAdditionalEdits>,
143 )
144 .add_request_handler(Server::forward_project_request::<proto::GetCodeActions>)
145 .add_request_handler(Server::forward_project_request::<proto::ApplyCodeAction>)
146 .add_request_handler(Server::forward_project_request::<proto::PrepareRename>)
147 .add_request_handler(Server::forward_project_request::<proto::PerformRename>)
148 .add_request_handler(Server::forward_project_request::<proto::ReloadBuffers>)
149 .add_request_handler(Server::forward_project_request::<proto::FormatBuffers>)
150 .add_request_handler(Server::forward_project_request::<proto::CreateProjectEntry>)
151 .add_request_handler(Server::forward_project_request::<proto::RenameProjectEntry>)
152 .add_request_handler(Server::forward_project_request::<proto::DeleteProjectEntry>)
153 .add_request_handler(Server::update_buffer)
154 .add_message_handler(Server::update_buffer_file)
155 .add_message_handler(Server::buffer_reloaded)
156 .add_message_handler(Server::buffer_saved)
157 .add_request_handler(Server::save_buffer)
158 .add_request_handler(Server::get_channels)
159 .add_request_handler(Server::get_users)
160 .add_request_handler(Server::fuzzy_search_users)
161 .add_request_handler(Server::request_contact)
162 .add_request_handler(Server::remove_contact)
163 .add_request_handler(Server::respond_to_contact_request)
164 .add_request_handler(Server::join_channel)
165 .add_message_handler(Server::leave_channel)
166 .add_request_handler(Server::send_channel_message)
167 .add_request_handler(Server::follow)
168 .add_message_handler(Server::unfollow)
169 .add_message_handler(Server::update_followers)
170 .add_request_handler(Server::get_channel_messages);
171
172 Arc::new(server)
173 }
174
175 fn add_message_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
176 where
177 F: 'static + Send + Sync + Fn(Arc<Self>, TypedEnvelope<M>) -> Fut,
178 Fut: 'static + Send + Future<Output = Result<()>>,
179 M: EnvelopedMessage,
180 {
181 let prev_handler = self.handlers.insert(
182 TypeId::of::<M>(),
183 Box::new(move |server, envelope| {
184 let envelope = envelope.into_any().downcast::<TypedEnvelope<M>>().unwrap();
185 let span = info_span!(
186 "handle message",
187 payload_type = envelope.payload_type_name(),
188 payload = format!("{:?}", envelope.payload).as_str(),
189 );
190 let future = (handler)(server, *envelope);
191 async move {
192 if let Err(error) = future.await {
193 tracing::error!(%error, "error handling message");
194 }
195 }
196 .instrument(span)
197 .boxed()
198 }),
199 );
200 if prev_handler.is_some() {
201 panic!("registered a handler for the same message twice");
202 }
203 self
204 }
205
206 /// Handle a request while holding a lock to the store. This is useful when we're registering
207 /// a connection but we want to respond on the connection before anybody else can send on it.
208 fn add_request_handler<F, Fut, M>(&mut self, handler: F) -> &mut Self
209 where
210 F: 'static + Send + Sync + Fn(Arc<Self>, TypedEnvelope<M>, Response<M>) -> Fut,
211 Fut: Send + Future<Output = Result<()>>,
212 M: RequestMessage,
213 {
214 let handler = Arc::new(handler);
215 self.add_message_handler(move |server, envelope| {
216 let receipt = envelope.receipt();
217 let handler = handler.clone();
218 async move {
219 let responded = Arc::new(AtomicBool::default());
220 let response = Response {
221 server: server.clone(),
222 responded: responded.clone(),
223 receipt: envelope.receipt(),
224 };
225 match (handler)(server.clone(), envelope, response).await {
226 Ok(()) => {
227 if responded.load(std::sync::atomic::Ordering::SeqCst) {
228 Ok(())
229 } else {
230 Err(anyhow!("handler did not send a response"))?
231 }
232 }
233 Err(error) => {
234 server.peer.respond_with_error(
235 receipt,
236 proto::Error {
237 message: error.to_string(),
238 },
239 )?;
240 Err(error)
241 }
242 }
243 }
244 })
245 }
246
247 pub fn handle_connection<E: Executor>(
248 self: &Arc<Self>,
249 connection: Connection,
250 address: String,
251 user_id: UserId,
252 mut send_connection_id: Option<mpsc::Sender<ConnectionId>>,
253 executor: E,
254 ) -> impl Future<Output = Result<()>> {
255 let mut this = self.clone();
256 let span = info_span!("handle connection", %user_id, %address);
257 async move {
258 let (connection_id, handle_io, mut incoming_rx) = this
259 .peer
260 .add_connection(connection, {
261 let executor = executor.clone();
262 move |duration| {
263 let timer = executor.sleep(duration);
264 async move {
265 timer.await;
266 }
267 }
268 })
269 .await;
270
271 tracing::info!(%user_id, %connection_id, %address, "connection opened");
272
273 if let Some(send_connection_id) = send_connection_id.as_mut() {
274 let _ = send_connection_id.send(connection_id).await;
275 }
276
277 let contacts = this.app_state.db.get_contacts(user_id).await?;
278
279 {
280 let mut store = this.store_mut().await;
281 store.add_connection(connection_id, user_id);
282 this.peer.send(connection_id, store.build_initial_contacts_update(contacts))?;
283 }
284 this.update_user_contacts(user_id).await?;
285
286 let handle_io = handle_io.fuse();
287 futures::pin_mut!(handle_io);
288 loop {
289 let next_message = incoming_rx.next().fuse();
290 futures::pin_mut!(next_message);
291 futures::select_biased! {
292 result = handle_io => {
293 if let Err(error) = result {
294 tracing::error!(%error, "error handling I/O");
295 }
296 break;
297 }
298 message = next_message => {
299 if let Some(message) = message {
300 let type_name = message.payload_type_name();
301 let span = tracing::info_span!("receive message", %user_id, %connection_id, %address, type_name);
302 async {
303 if let Some(handler) = this.handlers.get(&message.payload_type_id()) {
304 let notifications = this.notifications.clone();
305 let is_background = message.is_background();
306 let handle_message = (handler)(this.clone(), message);
307 let handle_message = async move {
308 handle_message.await;
309 if let Some(mut notifications) = notifications {
310 let _ = notifications.send(()).await;
311 }
312 };
313 if is_background {
314 executor.spawn_detached(handle_message);
315 } else {
316 handle_message.await;
317 }
318 } else {
319 tracing::error!("no message handler");
320 }
321 }.instrument(span).await;
322 } else {
323 tracing::info!(%user_id, %connection_id, %address, "connection closed");
324 break;
325 }
326 }
327 }
328 }
329
330 if let Err(error) = this.sign_out(connection_id).await {
331 tracing::error!(%error, "error signing out");
332 }
333
334 Ok(())
335 }.instrument(span)
336 }
337
338 async fn sign_out(self: &mut Arc<Self>, connection_id: ConnectionId) -> Result<()> {
339 self.peer.disconnect(connection_id);
340
341 let removed_user_id = {
342 let mut store = self.store_mut().await;
343 let removed_connection = store.remove_connection(connection_id)?;
344
345 for (project_id, project) in removed_connection.hosted_projects {
346 broadcast(connection_id, project.guests.keys().copied(), |conn_id| {
347 self.peer
348 .send(conn_id, proto::UnregisterProject { project_id })
349 });
350 }
351
352 for project_id in removed_connection.guest_project_ids {
353 if let Some(project) = store.project(project_id).trace_err() {
354 broadcast(connection_id, project.connection_ids(), |conn_id| {
355 self.peer.send(
356 conn_id,
357 proto::RemoveProjectCollaborator {
358 project_id,
359 peer_id: connection_id.0,
360 },
361 )
362 });
363 if project.guests.is_empty() {
364 self.peer
365 .send(
366 project.host_connection_id,
367 proto::ProjectUnshared { project_id },
368 )
369 .trace_err();
370 }
371 }
372 }
373
374 removed_connection.user_id
375 };
376
377 self.update_user_contacts(removed_user_id).await?;
378
379 Ok(())
380 }
381
382 async fn ping(
383 self: Arc<Server>,
384 _: TypedEnvelope<proto::Ping>,
385 response: Response<proto::Ping>,
386 ) -> Result<()> {
387 response.send(proto::Ack {})?;
388 Ok(())
389 }
390
391 async fn register_project(
392 self: Arc<Server>,
393 request: TypedEnvelope<proto::RegisterProject>,
394 response: Response<proto::RegisterProject>,
395 ) -> Result<()> {
396 let user_id;
397 let project_id;
398 {
399 let mut state = self.store_mut().await;
400 user_id = state.user_id_for_connection(request.sender_id)?;
401 project_id = state.register_project(request.sender_id, user_id);
402 };
403 self.update_user_contacts(user_id).await?;
404 response.send(proto::RegisterProjectResponse { project_id })?;
405 Ok(())
406 }
407
408 async fn unregister_project(
409 self: Arc<Server>,
410 request: TypedEnvelope<proto::UnregisterProject>,
411 ) -> Result<()> {
412 let user_id = {
413 let mut state = self.store_mut().await;
414 state.unregister_project(request.payload.project_id, request.sender_id)?;
415 state.user_id_for_connection(request.sender_id)?
416 };
417
418 self.update_user_contacts(user_id).await?;
419 Ok(())
420 }
421
422 // async fn share_project(
423 // self: Arc<Server>,
424 // request: TypedEnvelope<proto::ShareProject>,
425 // response: Response<proto::ShareProject>,
426 // ) -> Result<()> {
427 // let user_id = {
428 // let mut state = self.store_mut().await;
429 // state.share_project(request.payload.project_id, request.sender_id)?;
430 // state.user_id_for_connection(request.sender_id)?
431 // };
432 // self.update_user_contacts(user_id).await?;
433 // response.send(proto::Ack {})?;
434 // Ok(())
435 // }
436
437 async fn update_user_contacts(self: &Arc<Server>, user_id: UserId) -> Result<()> {
438 let contacts = self.app_state.db.get_contacts(user_id).await?;
439 let store = self.store().await;
440 let updated_contact = store.contact_for_user(user_id, false);
441 for contact in contacts {
442 if let db::Contact::Accepted {
443 user_id: contact_user_id,
444 ..
445 } = contact
446 {
447 for contact_conn_id in store.connection_ids_for_user(contact_user_id) {
448 self.peer
449 .send(
450 contact_conn_id,
451 proto::UpdateContacts {
452 contacts: vec![updated_contact.clone()],
453 remove_contacts: Default::default(),
454 incoming_requests: Default::default(),
455 remove_incoming_requests: Default::default(),
456 outgoing_requests: Default::default(),
457 remove_outgoing_requests: Default::default(),
458 },
459 )
460 .trace_err();
461 }
462 }
463 }
464 Ok(())
465 }
466
467 async fn join_project(
468 self: Arc<Server>,
469 request: TypedEnvelope<proto::JoinProject>,
470 response: Response<proto::JoinProject>,
471 ) -> Result<()> {
472 let project_id = request.payload.project_id;
473 let host_user_id;
474 let guest_user_id;
475 let host_connection_id;
476 {
477 let state = self.store().await;
478 let project = state.project(project_id)?;
479 host_user_id = project.host_user_id;
480 host_connection_id = project.host_connection_id;
481 guest_user_id = state.user_id_for_connection(request.sender_id)?;
482 };
483
484 let has_contact = self
485 .app_state
486 .db
487 .has_contact(guest_user_id, host_user_id)
488 .await?;
489 if !has_contact {
490 return Err(anyhow!("no such project"))?;
491 }
492
493 self.store_mut().await.request_join_project(
494 guest_user_id,
495 project_id,
496 response.into_receipt(),
497 )?;
498 self.peer.send(
499 host_connection_id,
500 proto::RequestJoinProject {
501 project_id,
502 requester_id: guest_user_id.to_proto(),
503 },
504 )?;
505 Ok(())
506 }
507
508 async fn respond_to_join_project_request(
509 self: Arc<Server>,
510 request: TypedEnvelope<proto::RespondToJoinProjectRequest>,
511 ) -> Result<()> {
512 let host_user_id;
513
514 {
515 let mut state = self.store_mut().await;
516 let project_id = request.payload.project_id;
517 let project = state.project(project_id)?;
518 if project.host_connection_id != request.sender_id {
519 Err(anyhow!("no such connection"))?;
520 }
521
522 host_user_id = project.host_user_id;
523 let guest_user_id = UserId::from_proto(request.payload.requester_id);
524
525 if !request.payload.allow {
526 let receipts = state
527 .deny_join_project_request(request.sender_id, guest_user_id, project_id)
528 .ok_or_else(|| anyhow!("no such request"))?;
529 for receipt in receipts {
530 self.peer.respond(
531 receipt,
532 proto::JoinProjectResponse {
533 variant: Some(proto::join_project_response::Variant::Decline(
534 proto::join_project_response::Decline {},
535 )),
536 },
537 )?;
538 }
539 return Ok(());
540 }
541
542 let (receipts_with_replica_ids, project) = state
543 .accept_join_project_request(request.sender_id, guest_user_id, project_id)
544 .ok_or_else(|| anyhow!("no such request"))?;
545
546 let peer_count = project.guests.len();
547 let mut collaborators = Vec::with_capacity(peer_count);
548 collaborators.push(proto::Collaborator {
549 peer_id: project.host_connection_id.0,
550 replica_id: 0,
551 user_id: project.host_user_id.to_proto(),
552 });
553 let worktrees = project
554 .worktrees
555 .iter()
556 .filter_map(|(id, shared_worktree)| {
557 let worktree = project.worktrees.get(&id)?;
558 Some(proto::Worktree {
559 id: *id,
560 root_name: worktree.root_name.clone(),
561 entries: shared_worktree.entries.values().cloned().collect(),
562 diagnostic_summaries: shared_worktree
563 .diagnostic_summaries
564 .values()
565 .cloned()
566 .collect(),
567 visible: worktree.visible,
568 scan_id: shared_worktree.scan_id,
569 })
570 })
571 .collect::<Vec<_>>();
572
573 // Add all guests other than the requesting user's own connections as collaborators
574 for (peer_conn_id, (peer_replica_id, peer_user_id)) in &project.guests {
575 if receipts_with_replica_ids
576 .iter()
577 .all(|(receipt, _)| receipt.sender_id != *peer_conn_id)
578 {
579 collaborators.push(proto::Collaborator {
580 peer_id: peer_conn_id.0,
581 replica_id: *peer_replica_id as u32,
582 user_id: peer_user_id.to_proto(),
583 });
584 }
585 }
586
587 for conn_id in project.connection_ids() {
588 for (receipt, replica_id) in &receipts_with_replica_ids {
589 if conn_id != receipt.sender_id {
590 self.peer.send(
591 conn_id,
592 proto::AddProjectCollaborator {
593 project_id,
594 collaborator: Some(proto::Collaborator {
595 peer_id: receipt.sender_id.0,
596 replica_id: *replica_id as u32,
597 user_id: guest_user_id.to_proto(),
598 }),
599 },
600 )?;
601 }
602 }
603 }
604
605 for (receipt, replica_id) in receipts_with_replica_ids {
606 self.peer.respond(
607 receipt,
608 proto::JoinProjectResponse {
609 variant: Some(proto::join_project_response::Variant::Accept(
610 proto::join_project_response::Accept {
611 worktrees: worktrees.clone(),
612 replica_id: replica_id as u32,
613 collaborators: collaborators.clone(),
614 language_servers: project.language_servers.clone(),
615 },
616 )),
617 },
618 )?;
619 }
620 }
621
622 self.update_user_contacts(host_user_id).await?;
623 Ok(())
624 }
625
626 async fn leave_project(
627 self: Arc<Server>,
628 request: TypedEnvelope<proto::LeaveProject>,
629 ) -> Result<()> {
630 let sender_id = request.sender_id;
631 let project_id = request.payload.project_id;
632 let project;
633 {
634 let mut state = self.store_mut().await;
635 project = state.leave_project(sender_id, project_id)?;
636 let unshare = project.connection_ids.len() <= 1;
637 broadcast(sender_id, project.connection_ids, |conn_id| {
638 self.peer.send(
639 conn_id,
640 proto::RemoveProjectCollaborator {
641 project_id,
642 peer_id: sender_id.0,
643 },
644 )
645 });
646 if unshare {
647 self.peer.send(
648 project.host_connection_id,
649 proto::ProjectUnshared { project_id },
650 )?;
651 }
652 }
653 self.update_user_contacts(project.host_user_id).await?;
654 Ok(())
655 }
656
657 async fn register_worktree(
658 self: Arc<Server>,
659 request: TypedEnvelope<proto::RegisterWorktree>,
660 response: Response<proto::RegisterWorktree>,
661 ) -> Result<()> {
662 let host_user_id;
663 {
664 let mut state = self.store_mut().await;
665 host_user_id = state.user_id_for_connection(request.sender_id)?;
666
667 let guest_connection_ids = state
668 .read_project(request.payload.project_id, request.sender_id)?
669 .guest_connection_ids();
670 state.register_worktree(
671 request.payload.project_id,
672 request.payload.worktree_id,
673 request.sender_id,
674 Worktree {
675 root_name: request.payload.root_name.clone(),
676 visible: request.payload.visible,
677 ..Default::default()
678 },
679 )?;
680
681 broadcast(request.sender_id, guest_connection_ids, |connection_id| {
682 self.peer
683 .forward_send(request.sender_id, connection_id, request.payload.clone())
684 });
685 }
686 self.update_user_contacts(host_user_id).await?;
687 response.send(proto::Ack {})?;
688 Ok(())
689 }
690
691 async fn unregister_worktree(
692 self: Arc<Server>,
693 request: TypedEnvelope<proto::UnregisterWorktree>,
694 ) -> Result<()> {
695 let host_user_id;
696 let project_id = request.payload.project_id;
697 let worktree_id = request.payload.worktree_id;
698 {
699 let mut state = self.store_mut().await;
700 let (_, guest_connection_ids) =
701 state.unregister_worktree(project_id, worktree_id, request.sender_id)?;
702 host_user_id = state.user_id_for_connection(request.sender_id)?;
703 broadcast(request.sender_id, guest_connection_ids, |conn_id| {
704 self.peer.send(
705 conn_id,
706 proto::UnregisterWorktree {
707 project_id,
708 worktree_id,
709 },
710 )
711 });
712 }
713 self.update_user_contacts(host_user_id).await?;
714 Ok(())
715 }
716
717 async fn update_worktree(
718 self: Arc<Server>,
719 request: TypedEnvelope<proto::UpdateWorktree>,
720 response: Response<proto::UpdateWorktree>,
721 ) -> Result<()> {
722 let connection_ids = self.store_mut().await.update_worktree(
723 request.sender_id,
724 request.payload.project_id,
725 request.payload.worktree_id,
726 &request.payload.removed_entries,
727 &request.payload.updated_entries,
728 request.payload.scan_id,
729 )?;
730
731 broadcast(request.sender_id, connection_ids, |connection_id| {
732 self.peer
733 .forward_send(request.sender_id, connection_id, request.payload.clone())
734 });
735 response.send(proto::Ack {})?;
736 Ok(())
737 }
738
739 async fn update_diagnostic_summary(
740 self: Arc<Server>,
741 request: TypedEnvelope<proto::UpdateDiagnosticSummary>,
742 ) -> Result<()> {
743 let summary = request
744 .payload
745 .summary
746 .clone()
747 .ok_or_else(|| anyhow!("invalid summary"))?;
748 let receiver_ids = self.store_mut().await.update_diagnostic_summary(
749 request.payload.project_id,
750 request.payload.worktree_id,
751 request.sender_id,
752 summary,
753 )?;
754
755 broadcast(request.sender_id, receiver_ids, |connection_id| {
756 self.peer
757 .forward_send(request.sender_id, connection_id, request.payload.clone())
758 });
759 Ok(())
760 }
761
762 async fn start_language_server(
763 self: Arc<Server>,
764 request: TypedEnvelope<proto::StartLanguageServer>,
765 ) -> Result<()> {
766 let receiver_ids = self.store_mut().await.start_language_server(
767 request.payload.project_id,
768 request.sender_id,
769 request
770 .payload
771 .server
772 .clone()
773 .ok_or_else(|| anyhow!("invalid language server"))?,
774 )?;
775 broadcast(request.sender_id, receiver_ids, |connection_id| {
776 self.peer
777 .forward_send(request.sender_id, connection_id, request.payload.clone())
778 });
779 Ok(())
780 }
781
782 async fn update_language_server(
783 self: Arc<Server>,
784 request: TypedEnvelope<proto::UpdateLanguageServer>,
785 ) -> Result<()> {
786 let receiver_ids = self
787 .store()
788 .await
789 .project_connection_ids(request.payload.project_id, request.sender_id)?;
790 broadcast(request.sender_id, receiver_ids, |connection_id| {
791 self.peer
792 .forward_send(request.sender_id, connection_id, request.payload.clone())
793 });
794 Ok(())
795 }
796
797 async fn forward_project_request<T>(
798 self: Arc<Server>,
799 request: TypedEnvelope<T>,
800 response: Response<T>,
801 ) -> Result<()>
802 where
803 T: EntityMessage + RequestMessage,
804 {
805 let host_connection_id = self
806 .store()
807 .await
808 .read_project(request.payload.remote_entity_id(), request.sender_id)?
809 .host_connection_id;
810
811 response.send(
812 self.peer
813 .forward_request(request.sender_id, host_connection_id, request.payload)
814 .await?,
815 )?;
816 Ok(())
817 }
818
819 async fn save_buffer(
820 self: Arc<Server>,
821 request: TypedEnvelope<proto::SaveBuffer>,
822 response: Response<proto::SaveBuffer>,
823 ) -> Result<()> {
824 let host = self
825 .store()
826 .await
827 .read_project(request.payload.project_id, request.sender_id)?
828 .host_connection_id;
829 let response_payload = self
830 .peer
831 .forward_request(request.sender_id, host, request.payload.clone())
832 .await?;
833
834 let mut guests = self
835 .store()
836 .await
837 .read_project(request.payload.project_id, request.sender_id)?
838 .connection_ids();
839 guests.retain(|guest_connection_id| *guest_connection_id != request.sender_id);
840 broadcast(host, guests, |conn_id| {
841 self.peer
842 .forward_send(host, conn_id, response_payload.clone())
843 });
844 response.send(response_payload)?;
845 Ok(())
846 }
847
848 async fn update_buffer(
849 self: Arc<Server>,
850 request: TypedEnvelope<proto::UpdateBuffer>,
851 response: Response<proto::UpdateBuffer>,
852 ) -> Result<()> {
853 let receiver_ids = self
854 .store()
855 .await
856 .project_connection_ids(request.payload.project_id, request.sender_id)?;
857 broadcast(request.sender_id, receiver_ids, |connection_id| {
858 self.peer
859 .forward_send(request.sender_id, connection_id, request.payload.clone())
860 });
861 response.send(proto::Ack {})?;
862 Ok(())
863 }
864
865 async fn update_buffer_file(
866 self: Arc<Server>,
867 request: TypedEnvelope<proto::UpdateBufferFile>,
868 ) -> Result<()> {
869 let receiver_ids = self
870 .store()
871 .await
872 .project_connection_ids(request.payload.project_id, request.sender_id)?;
873 broadcast(request.sender_id, receiver_ids, |connection_id| {
874 self.peer
875 .forward_send(request.sender_id, connection_id, request.payload.clone())
876 });
877 Ok(())
878 }
879
880 async fn buffer_reloaded(
881 self: Arc<Server>,
882 request: TypedEnvelope<proto::BufferReloaded>,
883 ) -> Result<()> {
884 let receiver_ids = self
885 .store()
886 .await
887 .project_connection_ids(request.payload.project_id, request.sender_id)?;
888 broadcast(request.sender_id, receiver_ids, |connection_id| {
889 self.peer
890 .forward_send(request.sender_id, connection_id, request.payload.clone())
891 });
892 Ok(())
893 }
894
895 async fn buffer_saved(
896 self: Arc<Server>,
897 request: TypedEnvelope<proto::BufferSaved>,
898 ) -> Result<()> {
899 let receiver_ids = self
900 .store()
901 .await
902 .project_connection_ids(request.payload.project_id, request.sender_id)?;
903 broadcast(request.sender_id, receiver_ids, |connection_id| {
904 self.peer
905 .forward_send(request.sender_id, connection_id, request.payload.clone())
906 });
907 Ok(())
908 }
909
910 async fn follow(
911 self: Arc<Self>,
912 request: TypedEnvelope<proto::Follow>,
913 response: Response<proto::Follow>,
914 ) -> Result<()> {
915 let leader_id = ConnectionId(request.payload.leader_id);
916 let follower_id = request.sender_id;
917 if !self
918 .store()
919 .await
920 .project_connection_ids(request.payload.project_id, follower_id)?
921 .contains(&leader_id)
922 {
923 Err(anyhow!("no such peer"))?;
924 }
925 let mut response_payload = self
926 .peer
927 .forward_request(request.sender_id, leader_id, request.payload)
928 .await?;
929 response_payload
930 .views
931 .retain(|view| view.leader_id != Some(follower_id.0));
932 response.send(response_payload)?;
933 Ok(())
934 }
935
936 async fn unfollow(self: Arc<Self>, request: TypedEnvelope<proto::Unfollow>) -> Result<()> {
937 let leader_id = ConnectionId(request.payload.leader_id);
938 if !self
939 .store()
940 .await
941 .project_connection_ids(request.payload.project_id, request.sender_id)?
942 .contains(&leader_id)
943 {
944 Err(anyhow!("no such peer"))?;
945 }
946 self.peer
947 .forward_send(request.sender_id, leader_id, request.payload)?;
948 Ok(())
949 }
950
951 async fn update_followers(
952 self: Arc<Self>,
953 request: TypedEnvelope<proto::UpdateFollowers>,
954 ) -> Result<()> {
955 let connection_ids = self
956 .store()
957 .await
958 .project_connection_ids(request.payload.project_id, request.sender_id)?;
959 let leader_id = request
960 .payload
961 .variant
962 .as_ref()
963 .and_then(|variant| match variant {
964 proto::update_followers::Variant::CreateView(payload) => payload.leader_id,
965 proto::update_followers::Variant::UpdateView(payload) => payload.leader_id,
966 proto::update_followers::Variant::UpdateActiveView(payload) => payload.leader_id,
967 });
968 for follower_id in &request.payload.follower_ids {
969 let follower_id = ConnectionId(*follower_id);
970 if connection_ids.contains(&follower_id) && Some(follower_id.0) != leader_id {
971 self.peer
972 .forward_send(request.sender_id, follower_id, request.payload.clone())?;
973 }
974 }
975 Ok(())
976 }
977
978 async fn get_channels(
979 self: Arc<Server>,
980 request: TypedEnvelope<proto::GetChannels>,
981 response: Response<proto::GetChannels>,
982 ) -> Result<()> {
983 let user_id = self
984 .store()
985 .await
986 .user_id_for_connection(request.sender_id)?;
987 let channels = self.app_state.db.get_accessible_channels(user_id).await?;
988 response.send(proto::GetChannelsResponse {
989 channels: channels
990 .into_iter()
991 .map(|chan| proto::Channel {
992 id: chan.id.to_proto(),
993 name: chan.name,
994 })
995 .collect(),
996 })?;
997 Ok(())
998 }
999
1000 async fn get_users(
1001 self: Arc<Server>,
1002 request: TypedEnvelope<proto::GetUsers>,
1003 response: Response<proto::GetUsers>,
1004 ) -> Result<()> {
1005 let user_ids = request
1006 .payload
1007 .user_ids
1008 .into_iter()
1009 .map(UserId::from_proto)
1010 .collect();
1011 let users = self
1012 .app_state
1013 .db
1014 .get_users_by_ids(user_ids)
1015 .await?
1016 .into_iter()
1017 .map(|user| proto::User {
1018 id: user.id.to_proto(),
1019 avatar_url: format!("https://github.com/{}.png?size=128", user.github_login),
1020 github_login: user.github_login,
1021 })
1022 .collect();
1023 response.send(proto::UsersResponse { users })?;
1024 Ok(())
1025 }
1026
1027 async fn fuzzy_search_users(
1028 self: Arc<Server>,
1029 request: TypedEnvelope<proto::FuzzySearchUsers>,
1030 response: Response<proto::FuzzySearchUsers>,
1031 ) -> Result<()> {
1032 let user_id = self
1033 .store()
1034 .await
1035 .user_id_for_connection(request.sender_id)?;
1036 let query = request.payload.query;
1037 let db = &self.app_state.db;
1038 let users = match query.len() {
1039 0 => vec![],
1040 1 | 2 => db
1041 .get_user_by_github_login(&query)
1042 .await?
1043 .into_iter()
1044 .collect(),
1045 _ => db.fuzzy_search_users(&query, 10).await?,
1046 };
1047 let users = users
1048 .into_iter()
1049 .filter(|user| user.id != user_id)
1050 .map(|user| proto::User {
1051 id: user.id.to_proto(),
1052 avatar_url: format!("https://github.com/{}.png?size=128", user.github_login),
1053 github_login: user.github_login,
1054 })
1055 .collect();
1056 response.send(proto::UsersResponse { users })?;
1057 Ok(())
1058 }
1059
1060 async fn request_contact(
1061 self: Arc<Server>,
1062 request: TypedEnvelope<proto::RequestContact>,
1063 response: Response<proto::RequestContact>,
1064 ) -> Result<()> {
1065 let requester_id = self
1066 .store()
1067 .await
1068 .user_id_for_connection(request.sender_id)?;
1069 let responder_id = UserId::from_proto(request.payload.responder_id);
1070 if requester_id == responder_id {
1071 return Err(anyhow!("cannot add yourself as a contact"))?;
1072 }
1073
1074 self.app_state
1075 .db
1076 .send_contact_request(requester_id, responder_id)
1077 .await?;
1078
1079 // Update outgoing contact requests of requester
1080 let mut update = proto::UpdateContacts::default();
1081 update.outgoing_requests.push(responder_id.to_proto());
1082 for connection_id in self.store().await.connection_ids_for_user(requester_id) {
1083 self.peer.send(connection_id, update.clone())?;
1084 }
1085
1086 // Update incoming contact requests of responder
1087 let mut update = proto::UpdateContacts::default();
1088 update
1089 .incoming_requests
1090 .push(proto::IncomingContactRequest {
1091 requester_id: requester_id.to_proto(),
1092 should_notify: true,
1093 });
1094 for connection_id in self.store().await.connection_ids_for_user(responder_id) {
1095 self.peer.send(connection_id, update.clone())?;
1096 }
1097
1098 response.send(proto::Ack {})?;
1099 Ok(())
1100 }
1101
1102 async fn respond_to_contact_request(
1103 self: Arc<Server>,
1104 request: TypedEnvelope<proto::RespondToContactRequest>,
1105 response: Response<proto::RespondToContactRequest>,
1106 ) -> Result<()> {
1107 let responder_id = self
1108 .store()
1109 .await
1110 .user_id_for_connection(request.sender_id)?;
1111 let requester_id = UserId::from_proto(request.payload.requester_id);
1112 if request.payload.response == proto::ContactRequestResponse::Dismiss as i32 {
1113 self.app_state
1114 .db
1115 .dismiss_contact_notification(responder_id, requester_id)
1116 .await?;
1117 } else {
1118 let accept = request.payload.response == proto::ContactRequestResponse::Accept as i32;
1119 self.app_state
1120 .db
1121 .respond_to_contact_request(responder_id, requester_id, accept)
1122 .await?;
1123
1124 let store = self.store().await;
1125 // Update responder with new contact
1126 let mut update = proto::UpdateContacts::default();
1127 if accept {
1128 update
1129 .contacts
1130 .push(store.contact_for_user(requester_id, false));
1131 }
1132 update
1133 .remove_incoming_requests
1134 .push(requester_id.to_proto());
1135 for connection_id in store.connection_ids_for_user(responder_id) {
1136 self.peer.send(connection_id, update.clone())?;
1137 }
1138
1139 // Update requester with new contact
1140 let mut update = proto::UpdateContacts::default();
1141 if accept {
1142 update
1143 .contacts
1144 .push(store.contact_for_user(responder_id, true));
1145 }
1146 update
1147 .remove_outgoing_requests
1148 .push(responder_id.to_proto());
1149 for connection_id in store.connection_ids_for_user(requester_id) {
1150 self.peer.send(connection_id, update.clone())?;
1151 }
1152 }
1153
1154 response.send(proto::Ack {})?;
1155 Ok(())
1156 }
1157
1158 async fn remove_contact(
1159 self: Arc<Server>,
1160 request: TypedEnvelope<proto::RemoveContact>,
1161 response: Response<proto::RemoveContact>,
1162 ) -> Result<()> {
1163 let requester_id = self
1164 .store()
1165 .await
1166 .user_id_for_connection(request.sender_id)?;
1167 let responder_id = UserId::from_proto(request.payload.user_id);
1168 self.app_state
1169 .db
1170 .remove_contact(requester_id, responder_id)
1171 .await?;
1172
1173 // Update outgoing contact requests of requester
1174 let mut update = proto::UpdateContacts::default();
1175 update
1176 .remove_outgoing_requests
1177 .push(responder_id.to_proto());
1178 for connection_id in self.store().await.connection_ids_for_user(requester_id) {
1179 self.peer.send(connection_id, update.clone())?;
1180 }
1181
1182 // Update incoming contact requests of responder
1183 let mut update = proto::UpdateContacts::default();
1184 update
1185 .remove_incoming_requests
1186 .push(requester_id.to_proto());
1187 for connection_id in self.store().await.connection_ids_for_user(responder_id) {
1188 self.peer.send(connection_id, update.clone())?;
1189 }
1190
1191 response.send(proto::Ack {})?;
1192 Ok(())
1193 }
1194
1195 // #[instrument(skip(self, state, user_ids))]
1196 // fn update_contacts_for_users<'a>(
1197 // self: &Arc<Self>,
1198 // state: &Store,
1199 // user_ids: impl IntoIterator<Item = &'a UserId>,
1200 // ) {
1201 // for user_id in user_ids {
1202 // let contacts = state.contacts_for_user(*user_id);
1203 // for connection_id in state.connection_ids_for_user(*user_id) {
1204 // self.peer
1205 // .send(
1206 // connection_id,
1207 // proto::UpdateContacts {
1208 // contacts: contacts.clone(),
1209 // pending_requests_from_user_ids: Default::default(),
1210 // pending_requests_to_user_ids: Default::default(),
1211 // },
1212 // )
1213 // .trace_err();
1214 // }
1215 // }
1216 // }
1217
1218 async fn join_channel(
1219 self: Arc<Self>,
1220 request: TypedEnvelope<proto::JoinChannel>,
1221 response: Response<proto::JoinChannel>,
1222 ) -> Result<()> {
1223 let user_id = self
1224 .store()
1225 .await
1226 .user_id_for_connection(request.sender_id)?;
1227 let channel_id = ChannelId::from_proto(request.payload.channel_id);
1228 if !self
1229 .app_state
1230 .db
1231 .can_user_access_channel(user_id, channel_id)
1232 .await?
1233 {
1234 Err(anyhow!("access denied"))?;
1235 }
1236
1237 self.store_mut()
1238 .await
1239 .join_channel(request.sender_id, channel_id);
1240 let messages = self
1241 .app_state
1242 .db
1243 .get_channel_messages(channel_id, MESSAGE_COUNT_PER_PAGE, None)
1244 .await?
1245 .into_iter()
1246 .map(|msg| proto::ChannelMessage {
1247 id: msg.id.to_proto(),
1248 body: msg.body,
1249 timestamp: msg.sent_at.unix_timestamp() as u64,
1250 sender_id: msg.sender_id.to_proto(),
1251 nonce: Some(msg.nonce.as_u128().into()),
1252 })
1253 .collect::<Vec<_>>();
1254 response.send(proto::JoinChannelResponse {
1255 done: messages.len() < MESSAGE_COUNT_PER_PAGE,
1256 messages,
1257 })?;
1258 Ok(())
1259 }
1260
1261 async fn leave_channel(
1262 self: Arc<Self>,
1263 request: TypedEnvelope<proto::LeaveChannel>,
1264 ) -> Result<()> {
1265 let user_id = self
1266 .store()
1267 .await
1268 .user_id_for_connection(request.sender_id)?;
1269 let channel_id = ChannelId::from_proto(request.payload.channel_id);
1270 if !self
1271 .app_state
1272 .db
1273 .can_user_access_channel(user_id, channel_id)
1274 .await?
1275 {
1276 Err(anyhow!("access denied"))?;
1277 }
1278
1279 self.store_mut()
1280 .await
1281 .leave_channel(request.sender_id, channel_id);
1282
1283 Ok(())
1284 }
1285
1286 async fn send_channel_message(
1287 self: Arc<Self>,
1288 request: TypedEnvelope<proto::SendChannelMessage>,
1289 response: Response<proto::SendChannelMessage>,
1290 ) -> Result<()> {
1291 let channel_id = ChannelId::from_proto(request.payload.channel_id);
1292 let user_id;
1293 let connection_ids;
1294 {
1295 let state = self.store().await;
1296 user_id = state.user_id_for_connection(request.sender_id)?;
1297 connection_ids = state.channel_connection_ids(channel_id)?;
1298 }
1299
1300 // Validate the message body.
1301 let body = request.payload.body.trim().to_string();
1302 if body.len() > MAX_MESSAGE_LEN {
1303 return Err(anyhow!("message is too long"))?;
1304 }
1305 if body.is_empty() {
1306 return Err(anyhow!("message can't be blank"))?;
1307 }
1308
1309 let timestamp = OffsetDateTime::now_utc();
1310 let nonce = request
1311 .payload
1312 .nonce
1313 .ok_or_else(|| anyhow!("nonce can't be blank"))?;
1314
1315 let message_id = self
1316 .app_state
1317 .db
1318 .create_channel_message(channel_id, user_id, &body, timestamp, nonce.clone().into())
1319 .await?
1320 .to_proto();
1321 let message = proto::ChannelMessage {
1322 sender_id: user_id.to_proto(),
1323 id: message_id,
1324 body,
1325 timestamp: timestamp.unix_timestamp() as u64,
1326 nonce: Some(nonce),
1327 };
1328 broadcast(request.sender_id, connection_ids, |conn_id| {
1329 self.peer.send(
1330 conn_id,
1331 proto::ChannelMessageSent {
1332 channel_id: channel_id.to_proto(),
1333 message: Some(message.clone()),
1334 },
1335 )
1336 });
1337 response.send(proto::SendChannelMessageResponse {
1338 message: Some(message),
1339 })?;
1340 Ok(())
1341 }
1342
1343 async fn get_channel_messages(
1344 self: Arc<Self>,
1345 request: TypedEnvelope<proto::GetChannelMessages>,
1346 response: Response<proto::GetChannelMessages>,
1347 ) -> Result<()> {
1348 let user_id = self
1349 .store()
1350 .await
1351 .user_id_for_connection(request.sender_id)?;
1352 let channel_id = ChannelId::from_proto(request.payload.channel_id);
1353 if !self
1354 .app_state
1355 .db
1356 .can_user_access_channel(user_id, channel_id)
1357 .await?
1358 {
1359 Err(anyhow!("access denied"))?;
1360 }
1361
1362 let messages = self
1363 .app_state
1364 .db
1365 .get_channel_messages(
1366 channel_id,
1367 MESSAGE_COUNT_PER_PAGE,
1368 Some(MessageId::from_proto(request.payload.before_message_id)),
1369 )
1370 .await?
1371 .into_iter()
1372 .map(|msg| proto::ChannelMessage {
1373 id: msg.id.to_proto(),
1374 body: msg.body,
1375 timestamp: msg.sent_at.unix_timestamp() as u64,
1376 sender_id: msg.sender_id.to_proto(),
1377 nonce: Some(msg.nonce.as_u128().into()),
1378 })
1379 .collect::<Vec<_>>();
1380 response.send(proto::GetChannelMessagesResponse {
1381 done: messages.len() < MESSAGE_COUNT_PER_PAGE,
1382 messages,
1383 })?;
1384 Ok(())
1385 }
1386
1387 async fn store<'a>(self: &'a Arc<Self>) -> StoreReadGuard<'a> {
1388 #[cfg(test)]
1389 tokio::task::yield_now().await;
1390 let guard = self.store.read().await;
1391 #[cfg(test)]
1392 tokio::task::yield_now().await;
1393 StoreReadGuard {
1394 guard,
1395 _not_send: PhantomData,
1396 }
1397 }
1398
1399 async fn store_mut<'a>(self: &'a Arc<Self>) -> StoreWriteGuard<'a> {
1400 #[cfg(test)]
1401 tokio::task::yield_now().await;
1402 let guard = self.store.write().await;
1403 #[cfg(test)]
1404 tokio::task::yield_now().await;
1405 StoreWriteGuard {
1406 guard,
1407 _not_send: PhantomData,
1408 }
1409 }
1410}
1411
1412impl<'a> Deref for StoreReadGuard<'a> {
1413 type Target = Store;
1414
1415 fn deref(&self) -> &Self::Target {
1416 &*self.guard
1417 }
1418}
1419
1420impl<'a> Deref for StoreWriteGuard<'a> {
1421 type Target = Store;
1422
1423 fn deref(&self) -> &Self::Target {
1424 &*self.guard
1425 }
1426}
1427
1428impl<'a> DerefMut for StoreWriteGuard<'a> {
1429 fn deref_mut(&mut self) -> &mut Self::Target {
1430 &mut *self.guard
1431 }
1432}
1433
1434impl<'a> Drop for StoreWriteGuard<'a> {
1435 fn drop(&mut self) {
1436 #[cfg(test)]
1437 self.check_invariants();
1438
1439 let metrics = self.metrics();
1440 tracing::info!(
1441 connections = metrics.connections,
1442 registered_projects = metrics.registered_projects,
1443 shared_projects = metrics.shared_projects,
1444 "metrics"
1445 );
1446 }
1447}
1448
1449impl Executor for RealExecutor {
1450 type Sleep = Sleep;
1451
1452 fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
1453 tokio::task::spawn(future);
1454 }
1455
1456 fn sleep(&self, duration: Duration) -> Self::Sleep {
1457 tokio::time::sleep(duration)
1458 }
1459}
1460
1461fn broadcast<F>(
1462 sender_id: ConnectionId,
1463 receiver_ids: impl IntoIterator<Item = ConnectionId>,
1464 mut f: F,
1465) where
1466 F: FnMut(ConnectionId) -> anyhow::Result<()>,
1467{
1468 for receiver_id in receiver_ids {
1469 if receiver_id != sender_id {
1470 f(receiver_id).trace_err();
1471 }
1472 }
1473}
1474
1475lazy_static! {
1476 static ref ZED_PROTOCOL_VERSION: HeaderName = HeaderName::from_static("x-zed-protocol-version");
1477}
1478
1479pub struct ProtocolVersion(u32);
1480
1481impl Header for ProtocolVersion {
1482 fn name() -> &'static HeaderName {
1483 &ZED_PROTOCOL_VERSION
1484 }
1485
1486 fn decode<'i, I>(values: &mut I) -> Result<Self, axum::headers::Error>
1487 where
1488 Self: Sized,
1489 I: Iterator<Item = &'i axum::http::HeaderValue>,
1490 {
1491 let version = values
1492 .next()
1493 .ok_or_else(|| axum::headers::Error::invalid())?
1494 .to_str()
1495 .map_err(|_| axum::headers::Error::invalid())?
1496 .parse()
1497 .map_err(|_| axum::headers::Error::invalid())?;
1498 Ok(Self(version))
1499 }
1500
1501 fn encode<E: Extend<axum::http::HeaderValue>>(&self, values: &mut E) {
1502 values.extend([self.0.to_string().parse().unwrap()]);
1503 }
1504}
1505
1506pub fn routes(app_state: Arc<AppState>) -> Router<Body> {
1507 let server = Server::new(app_state.clone(), None);
1508 Router::new()
1509 .route("/rpc", get(handle_websocket_request))
1510 .layer(
1511 ServiceBuilder::new()
1512 .layer(Extension(app_state))
1513 .layer(middleware::from_fn(auth::validate_header))
1514 .layer(Extension(server)),
1515 )
1516}
1517
1518pub async fn handle_websocket_request(
1519 TypedHeader(ProtocolVersion(protocol_version)): TypedHeader<ProtocolVersion>,
1520 ConnectInfo(socket_address): ConnectInfo<SocketAddr>,
1521 Extension(server): Extension<Arc<Server>>,
1522 Extension(user_id): Extension<UserId>,
1523 ws: WebSocketUpgrade,
1524) -> axum::response::Response {
1525 if protocol_version != rpc::PROTOCOL_VERSION {
1526 return (
1527 StatusCode::UPGRADE_REQUIRED,
1528 "client must be upgraded".to_string(),
1529 )
1530 .into_response();
1531 }
1532 let socket_address = socket_address.to_string();
1533 ws.on_upgrade(move |socket| {
1534 use util::ResultExt;
1535 let socket = socket
1536 .map_ok(to_tungstenite_message)
1537 .err_into()
1538 .with(|message| async move { Ok(to_axum_message(message)) });
1539 let connection = Connection::new(Box::pin(socket));
1540 async move {
1541 server
1542 .handle_connection(connection, socket_address, user_id, None, RealExecutor)
1543 .await
1544 .log_err();
1545 }
1546 })
1547}
1548
1549fn to_axum_message(message: TungsteniteMessage) -> AxumMessage {
1550 match message {
1551 TungsteniteMessage::Text(payload) => AxumMessage::Text(payload),
1552 TungsteniteMessage::Binary(payload) => AxumMessage::Binary(payload),
1553 TungsteniteMessage::Ping(payload) => AxumMessage::Ping(payload),
1554 TungsteniteMessage::Pong(payload) => AxumMessage::Pong(payload),
1555 TungsteniteMessage::Close(frame) => AxumMessage::Close(frame.map(|frame| AxumCloseFrame {
1556 code: frame.code.into(),
1557 reason: frame.reason,
1558 })),
1559 }
1560}
1561
1562fn to_tungstenite_message(message: AxumMessage) -> TungsteniteMessage {
1563 match message {
1564 AxumMessage::Text(payload) => TungsteniteMessage::Text(payload),
1565 AxumMessage::Binary(payload) => TungsteniteMessage::Binary(payload),
1566 AxumMessage::Ping(payload) => TungsteniteMessage::Ping(payload),
1567 AxumMessage::Pong(payload) => TungsteniteMessage::Pong(payload),
1568 AxumMessage::Close(frame) => {
1569 TungsteniteMessage::Close(frame.map(|frame| TungsteniteCloseFrame {
1570 code: frame.code.into(),
1571 reason: frame.reason,
1572 }))
1573 }
1574 }
1575}
1576
1577pub trait ResultExt {
1578 type Ok;
1579
1580 fn trace_err(self) -> Option<Self::Ok>;
1581}
1582
1583impl<T, E> ResultExt for Result<T, E>
1584where
1585 E: std::fmt::Debug,
1586{
1587 type Ok = T;
1588
1589 fn trace_err(self) -> Option<T> {
1590 match self {
1591 Ok(value) => Some(value),
1592 Err(error) => {
1593 tracing::error!("{:?}", error);
1594 None
1595 }
1596 }
1597 }
1598}
1599
1600#[cfg(test)]
1601mod tests {
1602 use super::*;
1603 use crate::{
1604 db::{tests::TestDb, UserId},
1605 AppState,
1606 };
1607 use ::rpc::Peer;
1608 use client::{
1609 self, test::FakeHttpClient, Channel, ChannelDetails, ChannelList, Client, Credentials,
1610 EstablishConnectionError, UserStore, RECEIVE_TIMEOUT,
1611 };
1612 use collections::{BTreeMap, HashSet};
1613 use editor::{
1614 self, ConfirmCodeAction, ConfirmCompletion, ConfirmRename, Editor, Input, Redo, Rename,
1615 ToOffset, ToggleCodeActions, Undo,
1616 };
1617 use gpui::{
1618 executor::{self, Deterministic},
1619 geometry::vector::vec2f,
1620 ModelHandle, TestAppContext, ViewHandle,
1621 };
1622 use language::{
1623 range_to_lsp, tree_sitter_rust, Diagnostic, DiagnosticEntry, FakeLspAdapter, Language,
1624 LanguageConfig, LanguageRegistry, OffsetRangeExt, Point, Rope,
1625 };
1626 use lsp::{self, FakeLanguageServer};
1627 use parking_lot::Mutex;
1628 use project::{
1629 fs::{FakeFs, Fs as _},
1630 search::SearchQuery,
1631 worktree::WorktreeHandle,
1632 DiagnosticSummary, Project, ProjectPath, WorktreeId,
1633 };
1634 use rand::prelude::*;
1635 use rpc::PeerId;
1636 use serde_json::json;
1637 use settings::Settings;
1638 use sqlx::types::time::OffsetDateTime;
1639 use std::{
1640 env,
1641 ops::Deref,
1642 path::{Path, PathBuf},
1643 rc::Rc,
1644 sync::{
1645 atomic::{AtomicBool, Ordering::SeqCst},
1646 Arc,
1647 },
1648 time::Duration,
1649 };
1650 use theme::ThemeRegistry;
1651 use workspace::{Item, SplitDirection, ToggleFollow, Workspace, WorkspaceParams};
1652
1653 #[cfg(test)]
1654 #[ctor::ctor]
1655 fn init_logger() {
1656 if std::env::var("RUST_LOG").is_ok() {
1657 env_logger::init();
1658 }
1659 }
1660
1661 #[gpui::test(iterations = 10)]
1662 async fn test_share_project(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
1663 let (window_b, _) = cx_b.add_window(|_| EmptyView);
1664 let lang_registry = Arc::new(LanguageRegistry::test());
1665 let fs = FakeFs::new(cx_a.background());
1666 cx_a.foreground().forbid_parking();
1667
1668 // Connect to a server as 2 clients.
1669 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1670 let client_a = server.create_client(cx_a, "user_a").await;
1671 let mut client_b = server.create_client(cx_b, "user_b").await;
1672 server
1673 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
1674 .await;
1675
1676 // Share a project as client A
1677 fs.insert_tree(
1678 "/a",
1679 json!({
1680 "a.txt": "a-contents",
1681 "b.txt": "b-contents",
1682 }),
1683 )
1684 .await;
1685 let project_a = cx_a.update(|cx| {
1686 Project::local(
1687 client_a.clone(),
1688 client_a.user_store.clone(),
1689 lang_registry.clone(),
1690 fs.clone(),
1691 cx,
1692 )
1693 });
1694 let (worktree_a, _) = project_a
1695 .update(cx_a, |p, cx| {
1696 p.find_or_create_local_worktree("/a", true, cx)
1697 })
1698 .await
1699 .unwrap();
1700 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1701 worktree_a
1702 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1703 .await;
1704
1705 // Join that project as client B
1706 let client_b_peer_id = client_b.peer_id;
1707 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1708
1709 let replica_id_b = project_b.read_with(cx_b, |project, _| {
1710 assert_eq!(
1711 project
1712 .collaborators()
1713 .get(&client_a.peer_id)
1714 .unwrap()
1715 .user
1716 .github_login,
1717 "user_a"
1718 );
1719 project.replica_id()
1720 });
1721 project_a
1722 .condition(&cx_a, |tree, _| {
1723 tree.collaborators()
1724 .get(&client_b_peer_id)
1725 .map_or(false, |collaborator| {
1726 collaborator.replica_id == replica_id_b
1727 && collaborator.user.github_login == "user_b"
1728 })
1729 })
1730 .await;
1731
1732 // Open the same file as client B and client A.
1733 let buffer_b = project_b
1734 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.txt"), cx))
1735 .await
1736 .unwrap();
1737 buffer_b.read_with(cx_b, |buf, _| assert_eq!(buf.text(), "b-contents"));
1738 project_a.read_with(cx_a, |project, cx| {
1739 assert!(project.has_open_buffer((worktree_id, "b.txt"), cx))
1740 });
1741 let buffer_a = project_a
1742 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "b.txt"), cx))
1743 .await
1744 .unwrap();
1745
1746 let editor_b = cx_b.add_view(window_b, |cx| Editor::for_buffer(buffer_b, None, cx));
1747
1748 // TODO
1749 // // Create a selection set as client B and see that selection set as client A.
1750 // buffer_a
1751 // .condition(&cx_a, |buffer, _| buffer.selection_sets().count() == 1)
1752 // .await;
1753
1754 // Edit the buffer as client B and see that edit as client A.
1755 editor_b.update(cx_b, |editor, cx| {
1756 editor.handle_input(&Input("ok, ".into()), cx)
1757 });
1758 buffer_a
1759 .condition(&cx_a, |buffer, _| buffer.text() == "ok, b-contents")
1760 .await;
1761
1762 // TODO
1763 // // Remove the selection set as client B, see those selections disappear as client A.
1764 cx_b.update(move |_| drop(editor_b));
1765 // buffer_a
1766 // .condition(&cx_a, |buffer, _| buffer.selection_sets().count() == 0)
1767 // .await;
1768
1769 // Dropping the client B's project removes client B from client A's collaborators.
1770 cx_b.update(move |_| {
1771 drop(client_b.project.take());
1772 drop(project_b);
1773 });
1774 project_a
1775 .condition(&cx_a, |project, _| project.collaborators().is_empty())
1776 .await;
1777 }
1778
1779 #[gpui::test(iterations = 10)]
1780 async fn test_unshare_project(
1781 deterministic: Arc<Deterministic>,
1782 cx_a: &mut TestAppContext,
1783 cx_b: &mut TestAppContext,
1784 ) {
1785 let lang_registry = Arc::new(LanguageRegistry::test());
1786 let fs = FakeFs::new(cx_a.background());
1787 cx_a.foreground().forbid_parking();
1788
1789 // Connect to a server as 2 clients.
1790 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1791 let client_a = server.create_client(cx_a, "user_a").await;
1792 let mut client_b = server.create_client(cx_b, "user_b").await;
1793 server
1794 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
1795 .await;
1796
1797 // Share a project as client A
1798 fs.insert_tree(
1799 "/a",
1800 json!({
1801 "a.txt": "a-contents",
1802 "b.txt": "b-contents",
1803 }),
1804 )
1805 .await;
1806 let project_a = cx_a.update(|cx| {
1807 Project::local(
1808 client_a.clone(),
1809 client_a.user_store.clone(),
1810 lang_registry.clone(),
1811 fs.clone(),
1812 cx,
1813 )
1814 });
1815 let (worktree_a, _) = project_a
1816 .update(cx_a, |p, cx| {
1817 p.find_or_create_local_worktree("/a", true, cx)
1818 })
1819 .await
1820 .unwrap();
1821 worktree_a
1822 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1823 .await;
1824 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1825
1826 // Join that project as client B
1827 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1828 assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1829 project_b
1830 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1831 .await
1832 .unwrap();
1833
1834 // When client B leaves the project, it gets automatically unshared.
1835 cx_b.update(|_| {
1836 drop(client_b.project.take());
1837 drop(project_b);
1838 });
1839 deterministic.run_until_parked();
1840 assert!(worktree_a.read_with(cx_a, |tree, _| !tree.as_local().unwrap().is_shared()));
1841
1842 // When client B joins again, the project gets re-shared.
1843 let project_b2 = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1844 assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1845 project_b2
1846 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1847 .await
1848 .unwrap();
1849 }
1850
1851 #[gpui::test(iterations = 10)]
1852 async fn test_host_disconnect(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
1853 let lang_registry = Arc::new(LanguageRegistry::test());
1854 let fs = FakeFs::new(cx_a.background());
1855 cx_a.foreground().forbid_parking();
1856
1857 // Connect to a server as 2 clients.
1858 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1859 let client_a = server.create_client(cx_a, "user_a").await;
1860 let mut client_b = server.create_client(cx_b, "user_b").await;
1861 server
1862 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
1863 .await;
1864
1865 // Share a project as client A
1866 fs.insert_tree(
1867 "/a",
1868 json!({
1869 "a.txt": "a-contents",
1870 "b.txt": "b-contents",
1871 }),
1872 )
1873 .await;
1874 let project_a = cx_a.update(|cx| {
1875 Project::local(
1876 client_a.clone(),
1877 client_a.user_store.clone(),
1878 lang_registry.clone(),
1879 fs.clone(),
1880 cx,
1881 )
1882 });
1883 let (worktree_a, _) = project_a
1884 .update(cx_a, |p, cx| {
1885 p.find_or_create_local_worktree("/a", true, cx)
1886 })
1887 .await
1888 .unwrap();
1889 worktree_a
1890 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1891 .await;
1892 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1893
1894 // Join that project as client B
1895 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1896 assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1897 project_b
1898 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1899 .await
1900 .unwrap();
1901
1902 // Drop client A's connection. Collaborators should disappear and the project should not be shown as shared.
1903 server.disconnect_client(client_a.current_user_id(cx_a));
1904 cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
1905 project_a
1906 .condition(cx_a, |project, _| project.collaborators().is_empty())
1907 .await;
1908 project_a.read_with(cx_a, |project, _| assert!(!project.is_shared()));
1909 project_b
1910 .condition(cx_b, |project, _| project.is_read_only())
1911 .await;
1912 assert!(worktree_a.read_with(cx_a, |tree, _| !tree.as_local().unwrap().is_shared()));
1913 cx_b.update(|_| {
1914 drop(project_b);
1915 });
1916
1917 // Ensure guests can still join.
1918 let project_b2 = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1919 assert!(worktree_a.read_with(cx_a, |tree, _| tree.as_local().unwrap().is_shared()));
1920 project_b2
1921 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
1922 .await
1923 .unwrap();
1924 }
1925
1926 #[gpui::test(iterations = 10)]
1927 async fn test_propagate_saves_and_fs_changes(
1928 cx_a: &mut TestAppContext,
1929 cx_b: &mut TestAppContext,
1930 cx_c: &mut TestAppContext,
1931 ) {
1932 let lang_registry = Arc::new(LanguageRegistry::test());
1933 let fs = FakeFs::new(cx_a.background());
1934 cx_a.foreground().forbid_parking();
1935
1936 // Connect to a server as 3 clients.
1937 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
1938 let client_a = server.create_client(cx_a, "user_a").await;
1939 let mut client_b = server.create_client(cx_b, "user_b").await;
1940 let mut client_c = server.create_client(cx_c, "user_c").await;
1941 server
1942 .make_contacts(vec![
1943 (&client_a, cx_a),
1944 (&client_b, cx_b),
1945 (&client_c, cx_c),
1946 ])
1947 .await;
1948
1949 // Share a worktree as client A.
1950 fs.insert_tree(
1951 "/a",
1952 json!({
1953 "file1": "",
1954 "file2": ""
1955 }),
1956 )
1957 .await;
1958 let project_a = cx_a.update(|cx| {
1959 Project::local(
1960 client_a.clone(),
1961 client_a.user_store.clone(),
1962 lang_registry.clone(),
1963 fs.clone(),
1964 cx,
1965 )
1966 });
1967 let (worktree_a, _) = project_a
1968 .update(cx_a, |p, cx| {
1969 p.find_or_create_local_worktree("/a", true, cx)
1970 })
1971 .await
1972 .unwrap();
1973 worktree_a
1974 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
1975 .await;
1976 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
1977
1978 // Join that worktree as clients B and C.
1979 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
1980 let project_c = client_c.build_remote_project(&project_a, cx_a, cx_c).await;
1981 let worktree_b = project_b.read_with(cx_b, |p, cx| p.worktrees(cx).next().unwrap());
1982 let worktree_c = project_c.read_with(cx_c, |p, cx| p.worktrees(cx).next().unwrap());
1983
1984 // Open and edit a buffer as both guests B and C.
1985 let buffer_b = project_b
1986 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1987 .await
1988 .unwrap();
1989 let buffer_c = project_c
1990 .update(cx_c, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1991 .await
1992 .unwrap();
1993 buffer_b.update(cx_b, |buf, cx| buf.edit([(0..0, "i-am-b, ")], cx));
1994 buffer_c.update(cx_c, |buf, cx| buf.edit([(0..0, "i-am-c, ")], cx));
1995
1996 // Open and edit that buffer as the host.
1997 let buffer_a = project_a
1998 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "file1"), cx))
1999 .await
2000 .unwrap();
2001
2002 buffer_a
2003 .condition(cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, ")
2004 .await;
2005 buffer_a.update(cx_a, |buf, cx| {
2006 buf.edit([(buf.len()..buf.len(), "i-am-a")], cx)
2007 });
2008
2009 // Wait for edits to propagate
2010 buffer_a
2011 .condition(cx_a, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
2012 .await;
2013 buffer_b
2014 .condition(cx_b, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
2015 .await;
2016 buffer_c
2017 .condition(cx_c, |buf, _| buf.text() == "i-am-c, i-am-b, i-am-a")
2018 .await;
2019
2020 // Edit the buffer as the host and concurrently save as guest B.
2021 let save_b = buffer_b.update(cx_b, |buf, cx| buf.save(cx));
2022 buffer_a.update(cx_a, |buf, cx| buf.edit([(0..0, "hi-a, ")], cx));
2023 save_b.await.unwrap();
2024 assert_eq!(
2025 fs.load("/a/file1".as_ref()).await.unwrap(),
2026 "hi-a, i-am-c, i-am-b, i-am-a"
2027 );
2028 buffer_a.read_with(cx_a, |buf, _| assert!(!buf.is_dirty()));
2029 buffer_b.read_with(cx_b, |buf, _| assert!(!buf.is_dirty()));
2030 buffer_c.condition(cx_c, |buf, _| !buf.is_dirty()).await;
2031
2032 worktree_a.flush_fs_events(cx_a).await;
2033
2034 // Make changes on host's file system, see those changes on guest worktrees.
2035 fs.rename(
2036 "/a/file1".as_ref(),
2037 "/a/file1-renamed".as_ref(),
2038 Default::default(),
2039 )
2040 .await
2041 .unwrap();
2042
2043 fs.rename("/a/file2".as_ref(), "/a/file3".as_ref(), Default::default())
2044 .await
2045 .unwrap();
2046 fs.insert_file(Path::new("/a/file4"), "4".into()).await;
2047
2048 worktree_a
2049 .condition(&cx_a, |tree, _| {
2050 tree.paths()
2051 .map(|p| p.to_string_lossy())
2052 .collect::<Vec<_>>()
2053 == ["file1-renamed", "file3", "file4"]
2054 })
2055 .await;
2056 worktree_b
2057 .condition(&cx_b, |tree, _| {
2058 tree.paths()
2059 .map(|p| p.to_string_lossy())
2060 .collect::<Vec<_>>()
2061 == ["file1-renamed", "file3", "file4"]
2062 })
2063 .await;
2064 worktree_c
2065 .condition(&cx_c, |tree, _| {
2066 tree.paths()
2067 .map(|p| p.to_string_lossy())
2068 .collect::<Vec<_>>()
2069 == ["file1-renamed", "file3", "file4"]
2070 })
2071 .await;
2072
2073 // Ensure buffer files are updated as well.
2074 buffer_a
2075 .condition(&cx_a, |buf, _| {
2076 buf.file().unwrap().path().to_str() == Some("file1-renamed")
2077 })
2078 .await;
2079 buffer_b
2080 .condition(&cx_b, |buf, _| {
2081 buf.file().unwrap().path().to_str() == Some("file1-renamed")
2082 })
2083 .await;
2084 buffer_c
2085 .condition(&cx_c, |buf, _| {
2086 buf.file().unwrap().path().to_str() == Some("file1-renamed")
2087 })
2088 .await;
2089 }
2090
2091 #[gpui::test(iterations = 10)]
2092 async fn test_fs_operations(
2093 executor: Arc<Deterministic>,
2094 cx_a: &mut TestAppContext,
2095 cx_b: &mut TestAppContext,
2096 ) {
2097 executor.forbid_parking();
2098 let fs = FakeFs::new(cx_a.background());
2099
2100 // Connect to a server as 2 clients.
2101 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2102 let mut client_a = server.create_client(cx_a, "user_a").await;
2103 let mut client_b = server.create_client(cx_b, "user_b").await;
2104 server
2105 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2106 .await;
2107
2108 // Share a project as client A
2109 fs.insert_tree(
2110 "/dir",
2111 json!({
2112 "a.txt": "a-contents",
2113 "b.txt": "b-contents",
2114 }),
2115 )
2116 .await;
2117
2118 let (project_a, worktree_id) = client_a.build_local_project(fs, "/dir", cx_a).await;
2119 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2120
2121 let worktree_a =
2122 project_a.read_with(cx_a, |project, cx| project.worktrees(cx).next().unwrap());
2123 let worktree_b =
2124 project_b.read_with(cx_b, |project, cx| project.worktrees(cx).next().unwrap());
2125
2126 let entry = project_b
2127 .update(cx_b, |project, cx| {
2128 project
2129 .create_entry((worktree_id, "c.txt"), false, cx)
2130 .unwrap()
2131 })
2132 .await
2133 .unwrap();
2134 worktree_a.read_with(cx_a, |worktree, _| {
2135 assert_eq!(
2136 worktree
2137 .paths()
2138 .map(|p| p.to_string_lossy())
2139 .collect::<Vec<_>>(),
2140 ["a.txt", "b.txt", "c.txt"]
2141 );
2142 });
2143 worktree_b.read_with(cx_b, |worktree, _| {
2144 assert_eq!(
2145 worktree
2146 .paths()
2147 .map(|p| p.to_string_lossy())
2148 .collect::<Vec<_>>(),
2149 ["a.txt", "b.txt", "c.txt"]
2150 );
2151 });
2152
2153 project_b
2154 .update(cx_b, |project, cx| {
2155 project.rename_entry(entry.id, Path::new("d.txt"), cx)
2156 })
2157 .unwrap()
2158 .await
2159 .unwrap();
2160 worktree_a.read_with(cx_a, |worktree, _| {
2161 assert_eq!(
2162 worktree
2163 .paths()
2164 .map(|p| p.to_string_lossy())
2165 .collect::<Vec<_>>(),
2166 ["a.txt", "b.txt", "d.txt"]
2167 );
2168 });
2169 worktree_b.read_with(cx_b, |worktree, _| {
2170 assert_eq!(
2171 worktree
2172 .paths()
2173 .map(|p| p.to_string_lossy())
2174 .collect::<Vec<_>>(),
2175 ["a.txt", "b.txt", "d.txt"]
2176 );
2177 });
2178
2179 let dir_entry = project_b
2180 .update(cx_b, |project, cx| {
2181 project
2182 .create_entry((worktree_id, "DIR"), true, cx)
2183 .unwrap()
2184 })
2185 .await
2186 .unwrap();
2187 worktree_a.read_with(cx_a, |worktree, _| {
2188 assert_eq!(
2189 worktree
2190 .paths()
2191 .map(|p| p.to_string_lossy())
2192 .collect::<Vec<_>>(),
2193 ["DIR", "a.txt", "b.txt", "d.txt"]
2194 );
2195 });
2196 worktree_b.read_with(cx_b, |worktree, _| {
2197 assert_eq!(
2198 worktree
2199 .paths()
2200 .map(|p| p.to_string_lossy())
2201 .collect::<Vec<_>>(),
2202 ["DIR", "a.txt", "b.txt", "d.txt"]
2203 );
2204 });
2205
2206 project_b
2207 .update(cx_b, |project, cx| {
2208 project.delete_entry(dir_entry.id, cx).unwrap()
2209 })
2210 .await
2211 .unwrap();
2212 worktree_a.read_with(cx_a, |worktree, _| {
2213 assert_eq!(
2214 worktree
2215 .paths()
2216 .map(|p| p.to_string_lossy())
2217 .collect::<Vec<_>>(),
2218 ["a.txt", "b.txt", "d.txt"]
2219 );
2220 });
2221 worktree_b.read_with(cx_b, |worktree, _| {
2222 assert_eq!(
2223 worktree
2224 .paths()
2225 .map(|p| p.to_string_lossy())
2226 .collect::<Vec<_>>(),
2227 ["a.txt", "b.txt", "d.txt"]
2228 );
2229 });
2230
2231 project_b
2232 .update(cx_b, |project, cx| {
2233 project.delete_entry(entry.id, cx).unwrap()
2234 })
2235 .await
2236 .unwrap();
2237 worktree_a.read_with(cx_a, |worktree, _| {
2238 assert_eq!(
2239 worktree
2240 .paths()
2241 .map(|p| p.to_string_lossy())
2242 .collect::<Vec<_>>(),
2243 ["a.txt", "b.txt"]
2244 );
2245 });
2246 worktree_b.read_with(cx_b, |worktree, _| {
2247 assert_eq!(
2248 worktree
2249 .paths()
2250 .map(|p| p.to_string_lossy())
2251 .collect::<Vec<_>>(),
2252 ["a.txt", "b.txt"]
2253 );
2254 });
2255 }
2256
2257 #[gpui::test(iterations = 10)]
2258 async fn test_buffer_conflict_after_save(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2259 cx_a.foreground().forbid_parking();
2260 let lang_registry = Arc::new(LanguageRegistry::test());
2261 let fs = FakeFs::new(cx_a.background());
2262
2263 // Connect to a server as 2 clients.
2264 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2265 let client_a = server.create_client(cx_a, "user_a").await;
2266 let mut client_b = server.create_client(cx_b, "user_b").await;
2267 server
2268 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2269 .await;
2270
2271 // Share a project as client A
2272 fs.insert_tree(
2273 "/dir",
2274 json!({
2275 "a.txt": "a-contents",
2276 }),
2277 )
2278 .await;
2279
2280 let project_a = cx_a.update(|cx| {
2281 Project::local(
2282 client_a.clone(),
2283 client_a.user_store.clone(),
2284 lang_registry.clone(),
2285 fs.clone(),
2286 cx,
2287 )
2288 });
2289 let (worktree_a, _) = project_a
2290 .update(cx_a, |p, cx| {
2291 p.find_or_create_local_worktree("/dir", true, cx)
2292 })
2293 .await
2294 .unwrap();
2295 worktree_a
2296 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2297 .await;
2298 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2299
2300 // Join that project as client B
2301 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2302
2303 // Open a buffer as client B
2304 let buffer_b = project_b
2305 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
2306 .await
2307 .unwrap();
2308
2309 buffer_b.update(cx_b, |buf, cx| buf.edit([(0..0, "world ")], cx));
2310 buffer_b.read_with(cx_b, |buf, _| {
2311 assert!(buf.is_dirty());
2312 assert!(!buf.has_conflict());
2313 });
2314
2315 buffer_b.update(cx_b, |buf, cx| buf.save(cx)).await.unwrap();
2316 buffer_b
2317 .condition(&cx_b, |buffer_b, _| !buffer_b.is_dirty())
2318 .await;
2319 buffer_b.read_with(cx_b, |buf, _| {
2320 assert!(!buf.has_conflict());
2321 });
2322
2323 buffer_b.update(cx_b, |buf, cx| buf.edit([(0..0, "hello ")], cx));
2324 buffer_b.read_with(cx_b, |buf, _| {
2325 assert!(buf.is_dirty());
2326 assert!(!buf.has_conflict());
2327 });
2328 }
2329
2330 #[gpui::test(iterations = 10)]
2331 async fn test_buffer_reloading(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2332 cx_a.foreground().forbid_parking();
2333 let lang_registry = Arc::new(LanguageRegistry::test());
2334 let fs = FakeFs::new(cx_a.background());
2335
2336 // Connect to a server as 2 clients.
2337 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2338 let client_a = server.create_client(cx_a, "user_a").await;
2339 let mut client_b = server.create_client(cx_b, "user_b").await;
2340 server
2341 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2342 .await;
2343
2344 // Share a project as client A
2345 fs.insert_tree(
2346 "/dir",
2347 json!({
2348 "a.txt": "a-contents",
2349 }),
2350 )
2351 .await;
2352
2353 let project_a = cx_a.update(|cx| {
2354 Project::local(
2355 client_a.clone(),
2356 client_a.user_store.clone(),
2357 lang_registry.clone(),
2358 fs.clone(),
2359 cx,
2360 )
2361 });
2362 let (worktree_a, _) = project_a
2363 .update(cx_a, |p, cx| {
2364 p.find_or_create_local_worktree("/dir", true, cx)
2365 })
2366 .await
2367 .unwrap();
2368 worktree_a
2369 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2370 .await;
2371 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2372
2373 // Join that project as client B
2374 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2375 let _worktree_b = project_b.update(cx_b, |p, cx| p.worktrees(cx).next().unwrap());
2376
2377 // Open a buffer as client B
2378 let buffer_b = project_b
2379 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
2380 .await
2381 .unwrap();
2382 buffer_b.read_with(cx_b, |buf, _| {
2383 assert!(!buf.is_dirty());
2384 assert!(!buf.has_conflict());
2385 });
2386
2387 fs.save(Path::new("/dir/a.txt"), &"new contents".into())
2388 .await
2389 .unwrap();
2390 buffer_b
2391 .condition(&cx_b, |buf, _| {
2392 buf.text() == "new contents" && !buf.is_dirty()
2393 })
2394 .await;
2395 buffer_b.read_with(cx_b, |buf, _| {
2396 assert!(!buf.has_conflict());
2397 });
2398 }
2399
2400 #[gpui::test(iterations = 10)]
2401 async fn test_editing_while_guest_opens_buffer(
2402 cx_a: &mut TestAppContext,
2403 cx_b: &mut TestAppContext,
2404 ) {
2405 cx_a.foreground().forbid_parking();
2406 let lang_registry = Arc::new(LanguageRegistry::test());
2407 let fs = FakeFs::new(cx_a.background());
2408
2409 // Connect to a server as 2 clients.
2410 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2411 let client_a = server.create_client(cx_a, "user_a").await;
2412 let mut client_b = server.create_client(cx_b, "user_b").await;
2413 server
2414 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2415 .await;
2416
2417 // Share a project as client A
2418 fs.insert_tree(
2419 "/dir",
2420 json!({
2421 "a.txt": "a-contents",
2422 }),
2423 )
2424 .await;
2425 let project_a = cx_a.update(|cx| {
2426 Project::local(
2427 client_a.clone(),
2428 client_a.user_store.clone(),
2429 lang_registry.clone(),
2430 fs.clone(),
2431 cx,
2432 )
2433 });
2434 let (worktree_a, _) = project_a
2435 .update(cx_a, |p, cx| {
2436 p.find_or_create_local_worktree("/dir", true, cx)
2437 })
2438 .await
2439 .unwrap();
2440 worktree_a
2441 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2442 .await;
2443 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2444
2445 // Join that project as client B
2446 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2447
2448 // Open a buffer as client A
2449 let buffer_a = project_a
2450 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx))
2451 .await
2452 .unwrap();
2453
2454 // Start opening the same buffer as client B
2455 let buffer_b = cx_b
2456 .background()
2457 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx)));
2458
2459 // Edit the buffer as client A while client B is still opening it.
2460 cx_b.background().simulate_random_delay().await;
2461 buffer_a.update(cx_a, |buf, cx| buf.edit([(0..0, "X")], cx));
2462 cx_b.background().simulate_random_delay().await;
2463 buffer_a.update(cx_a, |buf, cx| buf.edit([(1..1, "Y")], cx));
2464
2465 let text = buffer_a.read_with(cx_a, |buf, _| buf.text());
2466 let buffer_b = buffer_b.await.unwrap();
2467 buffer_b.condition(&cx_b, |buf, _| buf.text() == text).await;
2468 }
2469
2470 #[gpui::test(iterations = 10)]
2471 async fn test_leaving_worktree_while_opening_buffer(
2472 cx_a: &mut TestAppContext,
2473 cx_b: &mut TestAppContext,
2474 ) {
2475 cx_a.foreground().forbid_parking();
2476 let lang_registry = Arc::new(LanguageRegistry::test());
2477 let fs = FakeFs::new(cx_a.background());
2478
2479 // Connect to a server as 2 clients.
2480 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2481 let client_a = server.create_client(cx_a, "user_a").await;
2482 let mut client_b = server.create_client(cx_b, "user_b").await;
2483 server
2484 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2485 .await;
2486
2487 // Share a project as client A
2488 fs.insert_tree(
2489 "/dir",
2490 json!({
2491 "a.txt": "a-contents",
2492 }),
2493 )
2494 .await;
2495 let project_a = cx_a.update(|cx| {
2496 Project::local(
2497 client_a.clone(),
2498 client_a.user_store.clone(),
2499 lang_registry.clone(),
2500 fs.clone(),
2501 cx,
2502 )
2503 });
2504 let (worktree_a, _) = project_a
2505 .update(cx_a, |p, cx| {
2506 p.find_or_create_local_worktree("/dir", true, cx)
2507 })
2508 .await
2509 .unwrap();
2510 worktree_a
2511 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2512 .await;
2513 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2514
2515 // Join that project as client B
2516 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2517
2518 // See that a guest has joined as client A.
2519 project_a
2520 .condition(&cx_a, |p, _| p.collaborators().len() == 1)
2521 .await;
2522
2523 // Begin opening a buffer as client B, but leave the project before the open completes.
2524 let buffer_b = cx_b
2525 .background()
2526 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.txt"), cx)));
2527 cx_b.update(|_| {
2528 drop(client_b.project.take());
2529 drop(project_b);
2530 });
2531 drop(buffer_b);
2532
2533 // See that the guest has left.
2534 project_a
2535 .condition(&cx_a, |p, _| p.collaborators().len() == 0)
2536 .await;
2537 }
2538
2539 #[gpui::test(iterations = 10)]
2540 async fn test_leaving_project(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
2541 cx_a.foreground().forbid_parking();
2542 let lang_registry = Arc::new(LanguageRegistry::test());
2543 let fs = FakeFs::new(cx_a.background());
2544
2545 // Connect to a server as 2 clients.
2546 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2547 let client_a = server.create_client(cx_a, "user_a").await;
2548 let mut client_b = server.create_client(cx_b, "user_b").await;
2549 server
2550 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2551 .await;
2552
2553 // Share a project as client A
2554 fs.insert_tree(
2555 "/a",
2556 json!({
2557 "a.txt": "a-contents",
2558 "b.txt": "b-contents",
2559 }),
2560 )
2561 .await;
2562 let project_a = cx_a.update(|cx| {
2563 Project::local(
2564 client_a.clone(),
2565 client_a.user_store.clone(),
2566 lang_registry.clone(),
2567 fs.clone(),
2568 cx,
2569 )
2570 });
2571 let (worktree_a, _) = project_a
2572 .update(cx_a, |p, cx| {
2573 p.find_or_create_local_worktree("/a", true, cx)
2574 })
2575 .await
2576 .unwrap();
2577 worktree_a
2578 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2579 .await;
2580
2581 // Join that project as client B
2582 let _project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2583
2584 // Client A sees that a guest has joined.
2585 project_a
2586 .condition(cx_a, |p, _| p.collaborators().len() == 1)
2587 .await;
2588
2589 // Drop client B's connection and ensure client A observes client B leaving the project.
2590 client_b.disconnect(&cx_b.to_async()).unwrap();
2591 project_a
2592 .condition(cx_a, |p, _| p.collaborators().len() == 0)
2593 .await;
2594
2595 // Rejoin the project as client B
2596 let _project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2597
2598 // Client A sees that a guest has re-joined.
2599 project_a
2600 .condition(cx_a, |p, _| p.collaborators().len() == 1)
2601 .await;
2602
2603 // Simulate connection loss for client B and ensure client A observes client B leaving the project.
2604 client_b.wait_for_current_user(cx_b).await;
2605 server.disconnect_client(client_b.current_user_id(cx_b));
2606 cx_a.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
2607 project_a
2608 .condition(cx_a, |p, _| p.collaborators().len() == 0)
2609 .await;
2610 }
2611
2612 #[gpui::test(iterations = 10)]
2613 async fn test_collaborating_with_diagnostics(
2614 deterministic: Arc<Deterministic>,
2615 cx_a: &mut TestAppContext,
2616 cx_b: &mut TestAppContext,
2617 cx_c: &mut TestAppContext,
2618 ) {
2619 deterministic.forbid_parking();
2620 let lang_registry = Arc::new(LanguageRegistry::test());
2621 let fs = FakeFs::new(cx_a.background());
2622
2623 // Set up a fake language server.
2624 let mut language = Language::new(
2625 LanguageConfig {
2626 name: "Rust".into(),
2627 path_suffixes: vec!["rs".to_string()],
2628 ..Default::default()
2629 },
2630 Some(tree_sitter_rust::language()),
2631 );
2632 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
2633 lang_registry.add(Arc::new(language));
2634
2635 // Connect to a server as 2 clients.
2636 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2637 let client_a = server.create_client(cx_a, "user_a").await;
2638 let mut client_b = server.create_client(cx_b, "user_b").await;
2639 let mut client_c = server.create_client(cx_c, "user_c").await;
2640 server
2641 .make_contacts(vec![
2642 (&client_a, cx_a),
2643 (&client_b, cx_b),
2644 (&client_c, cx_c),
2645 ])
2646 .await;
2647
2648 // Share a project as client A
2649 fs.insert_tree(
2650 "/a",
2651 json!({
2652 "a.rs": "let one = two",
2653 "other.rs": "",
2654 }),
2655 )
2656 .await;
2657 let project_a = cx_a.update(|cx| {
2658 Project::local(
2659 client_a.clone(),
2660 client_a.user_store.clone(),
2661 lang_registry.clone(),
2662 fs.clone(),
2663 cx,
2664 )
2665 });
2666 let (worktree_a, _) = project_a
2667 .update(cx_a, |p, cx| {
2668 p.find_or_create_local_worktree("/a", true, cx)
2669 })
2670 .await
2671 .unwrap();
2672 worktree_a
2673 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2674 .await;
2675 let project_id = project_a.update(cx_a, |p, _| p.next_remote_id()).await;
2676 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2677
2678 // Cause the language server to start.
2679 let _buffer = cx_a
2680 .background()
2681 .spawn(project_a.update(cx_a, |project, cx| {
2682 project.open_buffer(
2683 ProjectPath {
2684 worktree_id,
2685 path: Path::new("other.rs").into(),
2686 },
2687 cx,
2688 )
2689 }))
2690 .await
2691 .unwrap();
2692
2693 // Join the worktree as client B.
2694 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2695
2696 // Simulate a language server reporting errors for a file.
2697 let mut fake_language_server = fake_language_servers.next().await.unwrap();
2698 fake_language_server
2699 .receive_notification::<lsp::notification::DidOpenTextDocument>()
2700 .await;
2701 fake_language_server.notify::<lsp::notification::PublishDiagnostics>(
2702 lsp::PublishDiagnosticsParams {
2703 uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
2704 version: None,
2705 diagnostics: vec![lsp::Diagnostic {
2706 severity: Some(lsp::DiagnosticSeverity::ERROR),
2707 range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 7)),
2708 message: "message 1".to_string(),
2709 ..Default::default()
2710 }],
2711 },
2712 );
2713
2714 // Wait for server to see the diagnostics update.
2715 deterministic.run_until_parked();
2716 {
2717 let store = server.store.read().await;
2718 let project = store.project(project_id).unwrap();
2719 let worktree = project.worktrees.get(&worktree_id.to_proto()).unwrap();
2720 assert!(!worktree.diagnostic_summaries.is_empty());
2721 }
2722
2723 // Ensure client B observes the new diagnostics.
2724 project_b.read_with(cx_b, |project, cx| {
2725 assert_eq!(
2726 project.diagnostic_summaries(cx).collect::<Vec<_>>(),
2727 &[(
2728 ProjectPath {
2729 worktree_id,
2730 path: Arc::from(Path::new("a.rs")),
2731 },
2732 DiagnosticSummary {
2733 error_count: 1,
2734 warning_count: 0,
2735 ..Default::default()
2736 },
2737 )]
2738 )
2739 });
2740
2741 // Join project as client C and observe the diagnostics.
2742 let project_c = client_c.build_remote_project(&project_a, cx_a, cx_c).await;
2743 project_c.read_with(cx_c, |project, cx| {
2744 assert_eq!(
2745 project.diagnostic_summaries(cx).collect::<Vec<_>>(),
2746 &[(
2747 ProjectPath {
2748 worktree_id,
2749 path: Arc::from(Path::new("a.rs")),
2750 },
2751 DiagnosticSummary {
2752 error_count: 1,
2753 warning_count: 0,
2754 ..Default::default()
2755 },
2756 )]
2757 )
2758 });
2759
2760 // Simulate a language server reporting more errors for a file.
2761 fake_language_server.notify::<lsp::notification::PublishDiagnostics>(
2762 lsp::PublishDiagnosticsParams {
2763 uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
2764 version: None,
2765 diagnostics: vec![
2766 lsp::Diagnostic {
2767 severity: Some(lsp::DiagnosticSeverity::ERROR),
2768 range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 7)),
2769 message: "message 1".to_string(),
2770 ..Default::default()
2771 },
2772 lsp::Diagnostic {
2773 severity: Some(lsp::DiagnosticSeverity::WARNING),
2774 range: lsp::Range::new(
2775 lsp::Position::new(0, 10),
2776 lsp::Position::new(0, 13),
2777 ),
2778 message: "message 2".to_string(),
2779 ..Default::default()
2780 },
2781 ],
2782 },
2783 );
2784
2785 // Clients B and C get the updated summaries
2786 deterministic.run_until_parked();
2787 project_b.read_with(cx_b, |project, cx| {
2788 assert_eq!(
2789 project.diagnostic_summaries(cx).collect::<Vec<_>>(),
2790 [(
2791 ProjectPath {
2792 worktree_id,
2793 path: Arc::from(Path::new("a.rs")),
2794 },
2795 DiagnosticSummary {
2796 error_count: 1,
2797 warning_count: 1,
2798 ..Default::default()
2799 },
2800 )]
2801 );
2802 });
2803 project_c.read_with(cx_c, |project, cx| {
2804 assert_eq!(
2805 project.diagnostic_summaries(cx).collect::<Vec<_>>(),
2806 [(
2807 ProjectPath {
2808 worktree_id,
2809 path: Arc::from(Path::new("a.rs")),
2810 },
2811 DiagnosticSummary {
2812 error_count: 1,
2813 warning_count: 1,
2814 ..Default::default()
2815 },
2816 )]
2817 );
2818 });
2819
2820 // Open the file with the errors on client B. They should be present.
2821 let buffer_b = cx_b
2822 .background()
2823 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
2824 .await
2825 .unwrap();
2826
2827 buffer_b.read_with(cx_b, |buffer, _| {
2828 assert_eq!(
2829 buffer
2830 .snapshot()
2831 .diagnostics_in_range::<_, Point>(0..buffer.len(), false)
2832 .map(|entry| entry)
2833 .collect::<Vec<_>>(),
2834 &[
2835 DiagnosticEntry {
2836 range: Point::new(0, 4)..Point::new(0, 7),
2837 diagnostic: Diagnostic {
2838 group_id: 0,
2839 message: "message 1".to_string(),
2840 severity: lsp::DiagnosticSeverity::ERROR,
2841 is_primary: true,
2842 ..Default::default()
2843 }
2844 },
2845 DiagnosticEntry {
2846 range: Point::new(0, 10)..Point::new(0, 13),
2847 diagnostic: Diagnostic {
2848 group_id: 1,
2849 severity: lsp::DiagnosticSeverity::WARNING,
2850 message: "message 2".to_string(),
2851 is_primary: true,
2852 ..Default::default()
2853 }
2854 }
2855 ]
2856 );
2857 });
2858
2859 // Simulate a language server reporting no errors for a file.
2860 fake_language_server.notify::<lsp::notification::PublishDiagnostics>(
2861 lsp::PublishDiagnosticsParams {
2862 uri: lsp::Url::from_file_path("/a/a.rs").unwrap(),
2863 version: None,
2864 diagnostics: vec![],
2865 },
2866 );
2867 deterministic.run_until_parked();
2868 project_a.read_with(cx_a, |project, cx| {
2869 assert_eq!(project.diagnostic_summaries(cx).collect::<Vec<_>>(), [])
2870 });
2871 project_b.read_with(cx_b, |project, cx| {
2872 assert_eq!(project.diagnostic_summaries(cx).collect::<Vec<_>>(), [])
2873 });
2874 project_c.read_with(cx_c, |project, cx| {
2875 assert_eq!(project.diagnostic_summaries(cx).collect::<Vec<_>>(), [])
2876 });
2877 }
2878
2879 #[gpui::test(iterations = 10)]
2880 async fn test_collaborating_with_completion(
2881 cx_a: &mut TestAppContext,
2882 cx_b: &mut TestAppContext,
2883 ) {
2884 cx_a.foreground().forbid_parking();
2885 let lang_registry = Arc::new(LanguageRegistry::test());
2886 let fs = FakeFs::new(cx_a.background());
2887
2888 // Set up a fake language server.
2889 let mut language = Language::new(
2890 LanguageConfig {
2891 name: "Rust".into(),
2892 path_suffixes: vec!["rs".to_string()],
2893 ..Default::default()
2894 },
2895 Some(tree_sitter_rust::language()),
2896 );
2897 let mut fake_language_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
2898 capabilities: lsp::ServerCapabilities {
2899 completion_provider: Some(lsp::CompletionOptions {
2900 trigger_characters: Some(vec![".".to_string()]),
2901 ..Default::default()
2902 }),
2903 ..Default::default()
2904 },
2905 ..Default::default()
2906 });
2907 lang_registry.add(Arc::new(language));
2908
2909 // Connect to a server as 2 clients.
2910 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
2911 let client_a = server.create_client(cx_a, "user_a").await;
2912 let mut client_b = server.create_client(cx_b, "user_b").await;
2913 server
2914 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
2915 .await;
2916
2917 // Share a project as client A
2918 fs.insert_tree(
2919 "/a",
2920 json!({
2921 "main.rs": "fn main() { a }",
2922 "other.rs": "",
2923 }),
2924 )
2925 .await;
2926 let project_a = cx_a.update(|cx| {
2927 Project::local(
2928 client_a.clone(),
2929 client_a.user_store.clone(),
2930 lang_registry.clone(),
2931 fs.clone(),
2932 cx,
2933 )
2934 });
2935 let (worktree_a, _) = project_a
2936 .update(cx_a, |p, cx| {
2937 p.find_or_create_local_worktree("/a", true, cx)
2938 })
2939 .await
2940 .unwrap();
2941 worktree_a
2942 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
2943 .await;
2944 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
2945
2946 // Join the worktree as client B.
2947 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
2948
2949 // Open a file in an editor as the guest.
2950 let buffer_b = project_b
2951 .update(cx_b, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx))
2952 .await
2953 .unwrap();
2954 let (window_b, _) = cx_b.add_window(|_| EmptyView);
2955 let editor_b = cx_b.add_view(window_b, |cx| {
2956 Editor::for_buffer(buffer_b.clone(), Some(project_b.clone()), cx)
2957 });
2958
2959 let fake_language_server = fake_language_servers.next().await.unwrap();
2960 buffer_b
2961 .condition(&cx_b, |buffer, _| !buffer.completion_triggers().is_empty())
2962 .await;
2963
2964 // Type a completion trigger character as the guest.
2965 editor_b.update(cx_b, |editor, cx| {
2966 editor.select_ranges([13..13], None, cx);
2967 editor.handle_input(&Input(".".into()), cx);
2968 cx.focus(&editor_b);
2969 });
2970
2971 // Receive a completion request as the host's language server.
2972 // Return some completions from the host's language server.
2973 cx_a.foreground().start_waiting();
2974 fake_language_server
2975 .handle_request::<lsp::request::Completion, _, _>(|params, _| async move {
2976 assert_eq!(
2977 params.text_document_position.text_document.uri,
2978 lsp::Url::from_file_path("/a/main.rs").unwrap(),
2979 );
2980 assert_eq!(
2981 params.text_document_position.position,
2982 lsp::Position::new(0, 14),
2983 );
2984
2985 Ok(Some(lsp::CompletionResponse::Array(vec![
2986 lsp::CompletionItem {
2987 label: "first_method(…)".into(),
2988 detail: Some("fn(&mut self, B) -> C".into()),
2989 text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
2990 new_text: "first_method($1)".to_string(),
2991 range: lsp::Range::new(
2992 lsp::Position::new(0, 14),
2993 lsp::Position::new(0, 14),
2994 ),
2995 })),
2996 insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
2997 ..Default::default()
2998 },
2999 lsp::CompletionItem {
3000 label: "second_method(…)".into(),
3001 detail: Some("fn(&mut self, C) -> D<E>".into()),
3002 text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
3003 new_text: "second_method()".to_string(),
3004 range: lsp::Range::new(
3005 lsp::Position::new(0, 14),
3006 lsp::Position::new(0, 14),
3007 ),
3008 })),
3009 insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
3010 ..Default::default()
3011 },
3012 ])))
3013 })
3014 .next()
3015 .await
3016 .unwrap();
3017 cx_a.foreground().finish_waiting();
3018
3019 // Open the buffer on the host.
3020 let buffer_a = project_a
3021 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx))
3022 .await
3023 .unwrap();
3024 buffer_a
3025 .condition(&cx_a, |buffer, _| buffer.text() == "fn main() { a. }")
3026 .await;
3027
3028 // Confirm a completion on the guest.
3029 editor_b
3030 .condition(&cx_b, |editor, _| editor.context_menu_visible())
3031 .await;
3032 editor_b.update(cx_b, |editor, cx| {
3033 editor.confirm_completion(&ConfirmCompletion { item_ix: Some(0) }, cx);
3034 assert_eq!(editor.text(cx), "fn main() { a.first_method() }");
3035 });
3036
3037 // Return a resolved completion from the host's language server.
3038 // The resolved completion has an additional text edit.
3039 fake_language_server.handle_request::<lsp::request::ResolveCompletionItem, _, _>(
3040 |params, _| async move {
3041 assert_eq!(params.label, "first_method(…)");
3042 Ok(lsp::CompletionItem {
3043 label: "first_method(…)".into(),
3044 detail: Some("fn(&mut self, B) -> C".into()),
3045 text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
3046 new_text: "first_method($1)".to_string(),
3047 range: lsp::Range::new(
3048 lsp::Position::new(0, 14),
3049 lsp::Position::new(0, 14),
3050 ),
3051 })),
3052 additional_text_edits: Some(vec![lsp::TextEdit {
3053 new_text: "use d::SomeTrait;\n".to_string(),
3054 range: lsp::Range::new(lsp::Position::new(0, 0), lsp::Position::new(0, 0)),
3055 }]),
3056 insert_text_format: Some(lsp::InsertTextFormat::SNIPPET),
3057 ..Default::default()
3058 })
3059 },
3060 );
3061
3062 // The additional edit is applied.
3063 buffer_a
3064 .condition(&cx_a, |buffer, _| {
3065 buffer.text() == "use d::SomeTrait;\nfn main() { a.first_method() }"
3066 })
3067 .await;
3068 buffer_b
3069 .condition(&cx_b, |buffer, _| {
3070 buffer.text() == "use d::SomeTrait;\nfn main() { a.first_method() }"
3071 })
3072 .await;
3073 }
3074
3075 #[gpui::test(iterations = 10)]
3076 async fn test_reloading_buffer_manually(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3077 cx_a.foreground().forbid_parking();
3078 let lang_registry = Arc::new(LanguageRegistry::test());
3079 let fs = FakeFs::new(cx_a.background());
3080
3081 // Connect to a server as 2 clients.
3082 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3083 let client_a = server.create_client(cx_a, "user_a").await;
3084 let mut client_b = server.create_client(cx_b, "user_b").await;
3085 server
3086 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3087 .await;
3088
3089 // Share a project as client A
3090 fs.insert_tree(
3091 "/a",
3092 json!({
3093 "a.rs": "let one = 1;",
3094 }),
3095 )
3096 .await;
3097 let project_a = cx_a.update(|cx| {
3098 Project::local(
3099 client_a.clone(),
3100 client_a.user_store.clone(),
3101 lang_registry.clone(),
3102 fs.clone(),
3103 cx,
3104 )
3105 });
3106 let (worktree_a, _) = project_a
3107 .update(cx_a, |p, cx| {
3108 p.find_or_create_local_worktree("/a", true, cx)
3109 })
3110 .await
3111 .unwrap();
3112 worktree_a
3113 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3114 .await;
3115 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3116 let buffer_a = project_a
3117 .update(cx_a, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx))
3118 .await
3119 .unwrap();
3120
3121 // Join the worktree as client B.
3122 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3123
3124 let buffer_b = cx_b
3125 .background()
3126 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3127 .await
3128 .unwrap();
3129 buffer_b.update(cx_b, |buffer, cx| {
3130 buffer.edit([(4..7, "six")], cx);
3131 buffer.edit([(10..11, "6")], cx);
3132 assert_eq!(buffer.text(), "let six = 6;");
3133 assert!(buffer.is_dirty());
3134 assert!(!buffer.has_conflict());
3135 });
3136 buffer_a
3137 .condition(cx_a, |buffer, _| buffer.text() == "let six = 6;")
3138 .await;
3139
3140 fs.save(Path::new("/a/a.rs"), &Rope::from("let seven = 7;"))
3141 .await
3142 .unwrap();
3143 buffer_a
3144 .condition(cx_a, |buffer, _| buffer.has_conflict())
3145 .await;
3146 buffer_b
3147 .condition(cx_b, |buffer, _| buffer.has_conflict())
3148 .await;
3149
3150 project_b
3151 .update(cx_b, |project, cx| {
3152 project.reload_buffers(HashSet::from_iter([buffer_b.clone()]), true, cx)
3153 })
3154 .await
3155 .unwrap();
3156 buffer_a.read_with(cx_a, |buffer, _| {
3157 assert_eq!(buffer.text(), "let seven = 7;");
3158 assert!(!buffer.is_dirty());
3159 assert!(!buffer.has_conflict());
3160 });
3161 buffer_b.read_with(cx_b, |buffer, _| {
3162 assert_eq!(buffer.text(), "let seven = 7;");
3163 assert!(!buffer.is_dirty());
3164 assert!(!buffer.has_conflict());
3165 });
3166
3167 buffer_a.update(cx_a, |buffer, cx| {
3168 // Undoing on the host is a no-op when the reload was initiated by the guest.
3169 buffer.undo(cx);
3170 assert_eq!(buffer.text(), "let seven = 7;");
3171 assert!(!buffer.is_dirty());
3172 assert!(!buffer.has_conflict());
3173 });
3174 buffer_b.update(cx_b, |buffer, cx| {
3175 // Undoing on the guest rolls back the buffer to before it was reloaded but the conflict gets cleared.
3176 buffer.undo(cx);
3177 assert_eq!(buffer.text(), "let six = 6;");
3178 assert!(buffer.is_dirty());
3179 assert!(!buffer.has_conflict());
3180 });
3181 }
3182
3183 #[gpui::test(iterations = 10)]
3184 async fn test_formatting_buffer(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3185 cx_a.foreground().forbid_parking();
3186 let lang_registry = Arc::new(LanguageRegistry::test());
3187 let fs = FakeFs::new(cx_a.background());
3188
3189 // Set up a fake language server.
3190 let mut language = Language::new(
3191 LanguageConfig {
3192 name: "Rust".into(),
3193 path_suffixes: vec!["rs".to_string()],
3194 ..Default::default()
3195 },
3196 Some(tree_sitter_rust::language()),
3197 );
3198 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3199 lang_registry.add(Arc::new(language));
3200
3201 // Connect to a server as 2 clients.
3202 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3203 let client_a = server.create_client(cx_a, "user_a").await;
3204 let mut client_b = server.create_client(cx_b, "user_b").await;
3205 server
3206 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3207 .await;
3208
3209 // Share a project as client A
3210 fs.insert_tree(
3211 "/a",
3212 json!({
3213 "a.rs": "let one = two",
3214 }),
3215 )
3216 .await;
3217 let project_a = cx_a.update(|cx| {
3218 Project::local(
3219 client_a.clone(),
3220 client_a.user_store.clone(),
3221 lang_registry.clone(),
3222 fs.clone(),
3223 cx,
3224 )
3225 });
3226 let (worktree_a, _) = project_a
3227 .update(cx_a, |p, cx| {
3228 p.find_or_create_local_worktree("/a", true, cx)
3229 })
3230 .await
3231 .unwrap();
3232 worktree_a
3233 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3234 .await;
3235 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3236
3237 // Join the project as client B.
3238 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3239
3240 let buffer_b = cx_b
3241 .background()
3242 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3243 .await
3244 .unwrap();
3245
3246 let fake_language_server = fake_language_servers.next().await.unwrap();
3247 fake_language_server.handle_request::<lsp::request::Formatting, _, _>(|_, _| async move {
3248 Ok(Some(vec![
3249 lsp::TextEdit {
3250 range: lsp::Range::new(lsp::Position::new(0, 4), lsp::Position::new(0, 4)),
3251 new_text: "h".to_string(),
3252 },
3253 lsp::TextEdit {
3254 range: lsp::Range::new(lsp::Position::new(0, 7), lsp::Position::new(0, 7)),
3255 new_text: "y".to_string(),
3256 },
3257 ]))
3258 });
3259
3260 project_b
3261 .update(cx_b, |project, cx| {
3262 project.format(HashSet::from_iter([buffer_b.clone()]), true, cx)
3263 })
3264 .await
3265 .unwrap();
3266 assert_eq!(
3267 buffer_b.read_with(cx_b, |buffer, _| buffer.text()),
3268 "let honey = two"
3269 );
3270 }
3271
3272 #[gpui::test(iterations = 10)]
3273 async fn test_definition(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3274 cx_a.foreground().forbid_parking();
3275 let lang_registry = Arc::new(LanguageRegistry::test());
3276 let fs = FakeFs::new(cx_a.background());
3277 fs.insert_tree(
3278 "/root-1",
3279 json!({
3280 "a.rs": "const ONE: usize = b::TWO + b::THREE;",
3281 }),
3282 )
3283 .await;
3284 fs.insert_tree(
3285 "/root-2",
3286 json!({
3287 "b.rs": "const TWO: usize = 2;\nconst THREE: usize = 3;",
3288 }),
3289 )
3290 .await;
3291
3292 // Set up a fake language server.
3293 let mut language = Language::new(
3294 LanguageConfig {
3295 name: "Rust".into(),
3296 path_suffixes: vec!["rs".to_string()],
3297 ..Default::default()
3298 },
3299 Some(tree_sitter_rust::language()),
3300 );
3301 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3302 lang_registry.add(Arc::new(language));
3303
3304 // Connect to a server as 2 clients.
3305 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3306 let client_a = server.create_client(cx_a, "user_a").await;
3307 let mut client_b = server.create_client(cx_b, "user_b").await;
3308 server
3309 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3310 .await;
3311
3312 // Share a project as client A
3313 let project_a = cx_a.update(|cx| {
3314 Project::local(
3315 client_a.clone(),
3316 client_a.user_store.clone(),
3317 lang_registry.clone(),
3318 fs.clone(),
3319 cx,
3320 )
3321 });
3322 let (worktree_a, _) = project_a
3323 .update(cx_a, |p, cx| {
3324 p.find_or_create_local_worktree("/root-1", true, cx)
3325 })
3326 .await
3327 .unwrap();
3328 worktree_a
3329 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3330 .await;
3331 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3332
3333 // Join the project as client B.
3334 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3335
3336 // Open the file on client B.
3337 let buffer_b = cx_b
3338 .background()
3339 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3340 .await
3341 .unwrap();
3342
3343 // Request the definition of a symbol as the guest.
3344 let fake_language_server = fake_language_servers.next().await.unwrap();
3345 fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
3346 |_, _| async move {
3347 Ok(Some(lsp::GotoDefinitionResponse::Scalar(
3348 lsp::Location::new(
3349 lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
3350 lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3351 ),
3352 )))
3353 },
3354 );
3355
3356 let definitions_1 = project_b
3357 .update(cx_b, |p, cx| p.definition(&buffer_b, 23, cx))
3358 .await
3359 .unwrap();
3360 cx_b.read(|cx| {
3361 assert_eq!(definitions_1.len(), 1);
3362 assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
3363 let target_buffer = definitions_1[0].buffer.read(cx);
3364 assert_eq!(
3365 target_buffer.text(),
3366 "const TWO: usize = 2;\nconst THREE: usize = 3;"
3367 );
3368 assert_eq!(
3369 definitions_1[0].range.to_point(target_buffer),
3370 Point::new(0, 6)..Point::new(0, 9)
3371 );
3372 });
3373
3374 // Try getting more definitions for the same buffer, ensuring the buffer gets reused from
3375 // the previous call to `definition`.
3376 fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
3377 |_, _| async move {
3378 Ok(Some(lsp::GotoDefinitionResponse::Scalar(
3379 lsp::Location::new(
3380 lsp::Url::from_file_path("/root-2/b.rs").unwrap(),
3381 lsp::Range::new(lsp::Position::new(1, 6), lsp::Position::new(1, 11)),
3382 ),
3383 )))
3384 },
3385 );
3386
3387 let definitions_2 = project_b
3388 .update(cx_b, |p, cx| p.definition(&buffer_b, 33, cx))
3389 .await
3390 .unwrap();
3391 cx_b.read(|cx| {
3392 assert_eq!(definitions_2.len(), 1);
3393 assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
3394 let target_buffer = definitions_2[0].buffer.read(cx);
3395 assert_eq!(
3396 target_buffer.text(),
3397 "const TWO: usize = 2;\nconst THREE: usize = 3;"
3398 );
3399 assert_eq!(
3400 definitions_2[0].range.to_point(target_buffer),
3401 Point::new(1, 6)..Point::new(1, 11)
3402 );
3403 });
3404 assert_eq!(definitions_1[0].buffer, definitions_2[0].buffer);
3405 }
3406
3407 #[gpui::test(iterations = 10)]
3408 async fn test_references(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3409 cx_a.foreground().forbid_parking();
3410 let lang_registry = Arc::new(LanguageRegistry::test());
3411 let fs = FakeFs::new(cx_a.background());
3412 fs.insert_tree(
3413 "/root-1",
3414 json!({
3415 "one.rs": "const ONE: usize = 1;",
3416 "two.rs": "const TWO: usize = one::ONE + one::ONE;",
3417 }),
3418 )
3419 .await;
3420 fs.insert_tree(
3421 "/root-2",
3422 json!({
3423 "three.rs": "const THREE: usize = two::TWO + one::ONE;",
3424 }),
3425 )
3426 .await;
3427
3428 // Set up a fake language server.
3429 let mut language = Language::new(
3430 LanguageConfig {
3431 name: "Rust".into(),
3432 path_suffixes: vec!["rs".to_string()],
3433 ..Default::default()
3434 },
3435 Some(tree_sitter_rust::language()),
3436 );
3437 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3438 lang_registry.add(Arc::new(language));
3439
3440 // Connect to a server as 2 clients.
3441 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3442 let client_a = server.create_client(cx_a, "user_a").await;
3443 let mut client_b = server.create_client(cx_b, "user_b").await;
3444 server
3445 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3446 .await;
3447
3448 // Share a project as client A
3449 let project_a = cx_a.update(|cx| {
3450 Project::local(
3451 client_a.clone(),
3452 client_a.user_store.clone(),
3453 lang_registry.clone(),
3454 fs.clone(),
3455 cx,
3456 )
3457 });
3458 let (worktree_a, _) = project_a
3459 .update(cx_a, |p, cx| {
3460 p.find_or_create_local_worktree("/root-1", true, cx)
3461 })
3462 .await
3463 .unwrap();
3464 worktree_a
3465 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3466 .await;
3467 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3468
3469 // Join the worktree as client B.
3470 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3471
3472 // Open the file on client B.
3473 let buffer_b = cx_b
3474 .background()
3475 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "one.rs"), cx)))
3476 .await
3477 .unwrap();
3478
3479 // Request references to a symbol as the guest.
3480 let fake_language_server = fake_language_servers.next().await.unwrap();
3481 fake_language_server.handle_request::<lsp::request::References, _, _>(
3482 |params, _| async move {
3483 assert_eq!(
3484 params.text_document_position.text_document.uri.as_str(),
3485 "file:///root-1/one.rs"
3486 );
3487 Ok(Some(vec![
3488 lsp::Location {
3489 uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
3490 range: lsp::Range::new(
3491 lsp::Position::new(0, 24),
3492 lsp::Position::new(0, 27),
3493 ),
3494 },
3495 lsp::Location {
3496 uri: lsp::Url::from_file_path("/root-1/two.rs").unwrap(),
3497 range: lsp::Range::new(
3498 lsp::Position::new(0, 35),
3499 lsp::Position::new(0, 38),
3500 ),
3501 },
3502 lsp::Location {
3503 uri: lsp::Url::from_file_path("/root-2/three.rs").unwrap(),
3504 range: lsp::Range::new(
3505 lsp::Position::new(0, 37),
3506 lsp::Position::new(0, 40),
3507 ),
3508 },
3509 ]))
3510 },
3511 );
3512
3513 let references = project_b
3514 .update(cx_b, |p, cx| p.references(&buffer_b, 7, cx))
3515 .await
3516 .unwrap();
3517 cx_b.read(|cx| {
3518 assert_eq!(references.len(), 3);
3519 assert_eq!(project_b.read(cx).worktrees(cx).count(), 2);
3520
3521 let two_buffer = references[0].buffer.read(cx);
3522 let three_buffer = references[2].buffer.read(cx);
3523 assert_eq!(
3524 two_buffer.file().unwrap().path().as_ref(),
3525 Path::new("two.rs")
3526 );
3527 assert_eq!(references[1].buffer, references[0].buffer);
3528 assert_eq!(
3529 three_buffer.file().unwrap().full_path(cx),
3530 Path::new("three.rs")
3531 );
3532
3533 assert_eq!(references[0].range.to_offset(&two_buffer), 24..27);
3534 assert_eq!(references[1].range.to_offset(&two_buffer), 35..38);
3535 assert_eq!(references[2].range.to_offset(&three_buffer), 37..40);
3536 });
3537 }
3538
3539 #[gpui::test(iterations = 10)]
3540 async fn test_project_search(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3541 cx_a.foreground().forbid_parking();
3542 let lang_registry = Arc::new(LanguageRegistry::test());
3543 let fs = FakeFs::new(cx_a.background());
3544 fs.insert_tree(
3545 "/root-1",
3546 json!({
3547 "a": "hello world",
3548 "b": "goodnight moon",
3549 "c": "a world of goo",
3550 "d": "world champion of clown world",
3551 }),
3552 )
3553 .await;
3554 fs.insert_tree(
3555 "/root-2",
3556 json!({
3557 "e": "disney world is fun",
3558 }),
3559 )
3560 .await;
3561
3562 // Connect to a server as 2 clients.
3563 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3564 let client_a = server.create_client(cx_a, "user_a").await;
3565 let mut client_b = server.create_client(cx_b, "user_b").await;
3566 server
3567 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3568 .await;
3569
3570 // Share a project as client A
3571 let project_a = cx_a.update(|cx| {
3572 Project::local(
3573 client_a.clone(),
3574 client_a.user_store.clone(),
3575 lang_registry.clone(),
3576 fs.clone(),
3577 cx,
3578 )
3579 });
3580
3581 let (worktree_1, _) = project_a
3582 .update(cx_a, |p, cx| {
3583 p.find_or_create_local_worktree("/root-1", true, cx)
3584 })
3585 .await
3586 .unwrap();
3587 worktree_1
3588 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3589 .await;
3590 let (worktree_2, _) = project_a
3591 .update(cx_a, |p, cx| {
3592 p.find_or_create_local_worktree("/root-2", true, cx)
3593 })
3594 .await
3595 .unwrap();
3596 worktree_2
3597 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3598 .await;
3599
3600 // Join the worktree as client B.
3601 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3602 let results = project_b
3603 .update(cx_b, |project, cx| {
3604 project.search(SearchQuery::text("world", false, false), cx)
3605 })
3606 .await
3607 .unwrap();
3608
3609 let mut ranges_by_path = results
3610 .into_iter()
3611 .map(|(buffer, ranges)| {
3612 buffer.read_with(cx_b, |buffer, cx| {
3613 let path = buffer.file().unwrap().full_path(cx);
3614 let offset_ranges = ranges
3615 .into_iter()
3616 .map(|range| range.to_offset(buffer))
3617 .collect::<Vec<_>>();
3618 (path, offset_ranges)
3619 })
3620 })
3621 .collect::<Vec<_>>();
3622 ranges_by_path.sort_by_key(|(path, _)| path.clone());
3623
3624 assert_eq!(
3625 ranges_by_path,
3626 &[
3627 (PathBuf::from("root-1/a"), vec![6..11]),
3628 (PathBuf::from("root-1/c"), vec![2..7]),
3629 (PathBuf::from("root-1/d"), vec![0..5, 24..29]),
3630 (PathBuf::from("root-2/e"), vec![7..12]),
3631 ]
3632 );
3633 }
3634
3635 #[gpui::test(iterations = 10)]
3636 async fn test_document_highlights(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3637 cx_a.foreground().forbid_parking();
3638 let lang_registry = Arc::new(LanguageRegistry::test());
3639 let fs = FakeFs::new(cx_a.background());
3640 fs.insert_tree(
3641 "/root-1",
3642 json!({
3643 "main.rs": "fn double(number: i32) -> i32 { number + number }",
3644 }),
3645 )
3646 .await;
3647
3648 // Set up a fake language server.
3649 let mut language = Language::new(
3650 LanguageConfig {
3651 name: "Rust".into(),
3652 path_suffixes: vec!["rs".to_string()],
3653 ..Default::default()
3654 },
3655 Some(tree_sitter_rust::language()),
3656 );
3657 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3658 lang_registry.add(Arc::new(language));
3659
3660 // Connect to a server as 2 clients.
3661 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3662 let client_a = server.create_client(cx_a, "user_a").await;
3663 let mut client_b = server.create_client(cx_b, "user_b").await;
3664 server
3665 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3666 .await;
3667
3668 // Share a project as client A
3669 let project_a = cx_a.update(|cx| {
3670 Project::local(
3671 client_a.clone(),
3672 client_a.user_store.clone(),
3673 lang_registry.clone(),
3674 fs.clone(),
3675 cx,
3676 )
3677 });
3678 let (worktree_a, _) = project_a
3679 .update(cx_a, |p, cx| {
3680 p.find_or_create_local_worktree("/root-1", true, cx)
3681 })
3682 .await
3683 .unwrap();
3684 worktree_a
3685 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3686 .await;
3687 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3688
3689 // Join the worktree as client B.
3690 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3691
3692 // Open the file on client B.
3693 let buffer_b = cx_b
3694 .background()
3695 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "main.rs"), cx)))
3696 .await
3697 .unwrap();
3698
3699 // Request document highlights as the guest.
3700 let fake_language_server = fake_language_servers.next().await.unwrap();
3701 fake_language_server.handle_request::<lsp::request::DocumentHighlightRequest, _, _>(
3702 |params, _| async move {
3703 assert_eq!(
3704 params
3705 .text_document_position_params
3706 .text_document
3707 .uri
3708 .as_str(),
3709 "file:///root-1/main.rs"
3710 );
3711 assert_eq!(
3712 params.text_document_position_params.position,
3713 lsp::Position::new(0, 34)
3714 );
3715 Ok(Some(vec![
3716 lsp::DocumentHighlight {
3717 kind: Some(lsp::DocumentHighlightKind::WRITE),
3718 range: lsp::Range::new(
3719 lsp::Position::new(0, 10),
3720 lsp::Position::new(0, 16),
3721 ),
3722 },
3723 lsp::DocumentHighlight {
3724 kind: Some(lsp::DocumentHighlightKind::READ),
3725 range: lsp::Range::new(
3726 lsp::Position::new(0, 32),
3727 lsp::Position::new(0, 38),
3728 ),
3729 },
3730 lsp::DocumentHighlight {
3731 kind: Some(lsp::DocumentHighlightKind::READ),
3732 range: lsp::Range::new(
3733 lsp::Position::new(0, 41),
3734 lsp::Position::new(0, 47),
3735 ),
3736 },
3737 ]))
3738 },
3739 );
3740
3741 let highlights = project_b
3742 .update(cx_b, |p, cx| p.document_highlights(&buffer_b, 34, cx))
3743 .await
3744 .unwrap();
3745 buffer_b.read_with(cx_b, |buffer, _| {
3746 let snapshot = buffer.snapshot();
3747
3748 let highlights = highlights
3749 .into_iter()
3750 .map(|highlight| (highlight.kind, highlight.range.to_offset(&snapshot)))
3751 .collect::<Vec<_>>();
3752 assert_eq!(
3753 highlights,
3754 &[
3755 (lsp::DocumentHighlightKind::WRITE, 10..16),
3756 (lsp::DocumentHighlightKind::READ, 32..38),
3757 (lsp::DocumentHighlightKind::READ, 41..47)
3758 ]
3759 )
3760 });
3761 }
3762
3763 #[gpui::test(iterations = 10)]
3764 async fn test_project_symbols(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
3765 cx_a.foreground().forbid_parking();
3766 let lang_registry = Arc::new(LanguageRegistry::test());
3767 let fs = FakeFs::new(cx_a.background());
3768 fs.insert_tree(
3769 "/code",
3770 json!({
3771 "crate-1": {
3772 "one.rs": "const ONE: usize = 1;",
3773 },
3774 "crate-2": {
3775 "two.rs": "const TWO: usize = 2; const THREE: usize = 3;",
3776 },
3777 "private": {
3778 "passwords.txt": "the-password",
3779 }
3780 }),
3781 )
3782 .await;
3783
3784 // Set up a fake language server.
3785 let mut language = Language::new(
3786 LanguageConfig {
3787 name: "Rust".into(),
3788 path_suffixes: vec!["rs".to_string()],
3789 ..Default::default()
3790 },
3791 Some(tree_sitter_rust::language()),
3792 );
3793 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3794 lang_registry.add(Arc::new(language));
3795
3796 // Connect to a server as 2 clients.
3797 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3798 let client_a = server.create_client(cx_a, "user_a").await;
3799 let mut client_b = server.create_client(cx_b, "user_b").await;
3800 server
3801 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3802 .await;
3803
3804 // Share a project as client A
3805 let project_a = cx_a.update(|cx| {
3806 Project::local(
3807 client_a.clone(),
3808 client_a.user_store.clone(),
3809 lang_registry.clone(),
3810 fs.clone(),
3811 cx,
3812 )
3813 });
3814 let (worktree_a, _) = project_a
3815 .update(cx_a, |p, cx| {
3816 p.find_or_create_local_worktree("/code/crate-1", true, cx)
3817 })
3818 .await
3819 .unwrap();
3820 worktree_a
3821 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3822 .await;
3823 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3824
3825 // Join the worktree as client B.
3826 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3827
3828 // Cause the language server to start.
3829 let _buffer = cx_b
3830 .background()
3831 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "one.rs"), cx)))
3832 .await
3833 .unwrap();
3834
3835 let fake_language_server = fake_language_servers.next().await.unwrap();
3836 fake_language_server.handle_request::<lsp::request::WorkspaceSymbol, _, _>(
3837 |_, _| async move {
3838 #[allow(deprecated)]
3839 Ok(Some(vec![lsp::SymbolInformation {
3840 name: "TWO".into(),
3841 location: lsp::Location {
3842 uri: lsp::Url::from_file_path("/code/crate-2/two.rs").unwrap(),
3843 range: lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3844 },
3845 kind: lsp::SymbolKind::CONSTANT,
3846 tags: None,
3847 container_name: None,
3848 deprecated: None,
3849 }]))
3850 },
3851 );
3852
3853 // Request the definition of a symbol as the guest.
3854 let symbols = project_b
3855 .update(cx_b, |p, cx| p.symbols("two", cx))
3856 .await
3857 .unwrap();
3858 assert_eq!(symbols.len(), 1);
3859 assert_eq!(symbols[0].name, "TWO");
3860
3861 // Open one of the returned symbols.
3862 let buffer_b_2 = project_b
3863 .update(cx_b, |project, cx| {
3864 project.open_buffer_for_symbol(&symbols[0], cx)
3865 })
3866 .await
3867 .unwrap();
3868 buffer_b_2.read_with(cx_b, |buffer, _| {
3869 assert_eq!(
3870 buffer.file().unwrap().path().as_ref(),
3871 Path::new("../crate-2/two.rs")
3872 );
3873 });
3874
3875 // Attempt to craft a symbol and violate host's privacy by opening an arbitrary file.
3876 let mut fake_symbol = symbols[0].clone();
3877 fake_symbol.path = Path::new("/code/secrets").into();
3878 let error = project_b
3879 .update(cx_b, |project, cx| {
3880 project.open_buffer_for_symbol(&fake_symbol, cx)
3881 })
3882 .await
3883 .unwrap_err();
3884 assert!(error.to_string().contains("invalid symbol signature"));
3885 }
3886
3887 #[gpui::test(iterations = 10)]
3888 async fn test_open_buffer_while_getting_definition_pointing_to_it(
3889 cx_a: &mut TestAppContext,
3890 cx_b: &mut TestAppContext,
3891 mut rng: StdRng,
3892 ) {
3893 cx_a.foreground().forbid_parking();
3894 let lang_registry = Arc::new(LanguageRegistry::test());
3895 let fs = FakeFs::new(cx_a.background());
3896 fs.insert_tree(
3897 "/root",
3898 json!({
3899 "a.rs": "const ONE: usize = b::TWO;",
3900 "b.rs": "const TWO: usize = 2",
3901 }),
3902 )
3903 .await;
3904
3905 // Set up a fake language server.
3906 let mut language = Language::new(
3907 LanguageConfig {
3908 name: "Rust".into(),
3909 path_suffixes: vec!["rs".to_string()],
3910 ..Default::default()
3911 },
3912 Some(tree_sitter_rust::language()),
3913 );
3914 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
3915 lang_registry.add(Arc::new(language));
3916
3917 // Connect to a server as 2 clients.
3918 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
3919 let client_a = server.create_client(cx_a, "user_a").await;
3920 let mut client_b = server.create_client(cx_b, "user_b").await;
3921 server
3922 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
3923 .await;
3924
3925 // Share a project as client A
3926 let project_a = cx_a.update(|cx| {
3927 Project::local(
3928 client_a.clone(),
3929 client_a.user_store.clone(),
3930 lang_registry.clone(),
3931 fs.clone(),
3932 cx,
3933 )
3934 });
3935
3936 let (worktree_a, _) = project_a
3937 .update(cx_a, |p, cx| {
3938 p.find_or_create_local_worktree("/root", true, cx)
3939 })
3940 .await
3941 .unwrap();
3942 worktree_a
3943 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
3944 .await;
3945 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
3946
3947 // Join the project as client B.
3948 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
3949
3950 let buffer_b1 = cx_b
3951 .background()
3952 .spawn(project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "a.rs"), cx)))
3953 .await
3954 .unwrap();
3955
3956 let fake_language_server = fake_language_servers.next().await.unwrap();
3957 fake_language_server.handle_request::<lsp::request::GotoDefinition, _, _>(
3958 |_, _| async move {
3959 Ok(Some(lsp::GotoDefinitionResponse::Scalar(
3960 lsp::Location::new(
3961 lsp::Url::from_file_path("/root/b.rs").unwrap(),
3962 lsp::Range::new(lsp::Position::new(0, 6), lsp::Position::new(0, 9)),
3963 ),
3964 )))
3965 },
3966 );
3967
3968 let definitions;
3969 let buffer_b2;
3970 if rng.gen() {
3971 definitions = project_b.update(cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
3972 buffer_b2 = project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
3973 } else {
3974 buffer_b2 = project_b.update(cx_b, |p, cx| p.open_buffer((worktree_id, "b.rs"), cx));
3975 definitions = project_b.update(cx_b, |p, cx| p.definition(&buffer_b1, 23, cx));
3976 }
3977
3978 let buffer_b2 = buffer_b2.await.unwrap();
3979 let definitions = definitions.await.unwrap();
3980 assert_eq!(definitions.len(), 1);
3981 assert_eq!(definitions[0].buffer, buffer_b2);
3982 }
3983
3984 #[gpui::test(iterations = 10)]
3985 async fn test_collaborating_with_code_actions(
3986 cx_a: &mut TestAppContext,
3987 cx_b: &mut TestAppContext,
3988 ) {
3989 cx_a.foreground().forbid_parking();
3990 let lang_registry = Arc::new(LanguageRegistry::test());
3991 let fs = FakeFs::new(cx_a.background());
3992 cx_b.update(|cx| editor::init(cx));
3993
3994 // Set up a fake language server.
3995 let mut language = Language::new(
3996 LanguageConfig {
3997 name: "Rust".into(),
3998 path_suffixes: vec!["rs".to_string()],
3999 ..Default::default()
4000 },
4001 Some(tree_sitter_rust::language()),
4002 );
4003 let mut fake_language_servers = language.set_fake_lsp_adapter(Default::default());
4004 lang_registry.add(Arc::new(language));
4005
4006 // Connect to a server as 2 clients.
4007 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4008 let client_a = server.create_client(cx_a, "user_a").await;
4009 let mut client_b = server.create_client(cx_b, "user_b").await;
4010 server
4011 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
4012 .await;
4013
4014 // Share a project as client A
4015 fs.insert_tree(
4016 "/a",
4017 json!({
4018 "main.rs": "mod other;\nfn main() { let foo = other::foo(); }",
4019 "other.rs": "pub fn foo() -> usize { 4 }",
4020 }),
4021 )
4022 .await;
4023 let project_a = cx_a.update(|cx| {
4024 Project::local(
4025 client_a.clone(),
4026 client_a.user_store.clone(),
4027 lang_registry.clone(),
4028 fs.clone(),
4029 cx,
4030 )
4031 });
4032 let (worktree_a, _) = project_a
4033 .update(cx_a, |p, cx| {
4034 p.find_or_create_local_worktree("/a", true, cx)
4035 })
4036 .await
4037 .unwrap();
4038 worktree_a
4039 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4040 .await;
4041 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
4042
4043 // Join the project as client B.
4044 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
4045 let mut params = cx_b.update(WorkspaceParams::test);
4046 params.languages = lang_registry.clone();
4047 params.project = project_b.clone();
4048 params.client = client_b.client.clone();
4049 params.user_store = client_b.user_store.clone();
4050
4051 let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(¶ms, cx));
4052 let editor_b = workspace_b
4053 .update(cx_b, |workspace, cx| {
4054 workspace.open_path((worktree_id, "main.rs"), true, cx)
4055 })
4056 .await
4057 .unwrap()
4058 .downcast::<Editor>()
4059 .unwrap();
4060
4061 let mut fake_language_server = fake_language_servers.next().await.unwrap();
4062 fake_language_server
4063 .handle_request::<lsp::request::CodeActionRequest, _, _>(|params, _| async move {
4064 assert_eq!(
4065 params.text_document.uri,
4066 lsp::Url::from_file_path("/a/main.rs").unwrap(),
4067 );
4068 assert_eq!(params.range.start, lsp::Position::new(0, 0));
4069 assert_eq!(params.range.end, lsp::Position::new(0, 0));
4070 Ok(None)
4071 })
4072 .next()
4073 .await;
4074
4075 // Move cursor to a location that contains code actions.
4076 editor_b.update(cx_b, |editor, cx| {
4077 editor.select_ranges([Point::new(1, 31)..Point::new(1, 31)], None, cx);
4078 cx.focus(&editor_b);
4079 });
4080
4081 fake_language_server
4082 .handle_request::<lsp::request::CodeActionRequest, _, _>(|params, _| async move {
4083 assert_eq!(
4084 params.text_document.uri,
4085 lsp::Url::from_file_path("/a/main.rs").unwrap(),
4086 );
4087 assert_eq!(params.range.start, lsp::Position::new(1, 31));
4088 assert_eq!(params.range.end, lsp::Position::new(1, 31));
4089
4090 Ok(Some(vec![lsp::CodeActionOrCommand::CodeAction(
4091 lsp::CodeAction {
4092 title: "Inline into all callers".to_string(),
4093 edit: Some(lsp::WorkspaceEdit {
4094 changes: Some(
4095 [
4096 (
4097 lsp::Url::from_file_path("/a/main.rs").unwrap(),
4098 vec![lsp::TextEdit::new(
4099 lsp::Range::new(
4100 lsp::Position::new(1, 22),
4101 lsp::Position::new(1, 34),
4102 ),
4103 "4".to_string(),
4104 )],
4105 ),
4106 (
4107 lsp::Url::from_file_path("/a/other.rs").unwrap(),
4108 vec![lsp::TextEdit::new(
4109 lsp::Range::new(
4110 lsp::Position::new(0, 0),
4111 lsp::Position::new(0, 27),
4112 ),
4113 "".to_string(),
4114 )],
4115 ),
4116 ]
4117 .into_iter()
4118 .collect(),
4119 ),
4120 ..Default::default()
4121 }),
4122 data: Some(json!({
4123 "codeActionParams": {
4124 "range": {
4125 "start": {"line": 1, "column": 31},
4126 "end": {"line": 1, "column": 31},
4127 }
4128 }
4129 })),
4130 ..Default::default()
4131 },
4132 )]))
4133 })
4134 .next()
4135 .await;
4136
4137 // Toggle code actions and wait for them to display.
4138 editor_b.update(cx_b, |editor, cx| {
4139 editor.toggle_code_actions(
4140 &ToggleCodeActions {
4141 deployed_from_indicator: false,
4142 },
4143 cx,
4144 );
4145 });
4146 editor_b
4147 .condition(&cx_b, |editor, _| editor.context_menu_visible())
4148 .await;
4149
4150 fake_language_server.remove_request_handler::<lsp::request::CodeActionRequest>();
4151
4152 // Confirming the code action will trigger a resolve request.
4153 let confirm_action = workspace_b
4154 .update(cx_b, |workspace, cx| {
4155 Editor::confirm_code_action(workspace, &ConfirmCodeAction { item_ix: Some(0) }, cx)
4156 })
4157 .unwrap();
4158 fake_language_server.handle_request::<lsp::request::CodeActionResolveRequest, _, _>(
4159 |_, _| async move {
4160 Ok(lsp::CodeAction {
4161 title: "Inline into all callers".to_string(),
4162 edit: Some(lsp::WorkspaceEdit {
4163 changes: Some(
4164 [
4165 (
4166 lsp::Url::from_file_path("/a/main.rs").unwrap(),
4167 vec![lsp::TextEdit::new(
4168 lsp::Range::new(
4169 lsp::Position::new(1, 22),
4170 lsp::Position::new(1, 34),
4171 ),
4172 "4".to_string(),
4173 )],
4174 ),
4175 (
4176 lsp::Url::from_file_path("/a/other.rs").unwrap(),
4177 vec![lsp::TextEdit::new(
4178 lsp::Range::new(
4179 lsp::Position::new(0, 0),
4180 lsp::Position::new(0, 27),
4181 ),
4182 "".to_string(),
4183 )],
4184 ),
4185 ]
4186 .into_iter()
4187 .collect(),
4188 ),
4189 ..Default::default()
4190 }),
4191 ..Default::default()
4192 })
4193 },
4194 );
4195
4196 // After the action is confirmed, an editor containing both modified files is opened.
4197 confirm_action.await.unwrap();
4198 let code_action_editor = workspace_b.read_with(cx_b, |workspace, cx| {
4199 workspace
4200 .active_item(cx)
4201 .unwrap()
4202 .downcast::<Editor>()
4203 .unwrap()
4204 });
4205 code_action_editor.update(cx_b, |editor, cx| {
4206 assert_eq!(editor.text(cx), "mod other;\nfn main() { let foo = 4; }\n");
4207 editor.undo(&Undo, cx);
4208 assert_eq!(
4209 editor.text(cx),
4210 "mod other;\nfn main() { let foo = other::foo(); }\npub fn foo() -> usize { 4 }"
4211 );
4212 editor.redo(&Redo, cx);
4213 assert_eq!(editor.text(cx), "mod other;\nfn main() { let foo = 4; }\n");
4214 });
4215 }
4216
4217 #[gpui::test(iterations = 10)]
4218 async fn test_collaborating_with_renames(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4219 cx_a.foreground().forbid_parking();
4220 let lang_registry = Arc::new(LanguageRegistry::test());
4221 let fs = FakeFs::new(cx_a.background());
4222 cx_b.update(|cx| editor::init(cx));
4223
4224 // Set up a fake language server.
4225 let mut language = Language::new(
4226 LanguageConfig {
4227 name: "Rust".into(),
4228 path_suffixes: vec!["rs".to_string()],
4229 ..Default::default()
4230 },
4231 Some(tree_sitter_rust::language()),
4232 );
4233 let mut fake_language_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
4234 capabilities: lsp::ServerCapabilities {
4235 rename_provider: Some(lsp::OneOf::Right(lsp::RenameOptions {
4236 prepare_provider: Some(true),
4237 work_done_progress_options: Default::default(),
4238 })),
4239 ..Default::default()
4240 },
4241 ..Default::default()
4242 });
4243 lang_registry.add(Arc::new(language));
4244
4245 // Connect to a server as 2 clients.
4246 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4247 let client_a = server.create_client(cx_a, "user_a").await;
4248 let mut client_b = server.create_client(cx_b, "user_b").await;
4249 server
4250 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
4251 .await;
4252
4253 // Share a project as client A
4254 fs.insert_tree(
4255 "/dir",
4256 json!({
4257 "one.rs": "const ONE: usize = 1;",
4258 "two.rs": "const TWO: usize = one::ONE + one::ONE;"
4259 }),
4260 )
4261 .await;
4262 let project_a = cx_a.update(|cx| {
4263 Project::local(
4264 client_a.clone(),
4265 client_a.user_store.clone(),
4266 lang_registry.clone(),
4267 fs.clone(),
4268 cx,
4269 )
4270 });
4271 let (worktree_a, _) = project_a
4272 .update(cx_a, |p, cx| {
4273 p.find_or_create_local_worktree("/dir", true, cx)
4274 })
4275 .await
4276 .unwrap();
4277 worktree_a
4278 .read_with(cx_a, |tree, _| tree.as_local().unwrap().scan_complete())
4279 .await;
4280 let worktree_id = worktree_a.read_with(cx_a, |tree, _| tree.id());
4281
4282 // Join the worktree as client B.
4283 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
4284 let mut params = cx_b.update(WorkspaceParams::test);
4285 params.languages = lang_registry.clone();
4286 params.project = project_b.clone();
4287 params.client = client_b.client.clone();
4288 params.user_store = client_b.user_store.clone();
4289
4290 let (_window_b, workspace_b) = cx_b.add_window(|cx| Workspace::new(¶ms, cx));
4291 let editor_b = workspace_b
4292 .update(cx_b, |workspace, cx| {
4293 workspace.open_path((worktree_id, "one.rs"), true, cx)
4294 })
4295 .await
4296 .unwrap()
4297 .downcast::<Editor>()
4298 .unwrap();
4299 let fake_language_server = fake_language_servers.next().await.unwrap();
4300
4301 // Move cursor to a location that can be renamed.
4302 let prepare_rename = editor_b.update(cx_b, |editor, cx| {
4303 editor.select_ranges([7..7], None, cx);
4304 editor.rename(&Rename, cx).unwrap()
4305 });
4306
4307 fake_language_server
4308 .handle_request::<lsp::request::PrepareRenameRequest, _, _>(|params, _| async move {
4309 assert_eq!(params.text_document.uri.as_str(), "file:///dir/one.rs");
4310 assert_eq!(params.position, lsp::Position::new(0, 7));
4311 Ok(Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
4312 lsp::Position::new(0, 6),
4313 lsp::Position::new(0, 9),
4314 ))))
4315 })
4316 .next()
4317 .await
4318 .unwrap();
4319 prepare_rename.await.unwrap();
4320 editor_b.update(cx_b, |editor, cx| {
4321 let rename = editor.pending_rename().unwrap();
4322 let buffer = editor.buffer().read(cx).snapshot(cx);
4323 assert_eq!(
4324 rename.range.start.to_offset(&buffer)..rename.range.end.to_offset(&buffer),
4325 6..9
4326 );
4327 rename.editor.update(cx, |rename_editor, cx| {
4328 rename_editor.buffer().update(cx, |rename_buffer, cx| {
4329 rename_buffer.edit([(0..3, "THREE")], cx);
4330 });
4331 });
4332 });
4333
4334 let confirm_rename = workspace_b.update(cx_b, |workspace, cx| {
4335 Editor::confirm_rename(workspace, &ConfirmRename, cx).unwrap()
4336 });
4337 fake_language_server
4338 .handle_request::<lsp::request::Rename, _, _>(|params, _| async move {
4339 assert_eq!(
4340 params.text_document_position.text_document.uri.as_str(),
4341 "file:///dir/one.rs"
4342 );
4343 assert_eq!(
4344 params.text_document_position.position,
4345 lsp::Position::new(0, 6)
4346 );
4347 assert_eq!(params.new_name, "THREE");
4348 Ok(Some(lsp::WorkspaceEdit {
4349 changes: Some(
4350 [
4351 (
4352 lsp::Url::from_file_path("/dir/one.rs").unwrap(),
4353 vec![lsp::TextEdit::new(
4354 lsp::Range::new(
4355 lsp::Position::new(0, 6),
4356 lsp::Position::new(0, 9),
4357 ),
4358 "THREE".to_string(),
4359 )],
4360 ),
4361 (
4362 lsp::Url::from_file_path("/dir/two.rs").unwrap(),
4363 vec![
4364 lsp::TextEdit::new(
4365 lsp::Range::new(
4366 lsp::Position::new(0, 24),
4367 lsp::Position::new(0, 27),
4368 ),
4369 "THREE".to_string(),
4370 ),
4371 lsp::TextEdit::new(
4372 lsp::Range::new(
4373 lsp::Position::new(0, 35),
4374 lsp::Position::new(0, 38),
4375 ),
4376 "THREE".to_string(),
4377 ),
4378 ],
4379 ),
4380 ]
4381 .into_iter()
4382 .collect(),
4383 ),
4384 ..Default::default()
4385 }))
4386 })
4387 .next()
4388 .await
4389 .unwrap();
4390 confirm_rename.await.unwrap();
4391
4392 let rename_editor = workspace_b.read_with(cx_b, |workspace, cx| {
4393 workspace
4394 .active_item(cx)
4395 .unwrap()
4396 .downcast::<Editor>()
4397 .unwrap()
4398 });
4399 rename_editor.update(cx_b, |editor, cx| {
4400 assert_eq!(
4401 editor.text(cx),
4402 "const THREE: usize = 1;\nconst TWO: usize = one::THREE + one::THREE;"
4403 );
4404 editor.undo(&Undo, cx);
4405 assert_eq!(
4406 editor.text(cx),
4407 "const ONE: usize = 1;\nconst TWO: usize = one::ONE + one::ONE;"
4408 );
4409 editor.redo(&Redo, cx);
4410 assert_eq!(
4411 editor.text(cx),
4412 "const THREE: usize = 1;\nconst TWO: usize = one::THREE + one::THREE;"
4413 );
4414 });
4415
4416 // Ensure temporary rename edits cannot be undone/redone.
4417 editor_b.update(cx_b, |editor, cx| {
4418 editor.undo(&Undo, cx);
4419 assert_eq!(editor.text(cx), "const ONE: usize = 1;");
4420 editor.undo(&Undo, cx);
4421 assert_eq!(editor.text(cx), "const ONE: usize = 1;");
4422 editor.redo(&Redo, cx);
4423 assert_eq!(editor.text(cx), "const THREE: usize = 1;");
4424 })
4425 }
4426
4427 #[gpui::test(iterations = 10)]
4428 async fn test_basic_chat(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4429 cx_a.foreground().forbid_parking();
4430
4431 // Connect to a server as 2 clients.
4432 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4433 let client_a = server.create_client(cx_a, "user_a").await;
4434 let client_b = server.create_client(cx_b, "user_b").await;
4435
4436 // Create an org that includes these 2 users.
4437 let db = &server.app_state.db;
4438 let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4439 db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4440 .await
4441 .unwrap();
4442 db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
4443 .await
4444 .unwrap();
4445
4446 // Create a channel that includes all the users.
4447 let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4448 db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4449 .await
4450 .unwrap();
4451 db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
4452 .await
4453 .unwrap();
4454 db.create_channel_message(
4455 channel_id,
4456 client_b.current_user_id(&cx_b),
4457 "hello A, it's B.",
4458 OffsetDateTime::now_utc(),
4459 1,
4460 )
4461 .await
4462 .unwrap();
4463
4464 let channels_a = cx_a
4465 .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4466 channels_a
4467 .condition(cx_a, |list, _| list.available_channels().is_some())
4468 .await;
4469 channels_a.read_with(cx_a, |list, _| {
4470 assert_eq!(
4471 list.available_channels().unwrap(),
4472 &[ChannelDetails {
4473 id: channel_id.to_proto(),
4474 name: "test-channel".to_string()
4475 }]
4476 )
4477 });
4478 let channel_a = channels_a.update(cx_a, |this, cx| {
4479 this.get_channel(channel_id.to_proto(), cx).unwrap()
4480 });
4481 channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
4482 channel_a
4483 .condition(&cx_a, |channel, _| {
4484 channel_messages(channel)
4485 == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4486 })
4487 .await;
4488
4489 let channels_b = cx_b
4490 .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
4491 channels_b
4492 .condition(cx_b, |list, _| list.available_channels().is_some())
4493 .await;
4494 channels_b.read_with(cx_b, |list, _| {
4495 assert_eq!(
4496 list.available_channels().unwrap(),
4497 &[ChannelDetails {
4498 id: channel_id.to_proto(),
4499 name: "test-channel".to_string()
4500 }]
4501 )
4502 });
4503
4504 let channel_b = channels_b.update(cx_b, |this, cx| {
4505 this.get_channel(channel_id.to_proto(), cx).unwrap()
4506 });
4507 channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
4508 channel_b
4509 .condition(&cx_b, |channel, _| {
4510 channel_messages(channel)
4511 == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4512 })
4513 .await;
4514
4515 channel_a
4516 .update(cx_a, |channel, cx| {
4517 channel
4518 .send_message("oh, hi B.".to_string(), cx)
4519 .unwrap()
4520 .detach();
4521 let task = channel.send_message("sup".to_string(), cx).unwrap();
4522 assert_eq!(
4523 channel_messages(channel),
4524 &[
4525 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4526 ("user_a".to_string(), "oh, hi B.".to_string(), true),
4527 ("user_a".to_string(), "sup".to_string(), true)
4528 ]
4529 );
4530 task
4531 })
4532 .await
4533 .unwrap();
4534
4535 channel_b
4536 .condition(&cx_b, |channel, _| {
4537 channel_messages(channel)
4538 == [
4539 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4540 ("user_a".to_string(), "oh, hi B.".to_string(), false),
4541 ("user_a".to_string(), "sup".to_string(), false),
4542 ]
4543 })
4544 .await;
4545
4546 assert_eq!(
4547 server
4548 .state()
4549 .await
4550 .channel(channel_id)
4551 .unwrap()
4552 .connection_ids
4553 .len(),
4554 2
4555 );
4556 cx_b.update(|_| drop(channel_b));
4557 server
4558 .condition(|state| state.channel(channel_id).unwrap().connection_ids.len() == 1)
4559 .await;
4560
4561 cx_a.update(|_| drop(channel_a));
4562 server
4563 .condition(|state| state.channel(channel_id).is_none())
4564 .await;
4565 }
4566
4567 #[gpui::test(iterations = 10)]
4568 async fn test_chat_message_validation(cx_a: &mut TestAppContext) {
4569 cx_a.foreground().forbid_parking();
4570
4571 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4572 let client_a = server.create_client(cx_a, "user_a").await;
4573
4574 let db = &server.app_state.db;
4575 let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4576 let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4577 db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4578 .await
4579 .unwrap();
4580 db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4581 .await
4582 .unwrap();
4583
4584 let channels_a = cx_a
4585 .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4586 channels_a
4587 .condition(cx_a, |list, _| list.available_channels().is_some())
4588 .await;
4589 let channel_a = channels_a.update(cx_a, |this, cx| {
4590 this.get_channel(channel_id.to_proto(), cx).unwrap()
4591 });
4592
4593 // Messages aren't allowed to be too long.
4594 channel_a
4595 .update(cx_a, |channel, cx| {
4596 let long_body = "this is long.\n".repeat(1024);
4597 channel.send_message(long_body, cx).unwrap()
4598 })
4599 .await
4600 .unwrap_err();
4601
4602 // Messages aren't allowed to be blank.
4603 channel_a.update(cx_a, |channel, cx| {
4604 channel.send_message(String::new(), cx).unwrap_err()
4605 });
4606
4607 // Leading and trailing whitespace are trimmed.
4608 channel_a
4609 .update(cx_a, |channel, cx| {
4610 channel
4611 .send_message("\n surrounded by whitespace \n".to_string(), cx)
4612 .unwrap()
4613 })
4614 .await
4615 .unwrap();
4616 assert_eq!(
4617 db.get_channel_messages(channel_id, 10, None)
4618 .await
4619 .unwrap()
4620 .iter()
4621 .map(|m| &m.body)
4622 .collect::<Vec<_>>(),
4623 &["surrounded by whitespace"]
4624 );
4625 }
4626
4627 #[gpui::test(iterations = 10)]
4628 async fn test_chat_reconnection(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
4629 cx_a.foreground().forbid_parking();
4630
4631 // Connect to a server as 2 clients.
4632 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4633 let client_a = server.create_client(cx_a, "user_a").await;
4634 let client_b = server.create_client(cx_b, "user_b").await;
4635 let mut status_b = client_b.status();
4636
4637 // Create an org that includes these 2 users.
4638 let db = &server.app_state.db;
4639 let org_id = db.create_org("Test Org", "test-org").await.unwrap();
4640 db.add_org_member(org_id, client_a.current_user_id(&cx_a), false)
4641 .await
4642 .unwrap();
4643 db.add_org_member(org_id, client_b.current_user_id(&cx_b), false)
4644 .await
4645 .unwrap();
4646
4647 // Create a channel that includes all the users.
4648 let channel_id = db.create_org_channel(org_id, "test-channel").await.unwrap();
4649 db.add_channel_member(channel_id, client_a.current_user_id(&cx_a), false)
4650 .await
4651 .unwrap();
4652 db.add_channel_member(channel_id, client_b.current_user_id(&cx_b), false)
4653 .await
4654 .unwrap();
4655 db.create_channel_message(
4656 channel_id,
4657 client_b.current_user_id(&cx_b),
4658 "hello A, it's B.",
4659 OffsetDateTime::now_utc(),
4660 2,
4661 )
4662 .await
4663 .unwrap();
4664
4665 let channels_a = cx_a
4666 .add_model(|cx| ChannelList::new(client_a.user_store.clone(), client_a.clone(), cx));
4667 channels_a
4668 .condition(cx_a, |list, _| list.available_channels().is_some())
4669 .await;
4670
4671 channels_a.read_with(cx_a, |list, _| {
4672 assert_eq!(
4673 list.available_channels().unwrap(),
4674 &[ChannelDetails {
4675 id: channel_id.to_proto(),
4676 name: "test-channel".to_string()
4677 }]
4678 )
4679 });
4680 let channel_a = channels_a.update(cx_a, |this, cx| {
4681 this.get_channel(channel_id.to_proto(), cx).unwrap()
4682 });
4683 channel_a.read_with(cx_a, |channel, _| assert!(channel.messages().is_empty()));
4684 channel_a
4685 .condition(&cx_a, |channel, _| {
4686 channel_messages(channel)
4687 == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4688 })
4689 .await;
4690
4691 let channels_b = cx_b
4692 .add_model(|cx| ChannelList::new(client_b.user_store.clone(), client_b.clone(), cx));
4693 channels_b
4694 .condition(cx_b, |list, _| list.available_channels().is_some())
4695 .await;
4696 channels_b.read_with(cx_b, |list, _| {
4697 assert_eq!(
4698 list.available_channels().unwrap(),
4699 &[ChannelDetails {
4700 id: channel_id.to_proto(),
4701 name: "test-channel".to_string()
4702 }]
4703 )
4704 });
4705
4706 let channel_b = channels_b.update(cx_b, |this, cx| {
4707 this.get_channel(channel_id.to_proto(), cx).unwrap()
4708 });
4709 channel_b.read_with(cx_b, |channel, _| assert!(channel.messages().is_empty()));
4710 channel_b
4711 .condition(&cx_b, |channel, _| {
4712 channel_messages(channel)
4713 == [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4714 })
4715 .await;
4716
4717 // Disconnect client B, ensuring we can still access its cached channel data.
4718 server.forbid_connections();
4719 server.disconnect_client(client_b.current_user_id(&cx_b));
4720 cx_b.foreground().advance_clock(rpc::RECEIVE_TIMEOUT);
4721 while !matches!(
4722 status_b.next().await,
4723 Some(client::Status::ReconnectionError { .. })
4724 ) {}
4725
4726 channels_b.read_with(cx_b, |channels, _| {
4727 assert_eq!(
4728 channels.available_channels().unwrap(),
4729 [ChannelDetails {
4730 id: channel_id.to_proto(),
4731 name: "test-channel".to_string()
4732 }]
4733 )
4734 });
4735 channel_b.read_with(cx_b, |channel, _| {
4736 assert_eq!(
4737 channel_messages(channel),
4738 [("user_b".to_string(), "hello A, it's B.".to_string(), false)]
4739 )
4740 });
4741
4742 // Send a message from client B while it is disconnected.
4743 channel_b
4744 .update(cx_b, |channel, cx| {
4745 let task = channel
4746 .send_message("can you see this?".to_string(), cx)
4747 .unwrap();
4748 assert_eq!(
4749 channel_messages(channel),
4750 &[
4751 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4752 ("user_b".to_string(), "can you see this?".to_string(), true)
4753 ]
4754 );
4755 task
4756 })
4757 .await
4758 .unwrap_err();
4759
4760 // Send a message from client A while B is disconnected.
4761 channel_a
4762 .update(cx_a, |channel, cx| {
4763 channel
4764 .send_message("oh, hi B.".to_string(), cx)
4765 .unwrap()
4766 .detach();
4767 let task = channel.send_message("sup".to_string(), cx).unwrap();
4768 assert_eq!(
4769 channel_messages(channel),
4770 &[
4771 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4772 ("user_a".to_string(), "oh, hi B.".to_string(), true),
4773 ("user_a".to_string(), "sup".to_string(), true)
4774 ]
4775 );
4776 task
4777 })
4778 .await
4779 .unwrap();
4780
4781 // Give client B a chance to reconnect.
4782 server.allow_connections();
4783 cx_b.foreground().advance_clock(Duration::from_secs(10));
4784
4785 // Verify that B sees the new messages upon reconnection, as well as the message client B
4786 // sent while offline.
4787 channel_b
4788 .condition(&cx_b, |channel, _| {
4789 channel_messages(channel)
4790 == [
4791 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4792 ("user_a".to_string(), "oh, hi B.".to_string(), false),
4793 ("user_a".to_string(), "sup".to_string(), false),
4794 ("user_b".to_string(), "can you see this?".to_string(), false),
4795 ]
4796 })
4797 .await;
4798
4799 // Ensure client A and B can communicate normally after reconnection.
4800 channel_a
4801 .update(cx_a, |channel, cx| {
4802 channel.send_message("you online?".to_string(), cx).unwrap()
4803 })
4804 .await
4805 .unwrap();
4806 channel_b
4807 .condition(&cx_b, |channel, _| {
4808 channel_messages(channel)
4809 == [
4810 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4811 ("user_a".to_string(), "oh, hi B.".to_string(), false),
4812 ("user_a".to_string(), "sup".to_string(), false),
4813 ("user_b".to_string(), "can you see this?".to_string(), false),
4814 ("user_a".to_string(), "you online?".to_string(), false),
4815 ]
4816 })
4817 .await;
4818
4819 channel_b
4820 .update(cx_b, |channel, cx| {
4821 channel.send_message("yep".to_string(), cx).unwrap()
4822 })
4823 .await
4824 .unwrap();
4825 channel_a
4826 .condition(&cx_a, |channel, _| {
4827 channel_messages(channel)
4828 == [
4829 ("user_b".to_string(), "hello A, it's B.".to_string(), false),
4830 ("user_a".to_string(), "oh, hi B.".to_string(), false),
4831 ("user_a".to_string(), "sup".to_string(), false),
4832 ("user_b".to_string(), "can you see this?".to_string(), false),
4833 ("user_a".to_string(), "you online?".to_string(), false),
4834 ("user_b".to_string(), "yep".to_string(), false),
4835 ]
4836 })
4837 .await;
4838 }
4839
4840 #[gpui::test(iterations = 10)]
4841 async fn test_contacts(
4842 deterministic: Arc<Deterministic>,
4843 cx_a: &mut TestAppContext,
4844 cx_b: &mut TestAppContext,
4845 cx_c: &mut TestAppContext,
4846 ) {
4847 cx_a.foreground().forbid_parking();
4848
4849 // Connect to a server as 3 clients.
4850 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
4851 let mut client_a = server.create_client(cx_a, "user_a").await;
4852 let mut client_b = server.create_client(cx_b, "user_b").await;
4853 let client_c = server.create_client(cx_c, "user_c").await;
4854 server
4855 .make_contacts(vec![
4856 (&client_a, cx_a),
4857 (&client_b, cx_b),
4858 (&client_c, cx_c),
4859 ])
4860 .await;
4861
4862 deterministic.run_until_parked();
4863 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
4864 client.user_store.read_with(*cx, |store, _| {
4865 assert_eq!(
4866 contacts(store),
4867 [
4868 ("user_a", true, vec![]),
4869 ("user_b", true, vec![]),
4870 ("user_c", true, vec![])
4871 ]
4872 )
4873 });
4874 }
4875
4876 // Share a project as client A.
4877 let fs = FakeFs::new(cx_a.background());
4878 fs.create_dir(Path::new("/a")).await.unwrap();
4879 let (project_a, _) = client_a.build_local_project(fs, "/a", cx_a).await;
4880
4881 deterministic.run_until_parked();
4882 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
4883 client.user_store.read_with(*cx, |store, _| {
4884 assert_eq!(
4885 contacts(store),
4886 [
4887 ("user_a", true, vec![("a", vec![])]),
4888 ("user_b", true, vec![]),
4889 ("user_c", true, vec![])
4890 ]
4891 )
4892 });
4893 }
4894
4895 let _project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
4896
4897 deterministic.run_until_parked();
4898 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
4899 client.user_store.read_with(*cx, |store, _| {
4900 assert_eq!(
4901 contacts(store),
4902 [
4903 ("user_a", true, vec![("a", vec!["user_b"])]),
4904 ("user_b", true, vec![]),
4905 ("user_c", true, vec![])
4906 ]
4907 )
4908 });
4909 }
4910
4911 // Add a local project as client B
4912 let fs = FakeFs::new(cx_b.background());
4913 fs.create_dir(Path::new("/b")).await.unwrap();
4914 let (_project_b, _) = client_b.build_local_project(fs, "/b", cx_a).await;
4915
4916 deterministic.run_until_parked();
4917 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
4918 client.user_store.read_with(*cx, |store, _| {
4919 assert_eq!(
4920 contacts(store),
4921 [
4922 ("user_a", true, vec![("a", vec!["user_b"])]),
4923 ("user_b", true, vec![("b", vec![])]),
4924 ("user_c", true, vec![])
4925 ]
4926 )
4927 });
4928 }
4929
4930 project_a
4931 .condition(&cx_a, |project, _| {
4932 project.collaborators().contains_key(&client_b.peer_id)
4933 })
4934 .await;
4935
4936 client_a.project.take();
4937 cx_a.update(move |_| drop(project_a));
4938 deterministic.run_until_parked();
4939 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
4940 client.user_store.read_with(*cx, |store, _| {
4941 assert_eq!(
4942 contacts(store),
4943 [
4944 ("user_a", true, vec![]),
4945 ("user_b", true, vec![("b", vec![])]),
4946 ("user_c", true, vec![])
4947 ]
4948 )
4949 });
4950 }
4951
4952 server.disconnect_client(client_c.current_user_id(cx_c));
4953 server.forbid_connections();
4954 deterministic.advance_clock(rpc::RECEIVE_TIMEOUT);
4955 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b)] {
4956 client.user_store.read_with(*cx, |store, _| {
4957 assert_eq!(
4958 contacts(store),
4959 [
4960 ("user_a", true, vec![]),
4961 ("user_b", true, vec![("b", vec![])]),
4962 ("user_c", false, vec![])
4963 ]
4964 )
4965 });
4966 }
4967 client_c
4968 .user_store
4969 .read_with(cx_c, |store, _| assert_eq!(contacts(store), []));
4970
4971 server.allow_connections();
4972 client_c
4973 .authenticate_and_connect(false, &cx_c.to_async())
4974 .await
4975 .unwrap();
4976
4977 deterministic.run_until_parked();
4978 for (client, cx) in [(&client_a, &cx_a), (&client_b, &cx_b), (&client_c, &cx_c)] {
4979 client.user_store.read_with(*cx, |store, _| {
4980 assert_eq!(
4981 contacts(store),
4982 [
4983 ("user_a", true, vec![]),
4984 ("user_b", true, vec![("b", vec![])]),
4985 ("user_c", true, vec![])
4986 ]
4987 )
4988 });
4989 }
4990
4991 fn contacts(user_store: &UserStore) -> Vec<(&str, bool, Vec<(&str, Vec<&str>)>)> {
4992 user_store
4993 .contacts()
4994 .iter()
4995 .map(|contact| {
4996 let projects = contact
4997 .projects
4998 .iter()
4999 .map(|p| {
5000 (
5001 p.worktree_root_names[0].as_str(),
5002 p.guests.iter().map(|p| p.github_login.as_str()).collect(),
5003 )
5004 })
5005 .collect();
5006 (contact.user.github_login.as_str(), contact.online, projects)
5007 })
5008 .collect()
5009 }
5010 }
5011
5012 #[gpui::test(iterations = 10)]
5013 async fn test_contact_requests(
5014 executor: Arc<Deterministic>,
5015 cx_a: &mut TestAppContext,
5016 cx_a2: &mut TestAppContext,
5017 cx_b: &mut TestAppContext,
5018 cx_b2: &mut TestAppContext,
5019 cx_c: &mut TestAppContext,
5020 cx_c2: &mut TestAppContext,
5021 ) {
5022 cx_a.foreground().forbid_parking();
5023
5024 // Connect to a server as 3 clients.
5025 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5026 let client_a = server.create_client(cx_a, "user_a").await;
5027 let client_a2 = server.create_client(cx_a2, "user_a").await;
5028 let client_b = server.create_client(cx_b, "user_b").await;
5029 let client_b2 = server.create_client(cx_b2, "user_b").await;
5030 let client_c = server.create_client(cx_c, "user_c").await;
5031 let client_c2 = server.create_client(cx_c2, "user_c").await;
5032
5033 assert_eq!(client_a.user_id().unwrap(), client_a2.user_id().unwrap());
5034 assert_eq!(client_b.user_id().unwrap(), client_b2.user_id().unwrap());
5035 assert_eq!(client_c.user_id().unwrap(), client_c2.user_id().unwrap());
5036
5037 // User A and User C request that user B become their contact.
5038 client_a
5039 .user_store
5040 .update(cx_a, |store, cx| {
5041 store.request_contact(client_b.user_id().unwrap(), cx)
5042 })
5043 .await
5044 .unwrap();
5045 client_c
5046 .user_store
5047 .update(cx_c, |store, cx| {
5048 store.request_contact(client_b.user_id().unwrap(), cx)
5049 })
5050 .await
5051 .unwrap();
5052 executor.run_until_parked();
5053
5054 // All users see the pending request appear in all their clients.
5055 assert_eq!(
5056 client_a.summarize_contacts(&cx_a).outgoing_requests,
5057 &["user_b"]
5058 );
5059 assert_eq!(
5060 client_a2.summarize_contacts(&cx_a2).outgoing_requests,
5061 &["user_b"]
5062 );
5063 assert_eq!(
5064 client_b.summarize_contacts(&cx_b).incoming_requests,
5065 &["user_a", "user_c"]
5066 );
5067 assert_eq!(
5068 client_b2.summarize_contacts(&cx_b2).incoming_requests,
5069 &["user_a", "user_c"]
5070 );
5071 assert_eq!(
5072 client_c.summarize_contacts(&cx_c).outgoing_requests,
5073 &["user_b"]
5074 );
5075 assert_eq!(
5076 client_c2.summarize_contacts(&cx_c2).outgoing_requests,
5077 &["user_b"]
5078 );
5079
5080 // Contact requests are present upon connecting (tested here via disconnect/reconnect)
5081 disconnect_and_reconnect(&client_a, cx_a).await;
5082 disconnect_and_reconnect(&client_b, cx_b).await;
5083 disconnect_and_reconnect(&client_c, cx_c).await;
5084 executor.run_until_parked();
5085 assert_eq!(
5086 client_a.summarize_contacts(&cx_a).outgoing_requests,
5087 &["user_b"]
5088 );
5089 assert_eq!(
5090 client_b.summarize_contacts(&cx_b).incoming_requests,
5091 &["user_a", "user_c"]
5092 );
5093 assert_eq!(
5094 client_c.summarize_contacts(&cx_c).outgoing_requests,
5095 &["user_b"]
5096 );
5097
5098 // User B accepts the request from user A.
5099 client_b
5100 .user_store
5101 .update(cx_b, |store, cx| {
5102 store.respond_to_contact_request(client_a.user_id().unwrap(), true, cx)
5103 })
5104 .await
5105 .unwrap();
5106
5107 executor.run_until_parked();
5108
5109 // User B sees user A as their contact now in all client, and the incoming request from them is removed.
5110 let contacts_b = client_b.summarize_contacts(&cx_b);
5111 assert_eq!(contacts_b.current, &["user_a", "user_b"]);
5112 assert_eq!(contacts_b.incoming_requests, &["user_c"]);
5113 let contacts_b2 = client_b2.summarize_contacts(&cx_b2);
5114 assert_eq!(contacts_b2.current, &["user_a", "user_b"]);
5115 assert_eq!(contacts_b2.incoming_requests, &["user_c"]);
5116
5117 // User A sees user B as their contact now in all clients, and the outgoing request to them is removed.
5118 let contacts_a = client_a.summarize_contacts(&cx_a);
5119 assert_eq!(contacts_a.current, &["user_a", "user_b"]);
5120 assert!(contacts_a.outgoing_requests.is_empty());
5121 let contacts_a2 = client_a2.summarize_contacts(&cx_a2);
5122 assert_eq!(contacts_a2.current, &["user_a", "user_b"]);
5123 assert!(contacts_a2.outgoing_requests.is_empty());
5124
5125 // Contacts are present upon connecting (tested here via disconnect/reconnect)
5126 disconnect_and_reconnect(&client_a, cx_a).await;
5127 disconnect_and_reconnect(&client_b, cx_b).await;
5128 disconnect_and_reconnect(&client_c, cx_c).await;
5129 executor.run_until_parked();
5130 assert_eq!(
5131 client_a.summarize_contacts(&cx_a).current,
5132 &["user_a", "user_b"]
5133 );
5134 assert_eq!(
5135 client_b.summarize_contacts(&cx_b).current,
5136 &["user_a", "user_b"]
5137 );
5138 assert_eq!(
5139 client_b.summarize_contacts(&cx_b).incoming_requests,
5140 &["user_c"]
5141 );
5142 assert_eq!(client_c.summarize_contacts(&cx_c).current, &["user_c"]);
5143 assert_eq!(
5144 client_c.summarize_contacts(&cx_c).outgoing_requests,
5145 &["user_b"]
5146 );
5147
5148 // User B rejects the request from user C.
5149 client_b
5150 .user_store
5151 .update(cx_b, |store, cx| {
5152 store.respond_to_contact_request(client_c.user_id().unwrap(), false, cx)
5153 })
5154 .await
5155 .unwrap();
5156
5157 executor.run_until_parked();
5158
5159 // User B doesn't see user C as their contact, and the incoming request from them is removed.
5160 let contacts_b = client_b.summarize_contacts(&cx_b);
5161 assert_eq!(contacts_b.current, &["user_a", "user_b"]);
5162 assert!(contacts_b.incoming_requests.is_empty());
5163 let contacts_b2 = client_b2.summarize_contacts(&cx_b2);
5164 assert_eq!(contacts_b2.current, &["user_a", "user_b"]);
5165 assert!(contacts_b2.incoming_requests.is_empty());
5166
5167 // User C doesn't see user B as their contact, and the outgoing request to them is removed.
5168 let contacts_c = client_c.summarize_contacts(&cx_c);
5169 assert_eq!(contacts_c.current, &["user_c"]);
5170 assert!(contacts_c.outgoing_requests.is_empty());
5171 let contacts_c2 = client_c2.summarize_contacts(&cx_c2);
5172 assert_eq!(contacts_c2.current, &["user_c"]);
5173 assert!(contacts_c2.outgoing_requests.is_empty());
5174
5175 // Incoming/outgoing requests are not present upon connecting (tested here via disconnect/reconnect)
5176 disconnect_and_reconnect(&client_a, cx_a).await;
5177 disconnect_and_reconnect(&client_b, cx_b).await;
5178 disconnect_and_reconnect(&client_c, cx_c).await;
5179 executor.run_until_parked();
5180 assert_eq!(
5181 client_a.summarize_contacts(&cx_a).current,
5182 &["user_a", "user_b"]
5183 );
5184 assert_eq!(
5185 client_b.summarize_contacts(&cx_b).current,
5186 &["user_a", "user_b"]
5187 );
5188 assert!(client_b
5189 .summarize_contacts(&cx_b)
5190 .incoming_requests
5191 .is_empty());
5192 assert_eq!(client_c.summarize_contacts(&cx_c).current, &["user_c"]);
5193 assert!(client_c
5194 .summarize_contacts(&cx_c)
5195 .outgoing_requests
5196 .is_empty());
5197
5198 async fn disconnect_and_reconnect(client: &TestClient, cx: &mut TestAppContext) {
5199 client.disconnect(&cx.to_async()).unwrap();
5200 client.clear_contacts(cx).await;
5201 client
5202 .authenticate_and_connect(false, &cx.to_async())
5203 .await
5204 .unwrap();
5205 }
5206 }
5207
5208 #[gpui::test(iterations = 10)]
5209 async fn test_following(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
5210 cx_a.foreground().forbid_parking();
5211 let fs = FakeFs::new(cx_a.background());
5212
5213 // 2 clients connect to a server.
5214 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5215 let mut client_a = server.create_client(cx_a, "user_a").await;
5216 let mut client_b = server.create_client(cx_b, "user_b").await;
5217 server
5218 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
5219 .await;
5220 cx_a.update(editor::init);
5221 cx_b.update(editor::init);
5222
5223 // Client A shares a project.
5224 fs.insert_tree(
5225 "/a",
5226 json!({
5227 "1.txt": "one",
5228 "2.txt": "two",
5229 "3.txt": "three",
5230 }),
5231 )
5232 .await;
5233 let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
5234
5235 // Client B joins the project.
5236 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
5237
5238 // Client A opens some editors.
5239 let workspace_a = client_a.build_workspace(&project_a, cx_a);
5240 let pane_a = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
5241 let editor_a1 = workspace_a
5242 .update(cx_a, |workspace, cx| {
5243 workspace.open_path((worktree_id, "1.txt"), true, cx)
5244 })
5245 .await
5246 .unwrap()
5247 .downcast::<Editor>()
5248 .unwrap();
5249 let editor_a2 = workspace_a
5250 .update(cx_a, |workspace, cx| {
5251 workspace.open_path((worktree_id, "2.txt"), true, cx)
5252 })
5253 .await
5254 .unwrap()
5255 .downcast::<Editor>()
5256 .unwrap();
5257
5258 // Client B opens an editor.
5259 let workspace_b = client_b.build_workspace(&project_b, cx_b);
5260 let editor_b1 = workspace_b
5261 .update(cx_b, |workspace, cx| {
5262 workspace.open_path((worktree_id, "1.txt"), true, cx)
5263 })
5264 .await
5265 .unwrap()
5266 .downcast::<Editor>()
5267 .unwrap();
5268
5269 let client_a_id = project_b.read_with(cx_b, |project, _| {
5270 project.collaborators().values().next().unwrap().peer_id
5271 });
5272 let client_b_id = project_a.read_with(cx_a, |project, _| {
5273 project.collaborators().values().next().unwrap().peer_id
5274 });
5275
5276 // When client B starts following client A, all visible view states are replicated to client B.
5277 editor_a1.update(cx_a, |editor, cx| editor.select_ranges([0..1], None, cx));
5278 editor_a2.update(cx_a, |editor, cx| editor.select_ranges([2..3], None, cx));
5279 workspace_b
5280 .update(cx_b, |workspace, cx| {
5281 workspace
5282 .toggle_follow(&ToggleFollow(client_a_id), cx)
5283 .unwrap()
5284 })
5285 .await
5286 .unwrap();
5287
5288 let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
5289 workspace
5290 .active_item(cx)
5291 .unwrap()
5292 .downcast::<Editor>()
5293 .unwrap()
5294 });
5295 assert!(cx_b.read(|cx| editor_b2.is_focused(cx)));
5296 assert_eq!(
5297 editor_b2.read_with(cx_b, |editor, cx| editor.project_path(cx)),
5298 Some((worktree_id, "2.txt").into())
5299 );
5300 assert_eq!(
5301 editor_b2.read_with(cx_b, |editor, cx| editor.selected_ranges(cx)),
5302 vec![2..3]
5303 );
5304 assert_eq!(
5305 editor_b1.read_with(cx_b, |editor, cx| editor.selected_ranges(cx)),
5306 vec![0..1]
5307 );
5308
5309 // When client A activates a different editor, client B does so as well.
5310 workspace_a.update(cx_a, |workspace, cx| {
5311 workspace.activate_item(&editor_a1, cx)
5312 });
5313 workspace_b
5314 .condition(cx_b, |workspace, cx| {
5315 workspace.active_item(cx).unwrap().id() == editor_b1.id()
5316 })
5317 .await;
5318
5319 // When client A navigates back and forth, client B does so as well.
5320 workspace_a
5321 .update(cx_a, |workspace, cx| {
5322 workspace::Pane::go_back(workspace, None, cx)
5323 })
5324 .await;
5325 workspace_b
5326 .condition(cx_b, |workspace, cx| {
5327 workspace.active_item(cx).unwrap().id() == editor_b2.id()
5328 })
5329 .await;
5330
5331 workspace_a
5332 .update(cx_a, |workspace, cx| {
5333 workspace::Pane::go_forward(workspace, None, cx)
5334 })
5335 .await;
5336 workspace_b
5337 .condition(cx_b, |workspace, cx| {
5338 workspace.active_item(cx).unwrap().id() == editor_b1.id()
5339 })
5340 .await;
5341
5342 // Changes to client A's editor are reflected on client B.
5343 editor_a1.update(cx_a, |editor, cx| {
5344 editor.select_ranges([1..1, 2..2], None, cx);
5345 });
5346 editor_b1
5347 .condition(cx_b, |editor, cx| {
5348 editor.selected_ranges(cx) == vec![1..1, 2..2]
5349 })
5350 .await;
5351
5352 editor_a1.update(cx_a, |editor, cx| editor.set_text("TWO", cx));
5353 editor_b1
5354 .condition(cx_b, |editor, cx| editor.text(cx) == "TWO")
5355 .await;
5356
5357 editor_a1.update(cx_a, |editor, cx| {
5358 editor.select_ranges([3..3], None, cx);
5359 editor.set_scroll_position(vec2f(0., 100.), cx);
5360 });
5361 editor_b1
5362 .condition(cx_b, |editor, cx| editor.selected_ranges(cx) == vec![3..3])
5363 .await;
5364
5365 // After unfollowing, client B stops receiving updates from client A.
5366 workspace_b.update(cx_b, |workspace, cx| {
5367 workspace.unfollow(&workspace.active_pane().clone(), cx)
5368 });
5369 workspace_a.update(cx_a, |workspace, cx| {
5370 workspace.activate_item(&editor_a2, cx)
5371 });
5372 cx_a.foreground().run_until_parked();
5373 assert_eq!(
5374 workspace_b.read_with(cx_b, |workspace, cx| workspace
5375 .active_item(cx)
5376 .unwrap()
5377 .id()),
5378 editor_b1.id()
5379 );
5380
5381 // Client A starts following client B.
5382 workspace_a
5383 .update(cx_a, |workspace, cx| {
5384 workspace
5385 .toggle_follow(&ToggleFollow(client_b_id), cx)
5386 .unwrap()
5387 })
5388 .await
5389 .unwrap();
5390 assert_eq!(
5391 workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
5392 Some(client_b_id)
5393 );
5394 assert_eq!(
5395 workspace_a.read_with(cx_a, |workspace, cx| workspace
5396 .active_item(cx)
5397 .unwrap()
5398 .id()),
5399 editor_a1.id()
5400 );
5401
5402 // Following interrupts when client B disconnects.
5403 client_b.disconnect(&cx_b.to_async()).unwrap();
5404 cx_a.foreground().run_until_parked();
5405 assert_eq!(
5406 workspace_a.read_with(cx_a, |workspace, _| workspace.leader_for_pane(&pane_a)),
5407 None
5408 );
5409 }
5410
5411 #[gpui::test(iterations = 10)]
5412 async fn test_peers_following_each_other(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
5413 cx_a.foreground().forbid_parking();
5414 let fs = FakeFs::new(cx_a.background());
5415
5416 // 2 clients connect to a server.
5417 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5418 let mut client_a = server.create_client(cx_a, "user_a").await;
5419 let mut client_b = server.create_client(cx_b, "user_b").await;
5420 server
5421 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
5422 .await;
5423 cx_a.update(editor::init);
5424 cx_b.update(editor::init);
5425
5426 // Client A shares a project.
5427 fs.insert_tree(
5428 "/a",
5429 json!({
5430 "1.txt": "one",
5431 "2.txt": "two",
5432 "3.txt": "three",
5433 "4.txt": "four",
5434 }),
5435 )
5436 .await;
5437 let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
5438
5439 // Client B joins the project.
5440 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
5441
5442 // Client A opens some editors.
5443 let workspace_a = client_a.build_workspace(&project_a, cx_a);
5444 let pane_a1 = workspace_a.read_with(cx_a, |workspace, _| workspace.active_pane().clone());
5445 let _editor_a1 = workspace_a
5446 .update(cx_a, |workspace, cx| {
5447 workspace.open_path((worktree_id, "1.txt"), true, cx)
5448 })
5449 .await
5450 .unwrap()
5451 .downcast::<Editor>()
5452 .unwrap();
5453
5454 // Client B opens an editor.
5455 let workspace_b = client_b.build_workspace(&project_b, cx_b);
5456 let pane_b1 = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
5457 let _editor_b1 = workspace_b
5458 .update(cx_b, |workspace, cx| {
5459 workspace.open_path((worktree_id, "2.txt"), true, cx)
5460 })
5461 .await
5462 .unwrap()
5463 .downcast::<Editor>()
5464 .unwrap();
5465
5466 // Clients A and B follow each other in split panes
5467 workspace_a
5468 .update(cx_a, |workspace, cx| {
5469 workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
5470 assert_ne!(*workspace.active_pane(), pane_a1);
5471 let leader_id = *project_a.read(cx).collaborators().keys().next().unwrap();
5472 workspace
5473 .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
5474 .unwrap()
5475 })
5476 .await
5477 .unwrap();
5478 workspace_b
5479 .update(cx_b, |workspace, cx| {
5480 workspace.split_pane(workspace.active_pane().clone(), SplitDirection::Right, cx);
5481 assert_ne!(*workspace.active_pane(), pane_b1);
5482 let leader_id = *project_b.read(cx).collaborators().keys().next().unwrap();
5483 workspace
5484 .toggle_follow(&workspace::ToggleFollow(leader_id), cx)
5485 .unwrap()
5486 })
5487 .await
5488 .unwrap();
5489
5490 workspace_a
5491 .update(cx_a, |workspace, cx| {
5492 workspace.activate_next_pane(cx);
5493 assert_eq!(*workspace.active_pane(), pane_a1);
5494 workspace.open_path((worktree_id, "3.txt"), true, cx)
5495 })
5496 .await
5497 .unwrap();
5498 workspace_b
5499 .update(cx_b, |workspace, cx| {
5500 workspace.activate_next_pane(cx);
5501 assert_eq!(*workspace.active_pane(), pane_b1);
5502 workspace.open_path((worktree_id, "4.txt"), true, cx)
5503 })
5504 .await
5505 .unwrap();
5506 cx_a.foreground().run_until_parked();
5507
5508 // Ensure leader updates don't change the active pane of followers
5509 workspace_a.read_with(cx_a, |workspace, _| {
5510 assert_eq!(*workspace.active_pane(), pane_a1);
5511 });
5512 workspace_b.read_with(cx_b, |workspace, _| {
5513 assert_eq!(*workspace.active_pane(), pane_b1);
5514 });
5515
5516 // Ensure peers following each other doesn't cause an infinite loop.
5517 assert_eq!(
5518 workspace_a.read_with(cx_a, |workspace, cx| workspace
5519 .active_item(cx)
5520 .unwrap()
5521 .project_path(cx)),
5522 Some((worktree_id, "3.txt").into())
5523 );
5524 workspace_a.update(cx_a, |workspace, cx| {
5525 assert_eq!(
5526 workspace.active_item(cx).unwrap().project_path(cx),
5527 Some((worktree_id, "3.txt").into())
5528 );
5529 workspace.activate_next_pane(cx);
5530 assert_eq!(
5531 workspace.active_item(cx).unwrap().project_path(cx),
5532 Some((worktree_id, "4.txt").into())
5533 );
5534 });
5535 workspace_b.update(cx_b, |workspace, cx| {
5536 assert_eq!(
5537 workspace.active_item(cx).unwrap().project_path(cx),
5538 Some((worktree_id, "4.txt").into())
5539 );
5540 workspace.activate_next_pane(cx);
5541 assert_eq!(
5542 workspace.active_item(cx).unwrap().project_path(cx),
5543 Some((worktree_id, "3.txt").into())
5544 );
5545 });
5546 }
5547
5548 #[gpui::test(iterations = 10)]
5549 async fn test_auto_unfollowing(cx_a: &mut TestAppContext, cx_b: &mut TestAppContext) {
5550 cx_a.foreground().forbid_parking();
5551 let fs = FakeFs::new(cx_a.background());
5552
5553 // 2 clients connect to a server.
5554 let mut server = TestServer::start(cx_a.foreground(), cx_a.background()).await;
5555 let mut client_a = server.create_client(cx_a, "user_a").await;
5556 let mut client_b = server.create_client(cx_b, "user_b").await;
5557 server
5558 .make_contacts(vec![(&client_a, cx_a), (&client_b, cx_b)])
5559 .await;
5560 cx_a.update(editor::init);
5561 cx_b.update(editor::init);
5562
5563 // Client A shares a project.
5564 fs.insert_tree(
5565 "/a",
5566 json!({
5567 "1.txt": "one",
5568 "2.txt": "two",
5569 "3.txt": "three",
5570 }),
5571 )
5572 .await;
5573 let (project_a, worktree_id) = client_a.build_local_project(fs.clone(), "/a", cx_a).await;
5574
5575 // Client B joins the project.
5576 let project_b = client_b.build_remote_project(&project_a, cx_a, cx_b).await;
5577
5578 // Client A opens some editors.
5579 let workspace_a = client_a.build_workspace(&project_a, cx_a);
5580 let _editor_a1 = workspace_a
5581 .update(cx_a, |workspace, cx| {
5582 workspace.open_path((worktree_id, "1.txt"), true, cx)
5583 })
5584 .await
5585 .unwrap()
5586 .downcast::<Editor>()
5587 .unwrap();
5588
5589 // Client B starts following client A.
5590 let workspace_b = client_b.build_workspace(&project_b, cx_b);
5591 let pane_b = workspace_b.read_with(cx_b, |workspace, _| workspace.active_pane().clone());
5592 let leader_id = project_b.read_with(cx_b, |project, _| {
5593 project.collaborators().values().next().unwrap().peer_id
5594 });
5595 workspace_b
5596 .update(cx_b, |workspace, cx| {
5597 workspace
5598 .toggle_follow(&ToggleFollow(leader_id), cx)
5599 .unwrap()
5600 })
5601 .await
5602 .unwrap();
5603 assert_eq!(
5604 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5605 Some(leader_id)
5606 );
5607 let editor_b2 = workspace_b.read_with(cx_b, |workspace, cx| {
5608 workspace
5609 .active_item(cx)
5610 .unwrap()
5611 .downcast::<Editor>()
5612 .unwrap()
5613 });
5614
5615 // When client B moves, it automatically stops following client A.
5616 editor_b2.update(cx_b, |editor, cx| editor.move_right(&editor::MoveRight, cx));
5617 assert_eq!(
5618 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5619 None
5620 );
5621
5622 workspace_b
5623 .update(cx_b, |workspace, cx| {
5624 workspace
5625 .toggle_follow(&ToggleFollow(leader_id), cx)
5626 .unwrap()
5627 })
5628 .await
5629 .unwrap();
5630 assert_eq!(
5631 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5632 Some(leader_id)
5633 );
5634
5635 // When client B edits, it automatically stops following client A.
5636 editor_b2.update(cx_b, |editor, cx| editor.insert("X", cx));
5637 assert_eq!(
5638 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5639 None
5640 );
5641
5642 workspace_b
5643 .update(cx_b, |workspace, cx| {
5644 workspace
5645 .toggle_follow(&ToggleFollow(leader_id), cx)
5646 .unwrap()
5647 })
5648 .await
5649 .unwrap();
5650 assert_eq!(
5651 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5652 Some(leader_id)
5653 );
5654
5655 // When client B scrolls, it automatically stops following client A.
5656 editor_b2.update(cx_b, |editor, cx| {
5657 editor.set_scroll_position(vec2f(0., 3.), cx)
5658 });
5659 assert_eq!(
5660 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5661 None
5662 );
5663
5664 workspace_b
5665 .update(cx_b, |workspace, cx| {
5666 workspace
5667 .toggle_follow(&ToggleFollow(leader_id), cx)
5668 .unwrap()
5669 })
5670 .await
5671 .unwrap();
5672 assert_eq!(
5673 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5674 Some(leader_id)
5675 );
5676
5677 // When client B activates a different pane, it continues following client A in the original pane.
5678 workspace_b.update(cx_b, |workspace, cx| {
5679 workspace.split_pane(pane_b.clone(), SplitDirection::Right, cx)
5680 });
5681 assert_eq!(
5682 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5683 Some(leader_id)
5684 );
5685
5686 workspace_b.update(cx_b, |workspace, cx| workspace.activate_next_pane(cx));
5687 assert_eq!(
5688 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5689 Some(leader_id)
5690 );
5691
5692 // When client B activates a different item in the original pane, it automatically stops following client A.
5693 workspace_b
5694 .update(cx_b, |workspace, cx| {
5695 workspace.open_path((worktree_id, "2.txt"), true, cx)
5696 })
5697 .await
5698 .unwrap();
5699 assert_eq!(
5700 workspace_b.read_with(cx_b, |workspace, _| workspace.leader_for_pane(&pane_b)),
5701 None
5702 );
5703 }
5704
5705 #[gpui::test(iterations = 100)]
5706 async fn test_random_collaboration(
5707 cx: &mut TestAppContext,
5708 deterministic: Arc<Deterministic>,
5709 rng: StdRng,
5710 ) {
5711 cx.foreground().forbid_parking();
5712 let max_peers = env::var("MAX_PEERS")
5713 .map(|i| i.parse().expect("invalid `MAX_PEERS` variable"))
5714 .unwrap_or(5);
5715 assert!(max_peers <= 5);
5716
5717 let max_operations = env::var("OPERATIONS")
5718 .map(|i| i.parse().expect("invalid `OPERATIONS` variable"))
5719 .unwrap_or(10);
5720
5721 let rng = Arc::new(Mutex::new(rng));
5722
5723 let guest_lang_registry = Arc::new(LanguageRegistry::test());
5724 let host_language_registry = Arc::new(LanguageRegistry::test());
5725
5726 let fs = FakeFs::new(cx.background());
5727 fs.insert_tree("/_collab", json!({"init": ""})).await;
5728
5729 let mut server = TestServer::start(cx.foreground(), cx.background()).await;
5730 let db = server.app_state.db.clone();
5731 let host_user_id = db.create_user("host", false).await.unwrap();
5732 for username in ["guest-1", "guest-2", "guest-3", "guest-4"] {
5733 let guest_user_id = db.create_user(username, false).await.unwrap();
5734 server
5735 .app_state
5736 .db
5737 .send_contact_request(guest_user_id, host_user_id)
5738 .await
5739 .unwrap();
5740 server
5741 .app_state
5742 .db
5743 .respond_to_contact_request(host_user_id, guest_user_id, true)
5744 .await
5745 .unwrap();
5746 }
5747
5748 let mut clients = Vec::new();
5749 let mut user_ids = Vec::new();
5750 let mut op_start_signals = Vec::new();
5751
5752 let mut next_entity_id = 100000;
5753 let mut host_cx = TestAppContext::new(
5754 cx.foreground_platform(),
5755 cx.platform(),
5756 deterministic.build_foreground(next_entity_id),
5757 deterministic.build_background(),
5758 cx.font_cache(),
5759 cx.leak_detector(),
5760 next_entity_id,
5761 );
5762 let host = server.create_client(&mut host_cx, "host").await;
5763 let host_project = host_cx.update(|cx| {
5764 Project::local(
5765 host.client.clone(),
5766 host.user_store.clone(),
5767 host_language_registry.clone(),
5768 fs.clone(),
5769 cx,
5770 )
5771 });
5772 let host_project_id = host_project
5773 .update(&mut host_cx, |p, _| p.next_remote_id())
5774 .await;
5775
5776 let (collab_worktree, _) = host_project
5777 .update(&mut host_cx, |project, cx| {
5778 project.find_or_create_local_worktree("/_collab", true, cx)
5779 })
5780 .await
5781 .unwrap();
5782 collab_worktree
5783 .read_with(&host_cx, |tree, _| tree.as_local().unwrap().scan_complete())
5784 .await;
5785
5786 // Set up fake language servers.
5787 let mut language = Language::new(
5788 LanguageConfig {
5789 name: "Rust".into(),
5790 path_suffixes: vec!["rs".to_string()],
5791 ..Default::default()
5792 },
5793 None,
5794 );
5795 let _fake_servers = language.set_fake_lsp_adapter(FakeLspAdapter {
5796 name: "the-fake-language-server",
5797 capabilities: lsp::LanguageServer::full_capabilities(),
5798 initializer: Some(Box::new({
5799 let rng = rng.clone();
5800 let fs = fs.clone();
5801 let project = host_project.downgrade();
5802 move |fake_server: &mut FakeLanguageServer| {
5803 fake_server.handle_request::<lsp::request::Completion, _, _>(
5804 |_, _| async move {
5805 Ok(Some(lsp::CompletionResponse::Array(vec![
5806 lsp::CompletionItem {
5807 text_edit: Some(lsp::CompletionTextEdit::Edit(lsp::TextEdit {
5808 range: lsp::Range::new(
5809 lsp::Position::new(0, 0),
5810 lsp::Position::new(0, 0),
5811 ),
5812 new_text: "the-new-text".to_string(),
5813 })),
5814 ..Default::default()
5815 },
5816 ])))
5817 },
5818 );
5819
5820 fake_server.handle_request::<lsp::request::CodeActionRequest, _, _>(
5821 |_, _| async move {
5822 Ok(Some(vec![lsp::CodeActionOrCommand::CodeAction(
5823 lsp::CodeAction {
5824 title: "the-code-action".to_string(),
5825 ..Default::default()
5826 },
5827 )]))
5828 },
5829 );
5830
5831 fake_server.handle_request::<lsp::request::PrepareRenameRequest, _, _>(
5832 |params, _| async move {
5833 Ok(Some(lsp::PrepareRenameResponse::Range(lsp::Range::new(
5834 params.position,
5835 params.position,
5836 ))))
5837 },
5838 );
5839
5840 fake_server.handle_request::<lsp::request::GotoDefinition, _, _>({
5841 let fs = fs.clone();
5842 let rng = rng.clone();
5843 move |_, _| {
5844 let fs = fs.clone();
5845 let rng = rng.clone();
5846 async move {
5847 let files = fs.files().await;
5848 let mut rng = rng.lock();
5849 let count = rng.gen_range::<usize, _>(1..3);
5850 let files = (0..count)
5851 .map(|_| files.choose(&mut *rng).unwrap())
5852 .collect::<Vec<_>>();
5853 log::info!("LSP: Returning definitions in files {:?}", &files);
5854 Ok(Some(lsp::GotoDefinitionResponse::Array(
5855 files
5856 .into_iter()
5857 .map(|file| lsp::Location {
5858 uri: lsp::Url::from_file_path(file).unwrap(),
5859 range: Default::default(),
5860 })
5861 .collect(),
5862 )))
5863 }
5864 }
5865 });
5866
5867 fake_server.handle_request::<lsp::request::DocumentHighlightRequest, _, _>({
5868 let rng = rng.clone();
5869 let project = project.clone();
5870 move |params, mut cx| {
5871 let highlights = if let Some(project) = project.upgrade(&cx) {
5872 project.update(&mut cx, |project, cx| {
5873 let path = params
5874 .text_document_position_params
5875 .text_document
5876 .uri
5877 .to_file_path()
5878 .unwrap();
5879 let (worktree, relative_path) =
5880 project.find_local_worktree(&path, cx)?;
5881 let project_path =
5882 ProjectPath::from((worktree.read(cx).id(), relative_path));
5883 let buffer =
5884 project.get_open_buffer(&project_path, cx)?.read(cx);
5885
5886 let mut highlights = Vec::new();
5887 let highlight_count = rng.lock().gen_range(1..=5);
5888 let mut prev_end = 0;
5889 for _ in 0..highlight_count {
5890 let range =
5891 buffer.random_byte_range(prev_end, &mut *rng.lock());
5892
5893 highlights.push(lsp::DocumentHighlight {
5894 range: range_to_lsp(range.to_point_utf16(buffer)),
5895 kind: Some(lsp::DocumentHighlightKind::READ),
5896 });
5897 prev_end = range.end;
5898 }
5899 Some(highlights)
5900 })
5901 } else {
5902 None
5903 };
5904 async move { Ok(highlights) }
5905 }
5906 });
5907 }
5908 })),
5909 ..Default::default()
5910 });
5911 host_language_registry.add(Arc::new(language));
5912
5913 let op_start_signal = futures::channel::mpsc::unbounded();
5914 user_ids.push(host.current_user_id(&host_cx));
5915 op_start_signals.push(op_start_signal.0);
5916 clients.push(host_cx.foreground().spawn(host.simulate_host(
5917 host_project,
5918 op_start_signal.1,
5919 rng.clone(),
5920 host_cx,
5921 )));
5922
5923 let disconnect_host_at = if rng.lock().gen_bool(0.2) {
5924 rng.lock().gen_range(0..max_operations)
5925 } else {
5926 max_operations
5927 };
5928 let mut available_guests = vec![
5929 "guest-1".to_string(),
5930 "guest-2".to_string(),
5931 "guest-3".to_string(),
5932 "guest-4".to_string(),
5933 ];
5934 let mut operations = 0;
5935 while operations < max_operations {
5936 if operations == disconnect_host_at {
5937 server.disconnect_client(user_ids[0]);
5938 cx.foreground().advance_clock(RECEIVE_TIMEOUT);
5939 drop(op_start_signals);
5940 let mut clients = futures::future::join_all(clients).await;
5941 cx.foreground().run_until_parked();
5942
5943 let (host, mut host_cx, host_err) = clients.remove(0);
5944 if let Some(host_err) = host_err {
5945 log::error!("host error - {:?}", host_err);
5946 }
5947 host.project
5948 .as_ref()
5949 .unwrap()
5950 .read_with(&host_cx, |project, _| assert!(!project.is_shared()));
5951 for (guest, mut guest_cx, guest_err) in clients {
5952 if let Some(guest_err) = guest_err {
5953 log::error!("{} error - {:?}", guest.username, guest_err);
5954 }
5955 // TODO
5956 // let contacts = server
5957 // .store
5958 // .read()
5959 // .await
5960 // .contacts_for_user(guest.current_user_id(&guest_cx));
5961 // assert!(!contacts
5962 // .iter()
5963 // .flat_map(|contact| &contact.projects)
5964 // .any(|project| project.id == host_project_id));
5965 guest
5966 .project
5967 .as_ref()
5968 .unwrap()
5969 .read_with(&guest_cx, |project, _| assert!(project.is_read_only()));
5970 guest_cx.update(|_| drop(guest));
5971 }
5972 host_cx.update(|_| drop(host));
5973
5974 return;
5975 }
5976
5977 let distribution = rng.lock().gen_range(0..100);
5978 match distribution {
5979 0..=19 if !available_guests.is_empty() => {
5980 let guest_ix = rng.lock().gen_range(0..available_guests.len());
5981 let guest_username = available_guests.remove(guest_ix);
5982 log::info!("Adding new connection for {}", guest_username);
5983 next_entity_id += 100000;
5984 let mut guest_cx = TestAppContext::new(
5985 cx.foreground_platform(),
5986 cx.platform(),
5987 deterministic.build_foreground(next_entity_id),
5988 deterministic.build_background(),
5989 cx.font_cache(),
5990 cx.leak_detector(),
5991 next_entity_id,
5992 );
5993 let guest = server.create_client(&mut guest_cx, &guest_username).await;
5994 let guest_project = Project::remote(
5995 host_project_id,
5996 guest.client.clone(),
5997 guest.user_store.clone(),
5998 guest_lang_registry.clone(),
5999 FakeFs::new(cx.background()),
6000 &mut guest_cx.to_async(),
6001 )
6002 .await
6003 .unwrap();
6004 let op_start_signal = futures::channel::mpsc::unbounded();
6005 user_ids.push(guest.current_user_id(&guest_cx));
6006 op_start_signals.push(op_start_signal.0);
6007 clients.push(guest_cx.foreground().spawn(guest.simulate_guest(
6008 guest_username.clone(),
6009 guest_project,
6010 op_start_signal.1,
6011 rng.clone(),
6012 guest_cx,
6013 )));
6014
6015 log::info!("Added connection for {}", guest_username);
6016 operations += 1;
6017 }
6018 20..=29 if clients.len() > 1 => {
6019 let guest_ix = rng.lock().gen_range(1..clients.len());
6020 log::info!("Removing guest {}", user_ids[guest_ix]);
6021 let removed_guest_id = user_ids.remove(guest_ix);
6022 let guest = clients.remove(guest_ix);
6023 op_start_signals.remove(guest_ix);
6024 server.disconnect_client(removed_guest_id);
6025 cx.foreground().advance_clock(RECEIVE_TIMEOUT);
6026 let (guest, mut guest_cx, guest_err) = guest.await;
6027 if let Some(guest_err) = guest_err {
6028 log::error!("{} error - {:?}", guest.username, guest_err);
6029 }
6030 guest
6031 .project
6032 .as_ref()
6033 .unwrap()
6034 .read_with(&guest_cx, |project, _| assert!(project.is_read_only()));
6035 // TODO
6036 // for user_id in &user_ids {
6037 // for contact in server.store.read().await.contacts_for_user(*user_id) {
6038 // assert_ne!(
6039 // contact.user_id, removed_guest_id.0 as u64,
6040 // "removed guest is still a contact of another peer"
6041 // );
6042 // for project in contact.projects {
6043 // for project_guest_id in project.guests {
6044 // assert_ne!(
6045 // project_guest_id, removed_guest_id.0 as u64,
6046 // "removed guest appears as still participating on a project"
6047 // );
6048 // }
6049 // }
6050 // }
6051 // }
6052
6053 log::info!("{} removed", guest.username);
6054 available_guests.push(guest.username.clone());
6055 guest_cx.update(|_| drop(guest));
6056
6057 operations += 1;
6058 }
6059 _ => {
6060 while operations < max_operations && rng.lock().gen_bool(0.7) {
6061 op_start_signals
6062 .choose(&mut *rng.lock())
6063 .unwrap()
6064 .unbounded_send(())
6065 .unwrap();
6066 operations += 1;
6067 }
6068
6069 if rng.lock().gen_bool(0.8) {
6070 cx.foreground().run_until_parked();
6071 }
6072 }
6073 }
6074 }
6075
6076 drop(op_start_signals);
6077 let mut clients = futures::future::join_all(clients).await;
6078 cx.foreground().run_until_parked();
6079
6080 let (host_client, mut host_cx, host_err) = clients.remove(0);
6081 if let Some(host_err) = host_err {
6082 panic!("host error - {:?}", host_err);
6083 }
6084 let host_project = host_client.project.as_ref().unwrap();
6085 let host_worktree_snapshots = host_project.read_with(&host_cx, |project, cx| {
6086 project
6087 .worktrees(cx)
6088 .map(|worktree| {
6089 let snapshot = worktree.read(cx).snapshot();
6090 (snapshot.id(), snapshot)
6091 })
6092 .collect::<BTreeMap<_, _>>()
6093 });
6094
6095 host_client
6096 .project
6097 .as_ref()
6098 .unwrap()
6099 .read_with(&host_cx, |project, cx| project.check_invariants(cx));
6100
6101 for (guest_client, mut guest_cx, guest_err) in clients.into_iter() {
6102 if let Some(guest_err) = guest_err {
6103 panic!("{} error - {:?}", guest_client.username, guest_err);
6104 }
6105 let worktree_snapshots =
6106 guest_client
6107 .project
6108 .as_ref()
6109 .unwrap()
6110 .read_with(&guest_cx, |project, cx| {
6111 project
6112 .worktrees(cx)
6113 .map(|worktree| {
6114 let worktree = worktree.read(cx);
6115 (worktree.id(), worktree.snapshot())
6116 })
6117 .collect::<BTreeMap<_, _>>()
6118 });
6119
6120 assert_eq!(
6121 worktree_snapshots.keys().collect::<Vec<_>>(),
6122 host_worktree_snapshots.keys().collect::<Vec<_>>(),
6123 "{} has different worktrees than the host",
6124 guest_client.username
6125 );
6126 for (id, host_snapshot) in &host_worktree_snapshots {
6127 let guest_snapshot = &worktree_snapshots[id];
6128 assert_eq!(
6129 guest_snapshot.root_name(),
6130 host_snapshot.root_name(),
6131 "{} has different root name than the host for worktree {}",
6132 guest_client.username,
6133 id
6134 );
6135 assert_eq!(
6136 guest_snapshot.entries(false).collect::<Vec<_>>(),
6137 host_snapshot.entries(false).collect::<Vec<_>>(),
6138 "{} has different snapshot than the host for worktree {}",
6139 guest_client.username,
6140 id
6141 );
6142 assert_eq!(guest_snapshot.scan_id(), host_snapshot.scan_id());
6143 }
6144
6145 guest_client
6146 .project
6147 .as_ref()
6148 .unwrap()
6149 .read_with(&guest_cx, |project, cx| project.check_invariants(cx));
6150
6151 for guest_buffer in &guest_client.buffers {
6152 let buffer_id = guest_buffer.read_with(&guest_cx, |buffer, _| buffer.remote_id());
6153 let host_buffer = host_project.read_with(&host_cx, |project, cx| {
6154 project.buffer_for_id(buffer_id, cx).expect(&format!(
6155 "host does not have buffer for guest:{}, peer:{}, id:{}",
6156 guest_client.username, guest_client.peer_id, buffer_id
6157 ))
6158 });
6159 let path = host_buffer
6160 .read_with(&host_cx, |buffer, cx| buffer.file().unwrap().full_path(cx));
6161
6162 assert_eq!(
6163 guest_buffer.read_with(&guest_cx, |buffer, _| buffer.deferred_ops_len()),
6164 0,
6165 "{}, buffer {}, path {:?} has deferred operations",
6166 guest_client.username,
6167 buffer_id,
6168 path,
6169 );
6170 assert_eq!(
6171 guest_buffer.read_with(&guest_cx, |buffer, _| buffer.text()),
6172 host_buffer.read_with(&host_cx, |buffer, _| buffer.text()),
6173 "{}, buffer {}, path {:?}, differs from the host's buffer",
6174 guest_client.username,
6175 buffer_id,
6176 path
6177 );
6178 }
6179
6180 guest_cx.update(|_| drop(guest_client));
6181 }
6182
6183 host_cx.update(|_| drop(host_client));
6184 }
6185
6186 struct TestServer {
6187 peer: Arc<Peer>,
6188 app_state: Arc<AppState>,
6189 server: Arc<Server>,
6190 foreground: Rc<executor::Foreground>,
6191 notifications: mpsc::UnboundedReceiver<()>,
6192 connection_killers: Arc<Mutex<HashMap<UserId, Arc<AtomicBool>>>>,
6193 forbid_connections: Arc<AtomicBool>,
6194 _test_db: TestDb,
6195 }
6196
6197 impl TestServer {
6198 async fn start(
6199 foreground: Rc<executor::Foreground>,
6200 background: Arc<executor::Background>,
6201 ) -> Self {
6202 let test_db = TestDb::fake(background);
6203 let app_state = Self::build_app_state(&test_db).await;
6204 let peer = Peer::new();
6205 let notifications = mpsc::unbounded();
6206 let server = Server::new(app_state.clone(), Some(notifications.0));
6207 Self {
6208 peer,
6209 app_state,
6210 server,
6211 foreground,
6212 notifications: notifications.1,
6213 connection_killers: Default::default(),
6214 forbid_connections: Default::default(),
6215 _test_db: test_db,
6216 }
6217 }
6218
6219 async fn create_client(&mut self, cx: &mut TestAppContext, name: &str) -> TestClient {
6220 cx.update(|cx| {
6221 let settings = Settings::test(cx);
6222 cx.set_global(settings);
6223 });
6224
6225 let http = FakeHttpClient::with_404_response();
6226 let user_id =
6227 if let Ok(Some(user)) = self.app_state.db.get_user_by_github_login(name).await {
6228 user.id
6229 } else {
6230 self.app_state.db.create_user(name, false).await.unwrap()
6231 };
6232 let client_name = name.to_string();
6233 let mut client = Client::new(http.clone());
6234 let server = self.server.clone();
6235 let connection_killers = self.connection_killers.clone();
6236 let forbid_connections = self.forbid_connections.clone();
6237 let (connection_id_tx, mut connection_id_rx) = mpsc::channel(16);
6238
6239 Arc::get_mut(&mut client)
6240 .unwrap()
6241 .override_authenticate(move |cx| {
6242 cx.spawn(|_| async move {
6243 let access_token = "the-token".to_string();
6244 Ok(Credentials {
6245 user_id: user_id.0 as u64,
6246 access_token,
6247 })
6248 })
6249 })
6250 .override_establish_connection(move |credentials, cx| {
6251 assert_eq!(credentials.user_id, user_id.0 as u64);
6252 assert_eq!(credentials.access_token, "the-token");
6253
6254 let server = server.clone();
6255 let connection_killers = connection_killers.clone();
6256 let forbid_connections = forbid_connections.clone();
6257 let client_name = client_name.clone();
6258 let connection_id_tx = connection_id_tx.clone();
6259 cx.spawn(move |cx| async move {
6260 if forbid_connections.load(SeqCst) {
6261 Err(EstablishConnectionError::other(anyhow!(
6262 "server is forbidding connections"
6263 )))
6264 } else {
6265 let (client_conn, server_conn, killed) =
6266 Connection::in_memory(cx.background());
6267 connection_killers.lock().insert(user_id, killed);
6268 cx.background()
6269 .spawn(server.handle_connection(
6270 server_conn,
6271 client_name,
6272 user_id,
6273 Some(connection_id_tx),
6274 cx.background(),
6275 ))
6276 .detach();
6277 Ok(client_conn)
6278 }
6279 })
6280 });
6281
6282 client
6283 .authenticate_and_connect(false, &cx.to_async())
6284 .await
6285 .unwrap();
6286
6287 Channel::init(&client);
6288 Project::init(&client);
6289 cx.update(|cx| {
6290 workspace::init(&client, cx);
6291 });
6292
6293 let peer_id = PeerId(connection_id_rx.next().await.unwrap().0);
6294 let user_store = cx.add_model(|cx| UserStore::new(client.clone(), http, cx));
6295
6296 let client = TestClient {
6297 client,
6298 peer_id,
6299 username: name.to_string(),
6300 user_store,
6301 language_registry: Arc::new(LanguageRegistry::test()),
6302 project: Default::default(),
6303 buffers: Default::default(),
6304 };
6305 client.wait_for_current_user(cx).await;
6306 client
6307 }
6308
6309 fn disconnect_client(&self, user_id: UserId) {
6310 self.connection_killers
6311 .lock()
6312 .remove(&user_id)
6313 .unwrap()
6314 .store(true, SeqCst);
6315 }
6316
6317 fn forbid_connections(&self) {
6318 self.forbid_connections.store(true, SeqCst);
6319 }
6320
6321 fn allow_connections(&self) {
6322 self.forbid_connections.store(false, SeqCst);
6323 }
6324
6325 async fn make_contacts(&self, mut clients: Vec<(&TestClient, &mut TestAppContext)>) {
6326 while let Some((client_a, cx_a)) = clients.pop() {
6327 for (client_b, cx_b) in &mut clients {
6328 client_a
6329 .user_store
6330 .update(cx_a, |store, cx| {
6331 store.request_contact(client_b.user_id().unwrap(), cx)
6332 })
6333 .await
6334 .unwrap();
6335 cx_a.foreground().run_until_parked();
6336 client_b
6337 .user_store
6338 .update(*cx_b, |store, cx| {
6339 store.respond_to_contact_request(client_a.user_id().unwrap(), true, cx)
6340 })
6341 .await
6342 .unwrap();
6343 }
6344 }
6345 }
6346
6347 async fn build_app_state(test_db: &TestDb) -> Arc<AppState> {
6348 Arc::new(AppState {
6349 db: test_db.db().clone(),
6350 api_token: Default::default(),
6351 })
6352 }
6353
6354 async fn state<'a>(&'a self) -> RwLockReadGuard<'a, Store> {
6355 self.server.store.read().await
6356 }
6357
6358 async fn condition<F>(&mut self, mut predicate: F)
6359 where
6360 F: FnMut(&Store) -> bool,
6361 {
6362 assert!(
6363 self.foreground.parking_forbidden(),
6364 "you must call forbid_parking to use server conditions so we don't block indefinitely"
6365 );
6366 while !(predicate)(&*self.server.store.read().await) {
6367 self.foreground.start_waiting();
6368 self.notifications.next().await;
6369 self.foreground.finish_waiting();
6370 }
6371 }
6372 }
6373
6374 impl Deref for TestServer {
6375 type Target = Server;
6376
6377 fn deref(&self) -> &Self::Target {
6378 &self.server
6379 }
6380 }
6381
6382 impl Drop for TestServer {
6383 fn drop(&mut self) {
6384 self.peer.reset();
6385 }
6386 }
6387
6388 struct TestClient {
6389 client: Arc<Client>,
6390 username: String,
6391 pub peer_id: PeerId,
6392 pub user_store: ModelHandle<UserStore>,
6393 language_registry: Arc<LanguageRegistry>,
6394 project: Option<ModelHandle<Project>>,
6395 buffers: HashSet<ModelHandle<language::Buffer>>,
6396 }
6397
6398 impl Deref for TestClient {
6399 type Target = Arc<Client>;
6400
6401 fn deref(&self) -> &Self::Target {
6402 &self.client
6403 }
6404 }
6405
6406 struct ContactsSummary {
6407 pub current: Vec<String>,
6408 pub outgoing_requests: Vec<String>,
6409 pub incoming_requests: Vec<String>,
6410 }
6411
6412 impl TestClient {
6413 pub fn current_user_id(&self, cx: &TestAppContext) -> UserId {
6414 UserId::from_proto(
6415 self.user_store
6416 .read_with(cx, |user_store, _| user_store.current_user().unwrap().id),
6417 )
6418 }
6419
6420 async fn wait_for_current_user(&self, cx: &TestAppContext) {
6421 let mut authed_user = self
6422 .user_store
6423 .read_with(cx, |user_store, _| user_store.watch_current_user());
6424 while authed_user.next().await.unwrap().is_none() {}
6425 }
6426
6427 async fn clear_contacts(&self, cx: &mut TestAppContext) {
6428 self.user_store
6429 .update(cx, |store, _| store.clear_contacts())
6430 .await;
6431 }
6432
6433 fn summarize_contacts(&self, cx: &TestAppContext) -> ContactsSummary {
6434 self.user_store.read_with(cx, |store, _| ContactsSummary {
6435 current: store
6436 .contacts()
6437 .iter()
6438 .map(|contact| contact.user.github_login.clone())
6439 .collect(),
6440 outgoing_requests: store
6441 .outgoing_contact_requests()
6442 .iter()
6443 .map(|user| user.github_login.clone())
6444 .collect(),
6445 incoming_requests: store
6446 .incoming_contact_requests()
6447 .iter()
6448 .map(|user| user.github_login.clone())
6449 .collect(),
6450 })
6451 }
6452
6453 async fn build_local_project(
6454 &mut self,
6455 fs: Arc<FakeFs>,
6456 root_path: impl AsRef<Path>,
6457 cx: &mut TestAppContext,
6458 ) -> (ModelHandle<Project>, WorktreeId) {
6459 let project = cx.update(|cx| {
6460 Project::local(
6461 self.client.clone(),
6462 self.user_store.clone(),
6463 self.language_registry.clone(),
6464 fs,
6465 cx,
6466 )
6467 });
6468 self.project = Some(project.clone());
6469 let (worktree, _) = project
6470 .update(cx, |p, cx| {
6471 p.find_or_create_local_worktree(root_path, true, cx)
6472 })
6473 .await
6474 .unwrap();
6475 worktree
6476 .read_with(cx, |tree, _| tree.as_local().unwrap().scan_complete())
6477 .await;
6478 project
6479 .update(cx, |project, _| project.next_remote_id())
6480 .await;
6481 (project, worktree.read_with(cx, |tree, _| tree.id()))
6482 }
6483
6484 async fn build_remote_project(
6485 &mut self,
6486 host_project: &ModelHandle<Project>,
6487 host_cx: &mut TestAppContext,
6488 guest_cx: &mut TestAppContext,
6489 ) -> ModelHandle<Project> {
6490 let host_project_id = host_project
6491 .read_with(host_cx, |project, _| project.next_remote_id())
6492 .await;
6493 let guest_user_id = self.user_id().unwrap();
6494 let languages =
6495 host_project.read_with(host_cx, |project, _| project.languages().clone());
6496 let project_b = guest_cx.spawn(|mut cx| {
6497 let user_store = self.user_store.clone();
6498 let guest_client = self.client.clone();
6499 async move {
6500 Project::remote(
6501 host_project_id,
6502 guest_client,
6503 user_store.clone(),
6504 languages,
6505 FakeFs::new(cx.background()),
6506 &mut cx,
6507 )
6508 .await
6509 .unwrap()
6510 }
6511 });
6512 host_cx.foreground().run_until_parked();
6513 host_project.update(host_cx, |project, cx| {
6514 project.respond_to_join_request(guest_user_id, true, cx)
6515 });
6516 let project = project_b.await;
6517 self.project = Some(project.clone());
6518 project
6519 }
6520
6521 fn build_workspace(
6522 &self,
6523 project: &ModelHandle<Project>,
6524 cx: &mut TestAppContext,
6525 ) -> ViewHandle<Workspace> {
6526 let (window_id, _) = cx.add_window(|_| EmptyView);
6527 cx.add_view(window_id, |cx| {
6528 let fs = project.read(cx).fs().clone();
6529 Workspace::new(
6530 &WorkspaceParams {
6531 fs,
6532 project: project.clone(),
6533 user_store: self.user_store.clone(),
6534 languages: self.language_registry.clone(),
6535 themes: ThemeRegistry::new((), cx.font_cache().clone()),
6536 channel_list: cx.add_model(|cx| {
6537 ChannelList::new(self.user_store.clone(), self.client.clone(), cx)
6538 }),
6539 client: self.client.clone(),
6540 },
6541 cx,
6542 )
6543 })
6544 }
6545
6546 async fn simulate_host(
6547 mut self,
6548 project: ModelHandle<Project>,
6549 op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
6550 rng: Arc<Mutex<StdRng>>,
6551 mut cx: TestAppContext,
6552 ) -> (Self, TestAppContext, Option<anyhow::Error>) {
6553 async fn simulate_host_internal(
6554 client: &mut TestClient,
6555 project: ModelHandle<Project>,
6556 mut op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
6557 rng: Arc<Mutex<StdRng>>,
6558 cx: &mut TestAppContext,
6559 ) -> anyhow::Result<()> {
6560 let fs = project.read_with(cx, |project, _| project.fs().clone());
6561
6562 cx.update(|cx| {
6563 cx.subscribe(&project, move |project, event, cx| {
6564 if let project::Event::ContactRequestedJoin(user) = event {
6565 log::info!("Host: accepting join request from {}", user.github_login);
6566 project.update(cx, |project, cx| {
6567 project.respond_to_join_request(user.id, true, cx)
6568 });
6569 }
6570 })
6571 .detach();
6572 });
6573
6574 while op_start_signal.next().await.is_some() {
6575 let distribution = rng.lock().gen_range::<usize, _>(0..100);
6576 let files = fs.as_fake().files().await;
6577 match distribution {
6578 0..=19 if !files.is_empty() => {
6579 let path = files.choose(&mut *rng.lock()).unwrap();
6580 let mut path = path.as_path();
6581 while let Some(parent_path) = path.parent() {
6582 path = parent_path;
6583 if rng.lock().gen() {
6584 break;
6585 }
6586 }
6587
6588 log::info!("Host: find/create local worktree {:?}", path);
6589 let find_or_create_worktree = project.update(cx, |project, cx| {
6590 project.find_or_create_local_worktree(path, true, cx)
6591 });
6592 if rng.lock().gen() {
6593 cx.background().spawn(find_or_create_worktree).detach();
6594 } else {
6595 find_or_create_worktree.await?;
6596 }
6597 }
6598 20..=79 if !files.is_empty() => {
6599 let buffer = if client.buffers.is_empty() || rng.lock().gen() {
6600 let file = files.choose(&mut *rng.lock()).unwrap();
6601 let (worktree, path) = project
6602 .update(cx, |project, cx| {
6603 project.find_or_create_local_worktree(
6604 file.clone(),
6605 true,
6606 cx,
6607 )
6608 })
6609 .await?;
6610 let project_path =
6611 worktree.read_with(cx, |worktree, _| (worktree.id(), path));
6612 log::info!(
6613 "Host: opening path {:?}, worktree {}, relative_path {:?}",
6614 file,
6615 project_path.0,
6616 project_path.1
6617 );
6618 let buffer = project
6619 .update(cx, |project, cx| project.open_buffer(project_path, cx))
6620 .await
6621 .unwrap();
6622 client.buffers.insert(buffer.clone());
6623 buffer
6624 } else {
6625 client
6626 .buffers
6627 .iter()
6628 .choose(&mut *rng.lock())
6629 .unwrap()
6630 .clone()
6631 };
6632
6633 if rng.lock().gen_bool(0.1) {
6634 cx.update(|cx| {
6635 log::info!(
6636 "Host: dropping buffer {:?}",
6637 buffer.read(cx).file().unwrap().full_path(cx)
6638 );
6639 client.buffers.remove(&buffer);
6640 drop(buffer);
6641 });
6642 } else {
6643 buffer.update(cx, |buffer, cx| {
6644 log::info!(
6645 "Host: updating buffer {:?} ({})",
6646 buffer.file().unwrap().full_path(cx),
6647 buffer.remote_id()
6648 );
6649
6650 if rng.lock().gen_bool(0.7) {
6651 buffer.randomly_edit(&mut *rng.lock(), 5, cx);
6652 } else {
6653 buffer.randomly_undo_redo(&mut *rng.lock(), cx);
6654 }
6655 });
6656 }
6657 }
6658 _ => loop {
6659 let path_component_count = rng.lock().gen_range::<usize, _>(1..=5);
6660 let mut path = PathBuf::new();
6661 path.push("/");
6662 for _ in 0..path_component_count {
6663 let letter = rng.lock().gen_range(b'a'..=b'z');
6664 path.push(std::str::from_utf8(&[letter]).unwrap());
6665 }
6666 path.set_extension("rs");
6667 let parent_path = path.parent().unwrap();
6668
6669 log::info!("Host: creating file {:?}", path,);
6670
6671 if fs.create_dir(&parent_path).await.is_ok()
6672 && fs.create_file(&path, Default::default()).await.is_ok()
6673 {
6674 break;
6675 } else {
6676 log::info!("Host: cannot create file");
6677 }
6678 },
6679 }
6680
6681 cx.background().simulate_random_delay().await;
6682 }
6683
6684 Ok(())
6685 }
6686
6687 let result =
6688 simulate_host_internal(&mut self, project.clone(), op_start_signal, rng, &mut cx)
6689 .await;
6690 log::info!("Host done");
6691 self.project = Some(project);
6692 (self, cx, result.err())
6693 }
6694
6695 pub async fn simulate_guest(
6696 mut self,
6697 guest_username: String,
6698 project: ModelHandle<Project>,
6699 op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
6700 rng: Arc<Mutex<StdRng>>,
6701 mut cx: TestAppContext,
6702 ) -> (Self, TestAppContext, Option<anyhow::Error>) {
6703 async fn simulate_guest_internal(
6704 client: &mut TestClient,
6705 guest_username: &str,
6706 project: ModelHandle<Project>,
6707 mut op_start_signal: futures::channel::mpsc::UnboundedReceiver<()>,
6708 rng: Arc<Mutex<StdRng>>,
6709 cx: &mut TestAppContext,
6710 ) -> anyhow::Result<()> {
6711 while op_start_signal.next().await.is_some() {
6712 let buffer = if client.buffers.is_empty() || rng.lock().gen() {
6713 let worktree = if let Some(worktree) =
6714 project.read_with(cx, |project, cx| {
6715 project
6716 .worktrees(&cx)
6717 .filter(|worktree| {
6718 let worktree = worktree.read(cx);
6719 worktree.is_visible()
6720 && worktree.entries(false).any(|e| e.is_file())
6721 })
6722 .choose(&mut *rng.lock())
6723 }) {
6724 worktree
6725 } else {
6726 cx.background().simulate_random_delay().await;
6727 continue;
6728 };
6729
6730 let (worktree_root_name, project_path) =
6731 worktree.read_with(cx, |worktree, _| {
6732 let entry = worktree
6733 .entries(false)
6734 .filter(|e| e.is_file())
6735 .choose(&mut *rng.lock())
6736 .unwrap();
6737 (
6738 worktree.root_name().to_string(),
6739 (worktree.id(), entry.path.clone()),
6740 )
6741 });
6742 log::info!(
6743 "{}: opening path {:?} in worktree {} ({})",
6744 guest_username,
6745 project_path.1,
6746 project_path.0,
6747 worktree_root_name,
6748 );
6749 let buffer = project
6750 .update(cx, |project, cx| {
6751 project.open_buffer(project_path.clone(), cx)
6752 })
6753 .await?;
6754 log::info!(
6755 "{}: opened path {:?} in worktree {} ({}) with buffer id {}",
6756 guest_username,
6757 project_path.1,
6758 project_path.0,
6759 worktree_root_name,
6760 buffer.read_with(cx, |buffer, _| buffer.remote_id())
6761 );
6762 client.buffers.insert(buffer.clone());
6763 buffer
6764 } else {
6765 client
6766 .buffers
6767 .iter()
6768 .choose(&mut *rng.lock())
6769 .unwrap()
6770 .clone()
6771 };
6772
6773 let choice = rng.lock().gen_range(0..100);
6774 match choice {
6775 0..=9 => {
6776 cx.update(|cx| {
6777 log::info!(
6778 "{}: dropping buffer {:?}",
6779 guest_username,
6780 buffer.read(cx).file().unwrap().full_path(cx)
6781 );
6782 client.buffers.remove(&buffer);
6783 drop(buffer);
6784 });
6785 }
6786 10..=19 => {
6787 let completions = project.update(cx, |project, cx| {
6788 log::info!(
6789 "{}: requesting completions for buffer {} ({:?})",
6790 guest_username,
6791 buffer.read(cx).remote_id(),
6792 buffer.read(cx).file().unwrap().full_path(cx)
6793 );
6794 let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
6795 project.completions(&buffer, offset, cx)
6796 });
6797 let completions = cx.background().spawn(async move {
6798 completions
6799 .await
6800 .map_err(|err| anyhow!("completions request failed: {:?}", err))
6801 });
6802 if rng.lock().gen_bool(0.3) {
6803 log::info!("{}: detaching completions request", guest_username);
6804 cx.update(|cx| completions.detach_and_log_err(cx));
6805 } else {
6806 completions.await?;
6807 }
6808 }
6809 20..=29 => {
6810 let code_actions = project.update(cx, |project, cx| {
6811 log::info!(
6812 "{}: requesting code actions for buffer {} ({:?})",
6813 guest_username,
6814 buffer.read(cx).remote_id(),
6815 buffer.read(cx).file().unwrap().full_path(cx)
6816 );
6817 let range = buffer.read(cx).random_byte_range(0, &mut *rng.lock());
6818 project.code_actions(&buffer, range, cx)
6819 });
6820 let code_actions = cx.background().spawn(async move {
6821 code_actions.await.map_err(|err| {
6822 anyhow!("code actions request failed: {:?}", err)
6823 })
6824 });
6825 if rng.lock().gen_bool(0.3) {
6826 log::info!("{}: detaching code actions request", guest_username);
6827 cx.update(|cx| code_actions.detach_and_log_err(cx));
6828 } else {
6829 code_actions.await?;
6830 }
6831 }
6832 30..=39 if buffer.read_with(cx, |buffer, _| buffer.is_dirty()) => {
6833 let (requested_version, save) = buffer.update(cx, |buffer, cx| {
6834 log::info!(
6835 "{}: saving buffer {} ({:?})",
6836 guest_username,
6837 buffer.remote_id(),
6838 buffer.file().unwrap().full_path(cx)
6839 );
6840 (buffer.version(), buffer.save(cx))
6841 });
6842 let save = cx.background().spawn(async move {
6843 let (saved_version, _) = save
6844 .await
6845 .map_err(|err| anyhow!("save request failed: {:?}", err))?;
6846 assert!(saved_version.observed_all(&requested_version));
6847 Ok::<_, anyhow::Error>(())
6848 });
6849 if rng.lock().gen_bool(0.3) {
6850 log::info!("{}: detaching save request", guest_username);
6851 cx.update(|cx| save.detach_and_log_err(cx));
6852 } else {
6853 save.await?;
6854 }
6855 }
6856 40..=44 => {
6857 let prepare_rename = project.update(cx, |project, cx| {
6858 log::info!(
6859 "{}: preparing rename for buffer {} ({:?})",
6860 guest_username,
6861 buffer.read(cx).remote_id(),
6862 buffer.read(cx).file().unwrap().full_path(cx)
6863 );
6864 let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
6865 project.prepare_rename(buffer, offset, cx)
6866 });
6867 let prepare_rename = cx.background().spawn(async move {
6868 prepare_rename.await.map_err(|err| {
6869 anyhow!("prepare rename request failed: {:?}", err)
6870 })
6871 });
6872 if rng.lock().gen_bool(0.3) {
6873 log::info!("{}: detaching prepare rename request", guest_username);
6874 cx.update(|cx| prepare_rename.detach_and_log_err(cx));
6875 } else {
6876 prepare_rename.await?;
6877 }
6878 }
6879 45..=49 => {
6880 let definitions = project.update(cx, |project, cx| {
6881 log::info!(
6882 "{}: requesting definitions for buffer {} ({:?})",
6883 guest_username,
6884 buffer.read(cx).remote_id(),
6885 buffer.read(cx).file().unwrap().full_path(cx)
6886 );
6887 let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
6888 project.definition(&buffer, offset, cx)
6889 });
6890 let definitions = cx.background().spawn(async move {
6891 definitions
6892 .await
6893 .map_err(|err| anyhow!("definitions request failed: {:?}", err))
6894 });
6895 if rng.lock().gen_bool(0.3) {
6896 log::info!("{}: detaching definitions request", guest_username);
6897 cx.update(|cx| definitions.detach_and_log_err(cx));
6898 } else {
6899 client
6900 .buffers
6901 .extend(definitions.await?.into_iter().map(|loc| loc.buffer));
6902 }
6903 }
6904 50..=54 => {
6905 let highlights = project.update(cx, |project, cx| {
6906 log::info!(
6907 "{}: requesting highlights for buffer {} ({:?})",
6908 guest_username,
6909 buffer.read(cx).remote_id(),
6910 buffer.read(cx).file().unwrap().full_path(cx)
6911 );
6912 let offset = rng.lock().gen_range(0..=buffer.read(cx).len());
6913 project.document_highlights(&buffer, offset, cx)
6914 });
6915 let highlights = cx.background().spawn(async move {
6916 highlights
6917 .await
6918 .map_err(|err| anyhow!("highlights request failed: {:?}", err))
6919 });
6920 if rng.lock().gen_bool(0.3) {
6921 log::info!("{}: detaching highlights request", guest_username);
6922 cx.update(|cx| highlights.detach_and_log_err(cx));
6923 } else {
6924 highlights.await?;
6925 }
6926 }
6927 55..=59 => {
6928 let search = project.update(cx, |project, cx| {
6929 let query = rng.lock().gen_range('a'..='z');
6930 log::info!("{}: project-wide search {:?}", guest_username, query);
6931 project.search(SearchQuery::text(query, false, false), cx)
6932 });
6933 let search = cx.background().spawn(async move {
6934 search
6935 .await
6936 .map_err(|err| anyhow!("search request failed: {:?}", err))
6937 });
6938 if rng.lock().gen_bool(0.3) {
6939 log::info!("{}: detaching search request", guest_username);
6940 cx.update(|cx| search.detach_and_log_err(cx));
6941 } else {
6942 client.buffers.extend(search.await?.into_keys());
6943 }
6944 }
6945 60..=69 => {
6946 let worktree = project
6947 .read_with(cx, |project, cx| {
6948 project
6949 .worktrees(&cx)
6950 .filter(|worktree| {
6951 let worktree = worktree.read(cx);
6952 worktree.is_visible()
6953 && worktree.entries(false).any(|e| e.is_file())
6954 && worktree
6955 .root_entry()
6956 .map_or(false, |e| e.is_dir())
6957 })
6958 .choose(&mut *rng.lock())
6959 })
6960 .unwrap();
6961 let (worktree_id, worktree_root_name) = worktree
6962 .read_with(cx, |worktree, _| {
6963 (worktree.id(), worktree.root_name().to_string())
6964 });
6965
6966 let mut new_name = String::new();
6967 for _ in 0..10 {
6968 let letter = rng.lock().gen_range('a'..='z');
6969 new_name.push(letter);
6970 }
6971 let mut new_path = PathBuf::new();
6972 new_path.push(new_name);
6973 new_path.set_extension("rs");
6974 log::info!(
6975 "{}: creating {:?} in worktree {} ({})",
6976 guest_username,
6977 new_path,
6978 worktree_id,
6979 worktree_root_name,
6980 );
6981 project
6982 .update(cx, |project, cx| {
6983 project.create_entry((worktree_id, new_path), false, cx)
6984 })
6985 .unwrap()
6986 .await?;
6987 }
6988 _ => {
6989 buffer.update(cx, |buffer, cx| {
6990 log::info!(
6991 "{}: updating buffer {} ({:?})",
6992 guest_username,
6993 buffer.remote_id(),
6994 buffer.file().unwrap().full_path(cx)
6995 );
6996 if rng.lock().gen_bool(0.7) {
6997 buffer.randomly_edit(&mut *rng.lock(), 5, cx);
6998 } else {
6999 buffer.randomly_undo_redo(&mut *rng.lock(), cx);
7000 }
7001 });
7002 }
7003 }
7004 cx.background().simulate_random_delay().await;
7005 }
7006 Ok(())
7007 }
7008
7009 let result = simulate_guest_internal(
7010 &mut self,
7011 &guest_username,
7012 project.clone(),
7013 op_start_signal,
7014 rng,
7015 &mut cx,
7016 )
7017 .await;
7018 log::info!("{}: done", guest_username);
7019
7020 self.project = Some(project);
7021 (self, cx, result.err())
7022 }
7023 }
7024
7025 impl Drop for TestClient {
7026 fn drop(&mut self) {
7027 self.client.tear_down();
7028 }
7029 }
7030
7031 impl Executor for Arc<gpui::executor::Background> {
7032 type Sleep = gpui::executor::Timer;
7033
7034 fn spawn_detached<F: 'static + Send + Future<Output = ()>>(&self, future: F) {
7035 self.spawn(future).detach();
7036 }
7037
7038 fn sleep(&self, duration: Duration) -> Self::Sleep {
7039 self.as_ref().timer(duration)
7040 }
7041 }
7042
7043 fn channel_messages(channel: &Channel) -> Vec<(String, String, bool)> {
7044 channel
7045 .messages()
7046 .cursor::<()>()
7047 .map(|m| {
7048 (
7049 m.sender.github_login.clone(),
7050 m.body.clone(),
7051 m.is_pending(),
7052 )
7053 })
7054 .collect()
7055 }
7056
7057 struct EmptyView;
7058
7059 impl gpui::Entity for EmptyView {
7060 type Event = ();
7061 }
7062
7063 impl gpui::View for EmptyView {
7064 fn ui_name() -> &'static str {
7065 "empty view"
7066 }
7067
7068 fn render(&mut self, _: &mut gpui::RenderContext<Self>) -> gpui::ElementBox {
7069 gpui::Element::boxed(gpui::elements::Empty::new())
7070 }
7071 }
7072}