1//! What the Live Share tests share: a relay, a host and guests on a notebook of two sections.
2#![allow(dead_code)]
3
4use notebook::{
5 EditStatus, Replica,
6 live::{
7 Hello,
8 share::{self, Guest, Host, Sharing},
9 },
10 session::{Notebook, Section},
11};
12use onestore::{
13 ExGuid,
14 op::{Edit, Op, PageOp},
15 protected::{Key, rekey},
16};
17use std::{
18 net::TcpListener,
19 path::Path,
20 sync::Arc,
21 thread,
22 time::{Duration, Instant},
23};
24
25#[path = "server.rs"]
26pub mod server;
27
28pub const PASSWORD: &str = "fixture password";
29
30/// A relay on this computer with `config`'s limits: its URL.
31pub fn relay(config: relay::server::Config) -> String {
32 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
33 let url = format!("ws://{}", listener.local_addr().unwrap());
34 thread::spawn(move || relay::server::serve(listener, config));
35 url
36}
37
38/// Another secret than `secret`, as a guess makes one.
39pub fn mistaken(secret: &str) -> String {
40 let first = if secret.starts_with('A') { 'B' } else { 'A' };
41 format!("{first}{}", &secret[1..])
42}
43
44pub fn hello(name: &str) -> Hello {
45 Hello::new(name.into(), None).unwrap()
46}
47
48pub fn until(what: &str, done: impl Fn() -> bool) {
49 let deadline = Instant::now() + Duration::from_secs(30);
50 while !done() {
51 assert!(Instant::now() < deadline, "{what}");
52 thread::sleep(Duration::from_millis(20));
53 }
54}
55
56/// A notebook folder holding `Garden.one`, a page reading "Original text", and a protected
57/// `Sealed.one` reading "Sealed text".
58pub fn notebook(root: &Path) -> std::path::PathBuf {
59 let folder = root.join("Garden");
60 std::fs::create_dir_all(&folder).unwrap();
61 std::fs::write(
62 folder.join("Garden.one"),
63 onestore::create_section("Garden.one", "Original text", "Fixture").unwrap(),
64 )
65 .unwrap();
66 let plain = onestore::create_section("Sealed.one", "Sealed text", "Fixture").unwrap();
67 let key = Key::new(PASSWORD).unwrap();
68 std::fs::write(
69 folder.join("Sealed.one"),
70 rekey(&plain, None, Some(&key)).unwrap(),
71 )
72 .unwrap();
73 folder
74}
75
76pub fn host(folder: &Path, cache: &Path, sharing: &Sharing, url: &str) -> Host {
77 let storage = Notebook::open(folder, cache).unwrap().into_storage();
78 Host::start(
79 storage,
80 hello("Ada"),
81 sharing.clone(),
82 "Garden",
83 None,
84 Some(url),
85 || {},
86 |_| Ok(()),
87 )
88 .unwrap()
89}
90
91/// The host's code once the relay has numbered it.
92pub fn code(host: &Host) -> String {
93 until("the code was never numbered", || {
94 host.code().is_some_and(|code| share::code(&code).is_some())
95 });
96 host.code().unwrap()
97}
98
99/// `name` joins with `code` and opens the notebook in `cache`.
100pub fn guest(name: &str, code: &str, url: &str, cache: &Path) -> (Arc<Guest>, Notebook) {
101 let welcome = share::join(hello(name), code, "", None, Some(url)).unwrap();
102 assert_eq!(
103 (welcome.notebook.as_str(), welcome.host.as_str()),
104 ("Garden", "Ada")
105 );
106 let guest = Guest::start(
107 hello(name),
108 welcome.share,
109 welcome.secret,
110 None,
111 Some(url),
112 || {},
113 )
114 .unwrap();
115 until("the host was never met", || guest.host().is_some());
116 let notebook = Notebook::open_hosted(Arc::clone(&guest), cache).unwrap();
117 (guest, notebook)
118}
119
120pub fn open(notebook: &Notebook, guest: &Arc<Guest>, path: &str, key: Option<&Key>) -> Section {
121 let replica = notebook.replica_path(path).unwrap();
122 std::fs::create_dir_all(replica.parent().unwrap()).unwrap();
123 let replica = Replica::open_or_create(&replica, key, || notebook.read_section(path)).unwrap();
124 Section::resume_hosted(path.into(), replica, Arc::clone(guest), || {}).unwrap()
125}
126
127pub fn replace(section: &Section, image: &[u8], range: std::ops::Range<u32>, with: &str) -> u64 {
128 let (space, text, _) = server::text(image);
129 replaced(section, space, text, range, with)
130}
131
132pub fn replaced(
133 section: &Section,
134 space: ExGuid,
135 text: ExGuid,
136 range: std::ops::Range<u32>,
137 with: &str,
138) -> u64 {
139 let op = PageOp::Text {
140 text,
141 range,
142 with: with.into(),
143 };
144 let edit = Edit {
145 at: 134_000_000_000_000_000,
146 ops: vec![Op::Page { space, op }],
147 };
148 section.replica().apply("Guest", edit).unwrap()
149}
150
151pub fn published(section: &Section, id: u64) {
152 until("the edit was never published", || {
153 matches!(
154 section.status(id).unwrap(),
155 Some(EditStatus::Published { .. })
156 )
157 });
158}
159
160pub struct PresenceRelay {
161 pub url: String,
162 pub state: Arc<std::sync::Mutex<RelayState>>,
163 thread: Option<thread::JoinHandle<()>>,
164}
165
166#[derive(Default)]
167pub struct RelayState {
168 pub blocked: bool,
169 pub access_connections: usize,
170 pub access_bytes: usize,
171 sockets: Vec<(bool, std::net::TcpStream)>,
172 stopped: bool,
173}
174
175impl PresenceRelay {
176 pub fn new(upstream: &str, secret: &[u8; 16]) -> Self {
177 use sha2::{Digest, Sha256};
178 use std::{
179 io::{self, Read, Write},
180 net::{Shutdown, TcpStream},
181 sync::Mutex,
182 };
183
184 let listener = TcpListener::bind("127.0.0.1:0").unwrap();
185 let url = format!("ws://{}", listener.local_addr().unwrap());
186 let upstream = upstream.strip_prefix("ws://").unwrap().to_owned();
187 let tag: String = Sha256::digest([&b"Snowbound room "[..], secret].concat())[..8]
188 .iter()
189 .map(|byte| format!("{byte:02x}"))
190 .collect();
191 let path = format!("/v1/room/{tag}");
192 let state = Arc::new(Mutex::new(RelayState::default()));
193 let serving = Arc::clone(&state);
194 let thread = thread::spawn(move || {
195 let mut workers = Vec::new();
196 for client in listener.incoming().flatten() {
197 if serving.lock().unwrap().stopped {
198 break;
199 }
200 let state = Arc::clone(&serving);
201 let upstream = upstream.clone();
202 let path = path.clone();
203 workers.push(thread::spawn(move || {
204 let _ = (|| -> io::Result<()> {
205 let mut client = client;
206 client.set_read_timeout(Some(Duration::from_secs(5)))?;
207 let head = relay::ws::head(&mut client)?;
208 let presence = head
209 .split(' ')
210 .nth(1)
211 .map(|target| target.split('?').next().unwrap())
212 == Some(path.as_str());
213 let mut upstream = TcpStream::connect(upstream)?;
214 {
215 let mut state = state.lock().unwrap();
216 if state.stopped {
217 return Ok(());
218 }
219 if presence && state.blocked {
220 client.write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\n\r\n")?;
221 return Ok(());
222 }
223 state.sockets.push((presence, client.try_clone()?));
224 state.sockets.push((presence, upstream.try_clone()?));
225 if !presence {
226 state.access_connections += 1;
227 }
228 }
229 upstream.write_all(head.as_bytes())?;
230 client.set_read_timeout(None)?;
231 let (mut from, mut to) = (client.try_clone()?, upstream.try_clone()?);
232 let counted = Arc::clone(&state);
233 let requests = thread::spawn(move || {
234 let mut bytes = [0; 16 << 10];
235 while let Ok(length) = from.read(&mut bytes) {
236 if length == 0 {
237 break;
238 }
239 if !presence {
240 counted.lock().unwrap().access_bytes += length;
241 }
242 if to.write_all(&bytes[..length]).is_err() {
243 break;
244 }
245 }
246 let _ = to.shutdown(Shutdown::Both);
247 });
248 let _ = io::copy(&mut upstream, &mut client);
249 let _ = client.shutdown(Shutdown::Both);
250 let _ = requests.join();
251 Ok(())
252 })();
253 }));
254 }
255 for worker in workers {
256 let _ = worker.join();
257 }
258 });
259 Self {
260 url,
261 state,
262 thread: Some(thread),
263 }
264 }
265
266 pub fn disconnect_presence(&self) {
267 let mut state = self.state.lock().unwrap();
268 state.blocked = true;
269 state.sockets.retain(|(presence, socket)| {
270 if *presence {
271 let _ = socket.shutdown(std::net::Shutdown::Both);
272 }
273 !*presence
274 });
275 }
276}
277
278impl Drop for PresenceRelay {
279 fn drop(&mut self) {
280 {
281 let mut state = self.state.lock().unwrap();
282 state.stopped = true;
283 for (_, socket) in &state.sockets {
284 let _ = socket.shutdown(std::net::Shutdown::Both);
285 }
286 }
287 let _ = std::net::TcpStream::connect(self.url.strip_prefix("ws://").unwrap());
288 if let Some(thread) = self.thread.take() {
289 let _ = thread.join();
290 }
291 }
292}