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