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