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