1use crate::{
2 rpc::{CLEANUP_TIMEOUT, RECONNECT_TIMEOUT},
3 tests::TestServer,
4};
5use call::ActiveCall;
6use channel::{Channel, ACKNOWLEDGE_DEBOUNCE_INTERVAL};
7use client::ParticipantIndex;
8use client::{Collaborator, UserId};
9use collab_ui::channel_view::ChannelView;
10use collections::HashMap;
11use editor::{Anchor, Editor, ToOffset};
12use futures::future;
13use gpui::{executor::Deterministic, ModelHandle, TestAppContext, ViewContext};
14use rpc::{
15 proto::{self, PeerId},
16 RECEIVE_TIMEOUT,
17};
18use serde_json::json;
19use std::{ops::Range, sync::Arc};
20
21#[gpui::test]
22async fn test_core_channel_buffers(
23 deterministic: Arc<Deterministic>,
24 cx_a: &mut TestAppContext,
25 cx_b: &mut TestAppContext,
26) {
27 deterministic.forbid_parking();
28 let mut server = TestServer::start(&deterministic).await;
29 let client_a = server.create_client(cx_a, "user_a").await;
30 let client_b = server.create_client(cx_b, "user_b").await;
31
32 let channel_id = server
33 .make_channel("zed", None, (&client_a, cx_a), &mut [(&client_b, cx_b)])
34 .await;
35
36 // Client A joins the channel buffer
37 let channel_buffer_a = client_a
38 .channel_store()
39 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx))
40 .await
41 .unwrap();
42
43 // Client A edits the buffer
44 let buffer_a = channel_buffer_a.read_with(cx_a, |buffer, _| buffer.buffer());
45 buffer_a.update(cx_a, |buffer, cx| {
46 buffer.edit([(0..0, "hello world")], None, cx)
47 });
48 buffer_a.update(cx_a, |buffer, cx| {
49 buffer.edit([(5..5, ", cruel")], None, cx)
50 });
51 buffer_a.update(cx_a, |buffer, cx| {
52 buffer.edit([(0..5, "goodbye")], None, cx)
53 });
54 buffer_a.update(cx_a, |buffer, cx| buffer.undo(cx));
55 assert_eq!(buffer_text(&buffer_a, cx_a), "hello, cruel world");
56 deterministic.run_until_parked();
57
58 // Client B joins the channel buffer
59 let channel_buffer_b = client_b
60 .channel_store()
61 .update(cx_b, |store, cx| store.open_channel_buffer(channel_id, cx))
62 .await
63 .unwrap();
64 channel_buffer_b.read_with(cx_b, |buffer, _| {
65 assert_collaborators(
66 buffer.collaborators(),
67 &[client_a.user_id(), client_b.user_id()],
68 );
69 });
70
71 // Client B sees the correct text, and then edits it
72 let buffer_b = channel_buffer_b.read_with(cx_b, |buffer, _| buffer.buffer());
73 assert_eq!(
74 buffer_b.read_with(cx_b, |buffer, _| buffer.remote_id()),
75 buffer_a.read_with(cx_a, |buffer, _| buffer.remote_id())
76 );
77 assert_eq!(buffer_text(&buffer_b, cx_b), "hello, cruel world");
78 buffer_b.update(cx_b, |buffer, cx| {
79 buffer.edit([(7..12, "beautiful")], None, cx)
80 });
81
82 // Both A and B see the new edit
83 deterministic.run_until_parked();
84 assert_eq!(buffer_text(&buffer_a, cx_a), "hello, beautiful world");
85 assert_eq!(buffer_text(&buffer_b, cx_b), "hello, beautiful world");
86
87 // Client A closes the channel buffer.
88 cx_a.update(|_| drop(channel_buffer_a));
89 deterministic.run_until_parked();
90
91 // Client B sees that client A is gone from the channel buffer.
92 channel_buffer_b.read_with(cx_b, |buffer, _| {
93 assert_collaborators(&buffer.collaborators(), &[client_b.user_id()]);
94 });
95
96 // Client A rejoins the channel buffer
97 let _channel_buffer_a = client_a
98 .channel_store()
99 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx))
100 .await
101 .unwrap();
102 deterministic.run_until_parked();
103
104 // Sanity test, make sure we saw A rejoining
105 channel_buffer_b.read_with(cx_b, |buffer, _| {
106 assert_collaborators(
107 &buffer.collaborators(),
108 &[client_a.user_id(), client_b.user_id()],
109 );
110 });
111
112 // Client A loses connection.
113 server.forbid_connections();
114 server.disconnect_client(client_a.peer_id().unwrap());
115 deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT);
116
117 // Client B observes A disconnect
118 channel_buffer_b.read_with(cx_b, |buffer, _| {
119 assert_collaborators(&buffer.collaborators(), &[client_b.user_id()]);
120 });
121
122 // TODO:
123 // - Test synchronizing offline updates, what happens to A's channel buffer when A disconnects
124 // - Test interaction with channel deletion while buffer is open
125}
126
127#[gpui::test]
128async fn test_channel_notes_participant_indices(
129 deterministic: Arc<Deterministic>,
130 mut cx_a: &mut TestAppContext,
131 mut cx_b: &mut TestAppContext,
132 cx_c: &mut TestAppContext,
133) {
134 deterministic.forbid_parking();
135 let mut server = TestServer::start(&deterministic).await;
136 let client_a = server.create_client(cx_a, "user_a").await;
137 let client_b = server.create_client(cx_b, "user_b").await;
138 let client_c = server.create_client(cx_c, "user_c").await;
139
140 let active_call_a = cx_a.read(ActiveCall::global);
141 let active_call_b = cx_b.read(ActiveCall::global);
142
143 cx_a.update(editor::init);
144 cx_b.update(editor::init);
145 cx_c.update(editor::init);
146
147 let channel_id = server
148 .make_channel(
149 "the-channel",
150 None,
151 (&client_a, cx_a),
152 &mut [(&client_b, cx_b), (&client_c, cx_c)],
153 )
154 .await;
155
156 client_a
157 .fs()
158 .insert_tree("/root", json!({"file.txt": "123"}))
159 .await;
160 let (project_a, worktree_id_a) = client_a.build_local_project("/root", cx_a).await;
161 let project_b = client_b.build_empty_local_project(cx_b);
162 let project_c = client_c.build_empty_local_project(cx_c);
163 let workspace_a = client_a.build_workspace(&project_a, cx_a).root(cx_a);
164 let workspace_b = client_b.build_workspace(&project_b, cx_b).root(cx_b);
165 let workspace_c = client_c.build_workspace(&project_c, cx_c).root(cx_c);
166
167 // Clients A, B, and C open the channel notes
168 let channel_view_a = cx_a
169 .update(|cx| ChannelView::open(channel_id, workspace_a.clone(), cx))
170 .await
171 .unwrap();
172 let channel_view_b = cx_b
173 .update(|cx| ChannelView::open(channel_id, workspace_b.clone(), cx))
174 .await
175 .unwrap();
176 let channel_view_c = cx_c
177 .update(|cx| ChannelView::open(channel_id, workspace_c.clone(), cx))
178 .await
179 .unwrap();
180
181 // Clients A, B, and C all insert and select some text
182 channel_view_a.update(cx_a, |notes, cx| {
183 notes.editor.update(cx, |editor, cx| {
184 editor.insert("a", cx);
185 editor.change_selections(None, cx, |selections| {
186 selections.select_ranges(vec![0..1]);
187 });
188 });
189 });
190 deterministic.run_until_parked();
191 channel_view_b.update(cx_b, |notes, cx| {
192 notes.editor.update(cx, |editor, cx| {
193 editor.move_down(&Default::default(), cx);
194 editor.insert("b", cx);
195 editor.change_selections(None, cx, |selections| {
196 selections.select_ranges(vec![1..2]);
197 });
198 });
199 });
200 deterministic.run_until_parked();
201 channel_view_c.update(cx_c, |notes, cx| {
202 notes.editor.update(cx, |editor, cx| {
203 editor.move_down(&Default::default(), cx);
204 editor.insert("c", cx);
205 editor.change_selections(None, cx, |selections| {
206 selections.select_ranges(vec![2..3]);
207 });
208 });
209 });
210
211 // Client A sees clients B and C without assigned colors, because they aren't
212 // in a call together.
213 deterministic.run_until_parked();
214 channel_view_a.update(cx_a, |notes, cx| {
215 notes.editor.update(cx, |editor, cx| {
216 assert_remote_selections(editor, &[(None, 1..2), (None, 2..3)], cx);
217 });
218 });
219
220 // Clients A and B join the same call.
221 for (call, cx) in [(&active_call_a, &mut cx_a), (&active_call_b, &mut cx_b)] {
222 call.update(*cx, |call, cx| call.join_channel(channel_id, cx))
223 .await
224 .unwrap();
225 }
226
227 // Clients A and B see each other with two different assigned colors. Client C
228 // still doesn't have a color.
229 deterministic.run_until_parked();
230 channel_view_a.update(cx_a, |notes, cx| {
231 notes.editor.update(cx, |editor, cx| {
232 assert_remote_selections(
233 editor,
234 &[(Some(ParticipantIndex(1)), 1..2), (None, 2..3)],
235 cx,
236 );
237 });
238 });
239 channel_view_b.update(cx_b, |notes, cx| {
240 notes.editor.update(cx, |editor, cx| {
241 assert_remote_selections(
242 editor,
243 &[(Some(ParticipantIndex(0)), 0..1), (None, 2..3)],
244 cx,
245 );
246 });
247 });
248
249 // Client A shares a project, and client B joins.
250 let project_id = active_call_a
251 .update(cx_a, |call, cx| call.share_project(project_a.clone(), cx))
252 .await
253 .unwrap();
254 let project_b = client_b.build_remote_project(project_id, cx_b).await;
255 let workspace_b = client_b.build_workspace(&project_b, cx_b).root(cx_b);
256
257 // Clients A and B open the same file.
258 let editor_a = workspace_a
259 .update(cx_a, |workspace, cx| {
260 workspace.open_path((worktree_id_a, "file.txt"), None, true, cx)
261 })
262 .await
263 .unwrap()
264 .downcast::<Editor>()
265 .unwrap();
266 let editor_b = workspace_b
267 .update(cx_b, |workspace, cx| {
268 workspace.open_path((worktree_id_a, "file.txt"), None, true, cx)
269 })
270 .await
271 .unwrap()
272 .downcast::<Editor>()
273 .unwrap();
274
275 editor_a.update(cx_a, |editor, cx| {
276 editor.change_selections(None, cx, |selections| {
277 selections.select_ranges(vec![0..1]);
278 });
279 });
280 editor_b.update(cx_b, |editor, cx| {
281 editor.change_selections(None, cx, |selections| {
282 selections.select_ranges(vec![2..3]);
283 });
284 });
285 deterministic.run_until_parked();
286
287 // Clients A and B see each other with the same colors as in the channel notes.
288 editor_a.update(cx_a, |editor, cx| {
289 assert_remote_selections(editor, &[(Some(ParticipantIndex(1)), 2..3)], cx);
290 });
291 editor_b.update(cx_b, |editor, cx| {
292 assert_remote_selections(editor, &[(Some(ParticipantIndex(0)), 0..1)], cx);
293 });
294}
295
296#[track_caller]
297fn assert_remote_selections(
298 editor: &mut Editor,
299 expected_selections: &[(Option<ParticipantIndex>, Range<usize>)],
300 cx: &mut ViewContext<Editor>,
301) {
302 let snapshot = editor.snapshot(cx);
303 let range = Anchor::min()..Anchor::max();
304 let remote_selections = snapshot
305 .remote_selections_in_range(&range, editor.collaboration_hub().unwrap(), cx)
306 .map(|s| {
307 let start = s.selection.start.to_offset(&snapshot.buffer_snapshot);
308 let end = s.selection.end.to_offset(&snapshot.buffer_snapshot);
309 (s.participant_index, start..end)
310 })
311 .collect::<Vec<_>>();
312 assert_eq!(
313 remote_selections, expected_selections,
314 "incorrect remote selections"
315 );
316}
317
318#[gpui::test]
319async fn test_multiple_handles_to_channel_buffer(
320 deterministic: Arc<Deterministic>,
321 cx_a: &mut TestAppContext,
322) {
323 deterministic.forbid_parking();
324 let mut server = TestServer::start(&deterministic).await;
325 let client_a = server.create_client(cx_a, "user_a").await;
326
327 let channel_id = server
328 .make_channel("the-channel", None, (&client_a, cx_a), &mut [])
329 .await;
330
331 let channel_buffer_1 = client_a
332 .channel_store()
333 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx));
334 let channel_buffer_2 = client_a
335 .channel_store()
336 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx));
337 let channel_buffer_3 = client_a
338 .channel_store()
339 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx));
340
341 // All concurrent tasks for opening a channel buffer return the same model handle.
342 let (channel_buffer, channel_buffer_2, channel_buffer_3) =
343 future::try_join3(channel_buffer_1, channel_buffer_2, channel_buffer_3)
344 .await
345 .unwrap();
346 let channel_buffer_model_id = channel_buffer.id();
347 assert_eq!(channel_buffer, channel_buffer_2);
348 assert_eq!(channel_buffer, channel_buffer_3);
349
350 channel_buffer.update(cx_a, |buffer, cx| {
351 buffer.buffer().update(cx, |buffer, cx| {
352 buffer.edit([(0..0, "hello")], None, cx);
353 })
354 });
355 deterministic.run_until_parked();
356
357 cx_a.update(|_| {
358 drop(channel_buffer);
359 drop(channel_buffer_2);
360 drop(channel_buffer_3);
361 });
362 deterministic.run_until_parked();
363
364 // The channel buffer can be reopened after dropping it.
365 let channel_buffer = client_a
366 .channel_store()
367 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx))
368 .await
369 .unwrap();
370 assert_ne!(channel_buffer.id(), channel_buffer_model_id);
371 channel_buffer.update(cx_a, |buffer, cx| {
372 buffer.buffer().update(cx, |buffer, _| {
373 assert_eq!(buffer.text(), "hello");
374 })
375 });
376}
377
378#[gpui::test]
379async fn test_channel_buffer_disconnect(
380 deterministic: Arc<Deterministic>,
381 cx_a: &mut TestAppContext,
382 cx_b: &mut TestAppContext,
383) {
384 deterministic.forbid_parking();
385 let mut server = TestServer::start(&deterministic).await;
386 let client_a = server.create_client(cx_a, "user_a").await;
387 let client_b = server.create_client(cx_b, "user_b").await;
388
389 let channel_id = server
390 .make_channel(
391 "the-channel",
392 None,
393 (&client_a, cx_a),
394 &mut [(&client_b, cx_b)],
395 )
396 .await;
397
398 let channel_buffer_a = client_a
399 .channel_store()
400 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx))
401 .await
402 .unwrap();
403 let channel_buffer_b = client_b
404 .channel_store()
405 .update(cx_b, |store, cx| store.open_channel_buffer(channel_id, cx))
406 .await
407 .unwrap();
408
409 server.forbid_connections();
410 server.disconnect_client(client_a.peer_id().unwrap());
411 deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT);
412
413 channel_buffer_a.update(cx_a, |buffer, _| {
414 assert_eq!(
415 buffer.channel().as_ref(),
416 &channel(channel_id, "the-channel")
417 );
418 assert!(!buffer.is_connected());
419 });
420
421 deterministic.run_until_parked();
422
423 server.allow_connections();
424 deterministic.advance_clock(RECEIVE_TIMEOUT + RECONNECT_TIMEOUT);
425
426 deterministic.run_until_parked();
427
428 client_a
429 .channel_store()
430 .update(cx_a, |channel_store, _| {
431 channel_store.remove_channel(channel_id)
432 })
433 .await
434 .unwrap();
435 deterministic.run_until_parked();
436
437 // Channel buffer observed the deletion
438 channel_buffer_b.update(cx_b, |buffer, _| {
439 assert_eq!(
440 buffer.channel().as_ref(),
441 &channel(channel_id, "the-channel")
442 );
443 assert!(!buffer.is_connected());
444 });
445}
446
447fn channel(id: u64, name: &'static str) -> Channel {
448 Channel {
449 id,
450 name: name.to_string(),
451 visibility: proto::ChannelVisibility::Members,
452 unseen_note_version: None,
453 unseen_message_id: None,
454 }
455}
456
457#[gpui::test]
458async fn test_rejoin_channel_buffer(
459 deterministic: Arc<Deterministic>,
460 cx_a: &mut TestAppContext,
461 cx_b: &mut TestAppContext,
462) {
463 deterministic.forbid_parking();
464 let mut server = TestServer::start(&deterministic).await;
465 let client_a = server.create_client(cx_a, "user_a").await;
466 let client_b = server.create_client(cx_b, "user_b").await;
467
468 let channel_id = server
469 .make_channel(
470 "the-channel",
471 None,
472 (&client_a, cx_a),
473 &mut [(&client_b, cx_b)],
474 )
475 .await;
476
477 let channel_buffer_a = client_a
478 .channel_store()
479 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx))
480 .await
481 .unwrap();
482 let channel_buffer_b = client_b
483 .channel_store()
484 .update(cx_b, |store, cx| store.open_channel_buffer(channel_id, cx))
485 .await
486 .unwrap();
487
488 channel_buffer_a.update(cx_a, |buffer, cx| {
489 buffer.buffer().update(cx, |buffer, cx| {
490 buffer.edit([(0..0, "1")], None, cx);
491 })
492 });
493 deterministic.run_until_parked();
494
495 // Client A disconnects.
496 server.forbid_connections();
497 server.disconnect_client(client_a.peer_id().unwrap());
498
499 // Both clients make an edit.
500 channel_buffer_a.update(cx_a, |buffer, cx| {
501 buffer.buffer().update(cx, |buffer, cx| {
502 buffer.edit([(1..1, "2")], None, cx);
503 })
504 });
505 channel_buffer_b.update(cx_b, |buffer, cx| {
506 buffer.buffer().update(cx, |buffer, cx| {
507 buffer.edit([(0..0, "0")], None, cx);
508 })
509 });
510
511 // Both clients see their own edit.
512 deterministic.run_until_parked();
513 channel_buffer_a.read_with(cx_a, |buffer, cx| {
514 assert_eq!(buffer.buffer().read(cx).text(), "12");
515 });
516 channel_buffer_b.read_with(cx_b, |buffer, cx| {
517 assert_eq!(buffer.buffer().read(cx).text(), "01");
518 });
519
520 // Client A reconnects. Both clients see each other's edits, and see
521 // the same collaborators.
522 server.allow_connections();
523 deterministic.advance_clock(RECEIVE_TIMEOUT);
524 channel_buffer_a.read_with(cx_a, |buffer, cx| {
525 assert_eq!(buffer.buffer().read(cx).text(), "012");
526 });
527 channel_buffer_b.read_with(cx_b, |buffer, cx| {
528 assert_eq!(buffer.buffer().read(cx).text(), "012");
529 });
530
531 channel_buffer_a.read_with(cx_a, |buffer_a, _| {
532 channel_buffer_b.read_with(cx_b, |buffer_b, _| {
533 assert_eq!(buffer_a.collaborators(), buffer_b.collaborators());
534 });
535 });
536}
537
538#[gpui::test]
539async fn test_channel_buffers_and_server_restarts(
540 deterministic: Arc<Deterministic>,
541 cx_a: &mut TestAppContext,
542 cx_b: &mut TestAppContext,
543 cx_c: &mut TestAppContext,
544) {
545 deterministic.forbid_parking();
546 let mut server = TestServer::start(&deterministic).await;
547 let client_a = server.create_client(cx_a, "user_a").await;
548 let client_b = server.create_client(cx_b, "user_b").await;
549 let client_c = server.create_client(cx_c, "user_c").await;
550
551 let channel_id = server
552 .make_channel(
553 "the-channel",
554 None,
555 (&client_a, cx_a),
556 &mut [(&client_b, cx_b), (&client_c, cx_c)],
557 )
558 .await;
559
560 let channel_buffer_a = client_a
561 .channel_store()
562 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx))
563 .await
564 .unwrap();
565 let channel_buffer_b = client_b
566 .channel_store()
567 .update(cx_b, |store, cx| store.open_channel_buffer(channel_id, cx))
568 .await
569 .unwrap();
570 let _channel_buffer_c = client_c
571 .channel_store()
572 .update(cx_c, |store, cx| store.open_channel_buffer(channel_id, cx))
573 .await
574 .unwrap();
575
576 channel_buffer_a.update(cx_a, |buffer, cx| {
577 buffer.buffer().update(cx, |buffer, cx| {
578 buffer.edit([(0..0, "1")], None, cx);
579 })
580 });
581 deterministic.run_until_parked();
582
583 // Client C can't reconnect.
584 client_c.override_establish_connection(|_, cx| cx.spawn(|_| future::pending()));
585
586 // Server stops.
587 server.reset().await;
588 deterministic.advance_clock(RECEIVE_TIMEOUT);
589
590 // While the server is down, both clients make an edit.
591 channel_buffer_a.update(cx_a, |buffer, cx| {
592 buffer.buffer().update(cx, |buffer, cx| {
593 buffer.edit([(1..1, "2")], None, cx);
594 })
595 });
596 channel_buffer_b.update(cx_b, |buffer, cx| {
597 buffer.buffer().update(cx, |buffer, cx| {
598 buffer.edit([(0..0, "0")], None, cx);
599 })
600 });
601
602 // Server restarts.
603 server.start().await.unwrap();
604 deterministic.advance_clock(CLEANUP_TIMEOUT);
605
606 // Clients reconnects. Clients A and B see each other's edits, and see
607 // that client C has disconnected.
608 channel_buffer_a.read_with(cx_a, |buffer, cx| {
609 assert_eq!(buffer.buffer().read(cx).text(), "012");
610 });
611 channel_buffer_b.read_with(cx_b, |buffer, cx| {
612 assert_eq!(buffer.buffer().read(cx).text(), "012");
613 });
614
615 channel_buffer_a.read_with(cx_a, |buffer_a, _| {
616 channel_buffer_b.read_with(cx_b, |buffer_b, _| {
617 assert_collaborators(
618 buffer_a.collaborators(),
619 &[client_a.user_id(), client_b.user_id()],
620 );
621 assert_eq!(buffer_a.collaborators(), buffer_b.collaborators());
622 });
623 });
624}
625
626#[gpui::test(iterations = 10)]
627async fn test_following_to_channel_notes_without_a_shared_project(
628 deterministic: Arc<Deterministic>,
629 mut cx_a: &mut TestAppContext,
630 mut cx_b: &mut TestAppContext,
631 mut cx_c: &mut TestAppContext,
632) {
633 deterministic.forbid_parking();
634 let mut server = TestServer::start(&deterministic).await;
635 let client_a = server.create_client(cx_a, "user_a").await;
636 let client_b = server.create_client(cx_b, "user_b").await;
637
638 let client_c = server.create_client(cx_c, "user_c").await;
639
640 cx_a.update(editor::init);
641 cx_b.update(editor::init);
642 cx_c.update(editor::init);
643 cx_a.update(collab_ui::channel_view::init);
644 cx_b.update(collab_ui::channel_view::init);
645 cx_c.update(collab_ui::channel_view::init);
646
647 let channel_1_id = server
648 .make_channel(
649 "channel-1",
650 None,
651 (&client_a, cx_a),
652 &mut [(&client_b, cx_b), (&client_c, cx_c)],
653 )
654 .await;
655 let channel_2_id = server
656 .make_channel(
657 "channel-2",
658 None,
659 (&client_a, cx_a),
660 &mut [(&client_b, cx_b), (&client_c, cx_c)],
661 )
662 .await;
663
664 // Clients A, B, and C join a channel.
665 let active_call_a = cx_a.read(ActiveCall::global);
666 let active_call_b = cx_b.read(ActiveCall::global);
667 let active_call_c = cx_c.read(ActiveCall::global);
668 for (call, cx) in [
669 (&active_call_a, &mut cx_a),
670 (&active_call_b, &mut cx_b),
671 (&active_call_c, &mut cx_c),
672 ] {
673 call.update(*cx, |call, cx| call.join_channel(channel_1_id, cx))
674 .await
675 .unwrap();
676 }
677 deterministic.run_until_parked();
678
679 // Clients A, B, and C all open their own unshared projects.
680 client_a.fs().insert_tree("/a", json!({})).await;
681 client_b.fs().insert_tree("/b", json!({})).await;
682 client_c.fs().insert_tree("/c", json!({})).await;
683 let (project_a, _) = client_a.build_local_project("/a", cx_a).await;
684 let (project_b, _) = client_b.build_local_project("/b", cx_b).await;
685 let (project_c, _) = client_b.build_local_project("/c", cx_c).await;
686 let workspace_a = client_a.build_workspace(&project_a, cx_a).root(cx_a);
687 let workspace_b = client_b.build_workspace(&project_b, cx_b).root(cx_b);
688 let _workspace_c = client_c.build_workspace(&project_c, cx_c).root(cx_c);
689
690 active_call_a
691 .update(cx_a, |call, cx| call.set_location(Some(&project_a), cx))
692 .await
693 .unwrap();
694
695 // Client A opens the notes for channel 1.
696 let channel_view_1_a = cx_a
697 .update(|cx| ChannelView::open(channel_1_id, workspace_a.clone(), cx))
698 .await
699 .unwrap();
700 channel_view_1_a.update(cx_a, |notes, cx| {
701 assert_eq!(notes.channel(cx).name, "channel-1");
702 notes.editor.update(cx, |editor, cx| {
703 editor.insert("Hello from A.", cx);
704 editor.change_selections(None, cx, |selections| {
705 selections.select_ranges(vec![3..4]);
706 });
707 });
708 });
709
710 // Client B follows client A.
711 workspace_b
712 .update(cx_b, |workspace, cx| {
713 workspace.follow(client_a.peer_id().unwrap(), cx).unwrap()
714 })
715 .await
716 .unwrap();
717
718 // Client B is taken to the notes for channel 1, with the same
719 // text selected as client A.
720 deterministic.run_until_parked();
721 let channel_view_1_b = workspace_b.read_with(cx_b, |workspace, cx| {
722 assert_eq!(
723 workspace.leader_for_pane(workspace.active_pane()),
724 Some(client_a.peer_id().unwrap())
725 );
726 workspace
727 .active_item(cx)
728 .expect("no active item")
729 .downcast::<ChannelView>()
730 .expect("active item is not a channel view")
731 });
732 channel_view_1_b.read_with(cx_b, |notes, cx| {
733 assert_eq!(notes.channel(cx).name, "channel-1");
734 let editor = notes.editor.read(cx);
735 assert_eq!(editor.text(cx), "Hello from A.");
736 assert_eq!(editor.selections.ranges::<usize>(cx), &[3..4]);
737 });
738
739 // Client A opens the notes for channel 2.
740 let channel_view_2_a = cx_a
741 .update(|cx| ChannelView::open(channel_2_id, workspace_a.clone(), cx))
742 .await
743 .unwrap();
744 channel_view_2_a.read_with(cx_a, |notes, cx| {
745 assert_eq!(notes.channel(cx).name, "channel-2");
746 });
747
748 // Client B is taken to the notes for channel 2.
749 deterministic.run_until_parked();
750 let channel_view_2_b = workspace_b.read_with(cx_b, |workspace, cx| {
751 assert_eq!(
752 workspace.leader_for_pane(workspace.active_pane()),
753 Some(client_a.peer_id().unwrap())
754 );
755 workspace
756 .active_item(cx)
757 .expect("no active item")
758 .downcast::<ChannelView>()
759 .expect("active item is not a channel view")
760 });
761 channel_view_2_b.read_with(cx_b, |notes, cx| {
762 assert_eq!(notes.channel(cx).name, "channel-2");
763 });
764}
765
766#[gpui::test]
767async fn test_channel_buffer_changes(
768 deterministic: Arc<Deterministic>,
769 cx_a: &mut TestAppContext,
770 cx_b: &mut TestAppContext,
771) {
772 deterministic.forbid_parking();
773 let mut server = TestServer::start(&deterministic).await;
774 let client_a = server.create_client(cx_a, "user_a").await;
775 let client_b = server.create_client(cx_b, "user_b").await;
776
777 let channel_id = server
778 .make_channel(
779 "the-channel",
780 None,
781 (&client_a, cx_a),
782 &mut [(&client_b, cx_b)],
783 )
784 .await;
785
786 let channel_buffer_a = client_a
787 .channel_store()
788 .update(cx_a, |store, cx| store.open_channel_buffer(channel_id, cx))
789 .await
790 .unwrap();
791
792 // Client A makes an edit, and client B should see that the note has changed.
793 channel_buffer_a.update(cx_a, |buffer, cx| {
794 buffer.buffer().update(cx, |buffer, cx| {
795 buffer.edit([(0..0, "1")], None, cx);
796 })
797 });
798 deterministic.run_until_parked();
799
800 let has_buffer_changed = cx_b.read(|cx| {
801 client_b
802 .channel_store()
803 .read(cx)
804 .has_channel_buffer_changed(channel_id)
805 .unwrap()
806 });
807 assert!(has_buffer_changed);
808
809 // Opening the buffer should clear the changed flag.
810 let project_b = client_b.build_empty_local_project(cx_b);
811 let workspace_b = client_b.build_workspace(&project_b, cx_b).root(cx_b);
812 let channel_view_b = cx_b
813 .update(|cx| ChannelView::open(channel_id, workspace_b.clone(), cx))
814 .await
815 .unwrap();
816 deterministic.run_until_parked();
817
818 let has_buffer_changed = cx_b.read(|cx| {
819 client_b
820 .channel_store()
821 .read(cx)
822 .has_channel_buffer_changed(channel_id)
823 .unwrap()
824 });
825 assert!(!has_buffer_changed);
826
827 // Editing the channel while the buffer is open should not show that the buffer has changed.
828 channel_buffer_a.update(cx_a, |buffer, cx| {
829 buffer.buffer().update(cx, |buffer, cx| {
830 buffer.edit([(0..0, "2")], None, cx);
831 })
832 });
833 deterministic.run_until_parked();
834
835 let has_buffer_changed = cx_b.read(|cx| {
836 client_b
837 .channel_store()
838 .read(cx)
839 .has_channel_buffer_changed(channel_id)
840 .unwrap()
841 });
842 assert!(!has_buffer_changed);
843
844 deterministic.advance_clock(ACKNOWLEDGE_DEBOUNCE_INTERVAL);
845
846 // Test that the server is tracking things correctly, and we retain our 'not changed'
847 // state across a disconnect
848 server.simulate_long_connection_interruption(client_b.peer_id().unwrap(), &deterministic);
849 let has_buffer_changed = cx_b.read(|cx| {
850 client_b
851 .channel_store()
852 .read(cx)
853 .has_channel_buffer_changed(channel_id)
854 .unwrap()
855 });
856 assert!(!has_buffer_changed);
857
858 // Closing the buffer should re-enable change tracking
859 cx_b.update(|cx| {
860 workspace_b.update(cx, |workspace, cx| {
861 workspace.close_all_items_and_panes(&Default::default(), cx)
862 });
863
864 drop(channel_view_b)
865 });
866
867 deterministic.run_until_parked();
868
869 channel_buffer_a.update(cx_a, |buffer, cx| {
870 buffer.buffer().update(cx, |buffer, cx| {
871 buffer.edit([(0..0, "3")], None, cx);
872 })
873 });
874 deterministic.run_until_parked();
875
876 let has_buffer_changed = cx_b.read(|cx| {
877 client_b
878 .channel_store()
879 .read(cx)
880 .has_channel_buffer_changed(channel_id)
881 .unwrap()
882 });
883 assert!(has_buffer_changed);
884}
885
886#[track_caller]
887fn assert_collaborators(collaborators: &HashMap<PeerId, Collaborator>, ids: &[Option<UserId>]) {
888 let mut user_ids = collaborators
889 .values()
890 .map(|collaborator| collaborator.user_id)
891 .collect::<Vec<_>>();
892 user_ids.sort();
893 assert_eq!(
894 user_ids,
895 ids.into_iter().map(|id| id.unwrap()).collect::<Vec<_>>()
896 );
897}
898
899fn buffer_text(channel_buffer: &ModelHandle<language::Buffer>, cx: &mut TestAppContext) -> String {
900 channel_buffer.read_with(cx, |buffer, _| buffer.text())
901}