link_cli/protocol/
server.rs

1//! A TCP server exposing a links store through the LiNo protocols.
2//!
3//! Each connection gets its own thread that only parses and formats
4//! messages; every request is executed by the thread that called
5//! [`LinksServer::serve`], one at a time, so the store needs neither `Send`
6//! nor locking. The protocol is detected per message and the reply uses the
7//! same protocol, so text and binary clients can share one server.
8
9use super::error::ProtocolResult;
10use super::format::format_document;
11use super::links_operations::LinksOperation;
12use super::mapping::LinoDocument;
13use super::packet::DecodeLimits;
14use super::protocols::{read_any_document, MessageFormat};
15use crate::link::Link;
16use crate::named_type_links::NamedTypeLinks;
17use crate::query_processor::QueryProcessor;
18use links_notation::LiNo;
19use std::io::{BufReader, BufWriter, Write};
20use std::net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
21use std::sync::atomic::{AtomicBool, Ordering};
22use std::sync::mpsc::{self, Receiver, Sender};
23use std::sync::Arc;
24use std::thread;
25
26/// Which protocols a server accepts.
27#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
28pub enum AcceptedProtocols {
29    /// Detect the protocol of every message (the default).
30    #[default]
31    Any,
32    /// Only [`TextLinoProtocol`](super::TextLinoProtocol) messages.
33    Text,
34    /// Only [`BinaryLinoProtocol`](super::BinaryLinoProtocol) messages.
35    Binary,
36}
37
38impl AcceptedProtocols {
39    fn accepts(self, format: MessageFormat) -> bool {
40        matches!(
41            (self, format),
42            (AcceptedProtocols::Any, _)
43                | (AcceptedProtocols::Text, MessageFormat::Text)
44                | (AcceptedProtocols::Binary, MessageFormat::Binary(_))
45        )
46    }
47}
48
49/// Server configuration.
50#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
51pub struct ServerOptions {
52    /// Print every request and reply to stderr.
53    pub trace: bool,
54    /// Passed to [`QueryProcessor::with_auto_create_missing_references`].
55    pub auto_create_missing_references: bool,
56    /// Protocols the server answers.
57    pub accept: AcceptedProtocols,
58    /// Limits applied to incoming messages.
59    pub limits: DecodeLimits,
60}
61
62enum Job {
63    Request {
64        document: LinoDocument,
65        reply: Sender<LinoDocument>,
66    },
67    Shutdown,
68}
69
70/// Stops a running [`LinksServer::serve`] from another thread.
71#[derive(Clone, Debug)]
72pub struct ShutdownHandle {
73    jobs: Sender<Job>,
74    stopping: Arc<AtomicBool>,
75    address: SocketAddr,
76}
77
78impl ShutdownHandle {
79    /// Asks the server to stop; `serve` returns after the current request.
80    pub fn shutdown(&self) {
81        self.stopping.store(true, Ordering::SeqCst);
82        let _ = self.jobs.send(Job::Shutdown);
83        // Wake the accept loop so it notices the flag.
84        let _ = TcpStream::connect(self.address);
85    }
86}
87
88/// A bound, not yet serving, LiNo server.
89#[derive(Debug)]
90pub struct LinksServer {
91    listener: TcpListener,
92    options: ServerOptions,
93    jobs: Sender<Job>,
94    queue: Receiver<Job>,
95    stopping: Arc<AtomicBool>,
96}
97
98impl LinksServer {
99    /// Binds to `address` (use port `0` for an ephemeral port).
100    pub fn bind(address: impl ToSocketAddrs, options: ServerOptions) -> ProtocolResult<Self> {
101        let listener = TcpListener::bind(address)?;
102        let (jobs, queue) = mpsc::channel();
103        Ok(Self {
104            listener,
105            options,
106            jobs,
107            queue,
108            stopping: Arc::new(AtomicBool::new(false)),
109        })
110    }
111
112    /// The address the server listens on.
113    pub fn local_addr(&self) -> ProtocolResult<SocketAddr> {
114        Ok(self.listener.local_addr()?)
115    }
116
117    /// A handle that stops [`serve`](Self::serve).
118    pub fn shutdown_handle(&self) -> ProtocolResult<ShutdownHandle> {
119        Ok(ShutdownHandle {
120            jobs: self.jobs.clone(),
121            stopping: Arc::clone(&self.stopping),
122            address: self.local_addr()?,
123        })
124    }
125
126    /// Serves `storage` until a [`ShutdownHandle`] stops the server.
127    pub fn serve<S: NamedTypeLinks>(self, storage: &mut S) -> ProtocolResult<()> {
128        let LinksServer {
129            listener,
130            options,
131            jobs,
132            queue,
133            stopping,
134        } = self;
135        let acceptor = {
136            let stopping = Arc::clone(&stopping);
137            thread::spawn(move || accept_loop(listener, jobs, stopping, options))
138        };
139        let processor = QueryProcessor::new(false)
140            .with_auto_create_missing_references(options.auto_create_missing_references);
141        while let Ok(job) = queue.recv() {
142            match job {
143                Job::Shutdown => break,
144                Job::Request { document, reply } => {
145                    if options.trace {
146                        eprintln!("[server] request: {}", format_document(&document));
147                    }
148                    let response = execute_request(storage, &processor, &document);
149                    if options.trace {
150                        eprintln!("[server] reply: {}", format_document(&response));
151                    }
152                    let _ = reply.send(response);
153                }
154            }
155        }
156        stopping.store(true, Ordering::SeqCst);
157        drop(queue);
158        let _ = acceptor.join();
159        Ok(())
160    }
161}
162
163fn accept_loop(
164    listener: TcpListener,
165    jobs: Sender<Job>,
166    stopping: Arc<AtomicBool>,
167    options: ServerOptions,
168) {
169    for stream in listener.incoming() {
170        if stopping.load(Ordering::SeqCst) {
171            return;
172        }
173        let Ok(stream) = stream else { continue };
174        let jobs = jobs.clone();
175        thread::spawn(move || {
176            if let Err(error) = handle_connection(stream, jobs, options) {
177                if options.trace {
178                    eprintln!("[server] connection closed: {error}");
179                }
180            }
181        });
182    }
183}
184
185fn handle_connection(
186    stream: TcpStream,
187    jobs: Sender<Job>,
188    options: ServerOptions,
189) -> ProtocolResult<()> {
190    let _ = stream.set_nodelay(true);
191    let mut writer = BufWriter::new(stream.try_clone()?);
192    let mut reader = BufReader::new(stream);
193    loop {
194        let (document, format) = match read_any_document(&mut reader, &options.limits) {
195            Ok(Some(message)) => message,
196            Ok(None) => {
197                if options.trace {
198                    eprintln!("[server] client hung up");
199                }
200                return Ok(());
201            }
202            Err(error) => {
203                // The stream may be out of sync; answer in text and hang up.
204                let reply = error_document(&error.to_string());
205                let _ = MessageFormat::Text
206                    .protocol(options.limits)
207                    .write_document(&mut writer, &reply);
208                return Err(error);
209            }
210        };
211        let response = if options.accept.accepts(format) {
212            let (reply, response) = mpsc::channel();
213            // Once `serve` returns, no request may touch the store.
214            jobs.send(Job::Request { document, reply })
215                .ok()
216                .and_then(|()| response.recv().ok())
217                .unwrap_or_else(|| error_document("server is shutting down"))
218        } else {
219            error_document("this server does not accept this protocol")
220        };
221        format
222            .protocol(options.limits)
223            .write_document(&mut writer, &response)?;
224        writer.flush()?;
225    }
226}
227
228/// The reply document for a failed request: `(error: 'message')`.
229pub fn error_document(message: &str) -> LinoDocument {
230    vec![LiNo::Link {
231        id: Some("error".to_string()),
232        values: vec![LiNo::Ref(message.to_string())],
233    }]
234}
235
236/// Returns the message of an `(error: 'message')` reply.
237pub fn error_message(document: &[LiNo<String>]) -> Option<&str> {
238    match document {
239        [LiNo::Link {
240            id: Some(id),
241            values,
242        }] if id == "error" => match values.as_slice() {
243            [LiNo::Ref(message)] => Some(message),
244            _ => None,
245        },
246        _ => None,
247    }
248}
249
250/// Executes one request against `storage`.
251///
252/// A non-empty document is a substitution query; the reply holds one
253/// `(before) (after)` line per change, exactly like `clink --changes`. The
254/// empty document asks for every link, one `(index: source target)` per line.
255/// Failures produce an [`error_document`].
256pub fn execute_request<S: NamedTypeLinks>(
257    storage: &mut S,
258    processor: &QueryProcessor,
259    document: &[LiNo<String>],
260) -> LinoDocument {
261    match try_execute_request(storage, processor, document) {
262        Ok(reply) => reply,
263        Err(error) => error_document(&format!("{error:#}")),
264    }
265}
266
267fn try_execute_request<S: NamedTypeLinks>(
268    storage: &mut S,
269    processor: &QueryProcessor,
270    document: &[LiNo<String>],
271) -> anyhow::Result<LinoDocument> {
272    if let Some(operation) = LinksOperation::from_document(document)? {
273        return operation.execute(storage);
274    }
275    if document.is_empty() {
276        let mut links = storage.all_links();
277        links.sort_by_key(|link| link.index);
278        return links.iter().map(|link| link_lino(storage, link)).collect();
279    }
280    let changes = processor.process_query(storage, &format_document(document))?;
281    if !changes.is_empty() {
282        storage.save()?;
283    }
284    changes
285        .iter()
286        .map(|(before, after)| {
287            Ok(LiNo::Link {
288                id: None,
289                values: vec![
290                    change_side(storage, before.as_ref())?,
291                    change_side(storage, after.as_ref())?,
292                ],
293            })
294        })
295        .collect()
296}
297
298/// `(index: source target)`, naming every reference that has a name.
299fn link_lino<S: NamedTypeLinks>(storage: &mut S, link: &Link) -> anyhow::Result<LiNo<String>> {
300    Ok(LiNo::Link {
301        id: Some(reference_name(storage, link.index)?),
302        values: vec![
303            LiNo::Ref(reference_name(storage, link.source)?),
304            LiNo::Ref(reference_name(storage, link.target)?),
305        ],
306    })
307}
308
309/// `()` for a missing side of a change, `((index: source target))` otherwise.
310fn change_side<S: NamedTypeLinks>(
311    storage: &mut S,
312    link: Option<&Link>,
313) -> anyhow::Result<LiNo<String>> {
314    let values = match link {
315        Some(link) => vec![link_lino(storage, link)?],
316        None => Vec::new(),
317    };
318    Ok(LiNo::Link { id: None, values })
319}
320
321fn reference_name<S: NamedTypeLinks>(storage: &mut S, id: u32) -> anyhow::Result<String> {
322    Ok(storage.get_name(id)?.unwrap_or_else(|| id.to_string()))
323}
link_cli/protocol/server.rs

link_cli/protocol/
server.rs
