git_policy_cli/service/
service_reply.rs

1//! What:
2//!  The service's side of one waiting client: the queue of messages the landing thread answers with, and the loop that writes
3//!  them to the client's connection together with a heartbeat while nothing else is to say.
4//! Why:
5//!  A client waits for its answer with no clock of its own (`service_client.rs`), so the only thing that ends its wait is
6//!  silence. The thread that read the request has nothing else to do while the request is queued or landing, so it writes
7//!  the heartbeats, and a long batch, a long queue or a slow disk in front of the landing never reads as a dead service. The
8//!  landing thread hands its messages to that same thread through a channel instead of writing the connection itself, so the
9//!  connection has one writer, a line is written whole, and no lock guards it.
10//!
11//! In TS you'd write (pseudocode):
12//! ```ts
13//! const [reply, messages] = makeReply(); queue.push({ request, reply }); await writeUntilAnswered(socket, messages, HEARTBEAT_EVERY);
14//! ```
15
16use crate::service_protocol::{ServiceMessage, encode_service};
17use std::io::Write;
18use std::os::unix::net::UnixStream;
19use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
20use std::time::Duration;
21
22/// What:
23///  The landing thread's end of one client's messages.
24/// Why:
25///   Dropping it is what answering the request ends with: the writer sees the channel closed, writes what is left, and closes
26///  the connection.
27///
28/// In TS you'd write (pseudocode):
29/// ```ts
30/// class Reply { constructor(private queue: Channel<ServiceMessage>) {} send(message) { this.queue.push(message) } }
31/// ```
32pub struct Reply {
33    /// Where the messages for the client go.
34    sender: Sender<ServiceMessage>,
35}
36
37impl Reply {
38    /// A reply and the end the writer loop reads.
39    pub fn new() -> (Self, Receiver<ServiceMessage>) {
40        let (sender, receiver) = channel();
41        return (Self { sender }, receiver);
42    }
43
44    /// Queues `message` for the client; a client that went away has nothing to be told.
45    pub fn send(&self, message: &ServiceMessage) {
46        let _queued = self.sender.send(message.clone());
47    }
48}
49
50/// Writes `message` and a newline to `stream` as one write; false when the client is gone.
51fn write_line(stream: &UnixStream, message: &ServiceMessage) -> bool {
52    let mut line: String = encode_service(message);
53    line.push('\n');
54    let mut writer: &UnixStream = stream;
55    return writer.write_all(line.as_bytes()).is_ok();
56}
57
58/// What:
59///  Writes each queued message to the client, and a heartbeat whenever `every` passes with nothing queued, until the reply is
60///  dropped or the client is gone.
61/// Why:
62///   The heartbeat is the proof of life the client's wait rests on; the queued messages come first, in order, so a beat never
63///  stands between a notice and the answer that follows it.
64///
65/// In TS you'd write (pseudocode):
66/// ```ts
67/// for (;;) { const m = await messages.next({ timeout: every }); if (m === END) return; if (!write(m ?? heartbeat)) return; }
68/// ```
69pub fn write_until_answered(stream: &UnixStream, messages: &Receiver<ServiceMessage>, every: Duration) {
70    loop {
71        match messages.recv_timeout(every) {
72            Ok(message) => {
73                if !write_line(stream, &message) {
74                    return;
75                }
76            }
77            Err(RecvTimeoutError::Timeout) => {
78                if !write_line(stream, &ServiceMessage::Heartbeat) {
79                    return;
80                }
81            }
82            Err(RecvTimeoutError::Disconnected) => return,
83        }
84    }
85}
86
87/// The reply and its beat are controlled in the test module.
88#[cfg(test)]
89#[path = "service_reply_tests.rs"]
90mod tests;
git_policy_cli/service/service_reply.rs

git_policy_cli/service/
service_reply.rs
