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