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