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