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