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