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    write_until_answered_noting(stream, messages, every, &mut || { return None; });
71}
72
73/// What:
74///  `write_until_answered`, where `note` is asked for a message whenever the beat is due, and what it returns is written before the
75///  beat.
76/// Why:
77///   A client queued behind a write that never ends is told what the service is stuck in (`service_busy.rs`); the beat is the
78///  one moment this thread has nothing queued to write, so the notice rides on it and is asked for at most once a beat.
79///
80/// In TS you'd write (pseudocode):
81/// ```ts
82/// for (;;) { const m = await messages.next({ timeout: every }); if (m === END) return; if (m === undefined) { const n = note(); if (n && !write(n)) return; } if (!write(m ?? heartbeat)) return; }
83/// ```
84pub fn write_until_answered_noting(
85    stream: &UnixStream,
86    messages: &Receiver<ServiceMessage>,
87    every: Duration,
88    note: &mut dyn FnMut() -> Option<ServiceMessage>,
89) {
90    loop {
91        match messages.recv_timeout(every) {
92            Ok(message) => {
93                if !write_line(stream, &message) {
94                    return;
95                }
96            }
97            Err(RecvTimeoutError::Timeout) => {
98                if let Some(notice) = note()
99                    && !write_line(stream, &notice)
100                {
101                    return;
102                }
103                if !write_line(stream, &ServiceMessage::Heartbeat) {
104                    return;
105                }
106            }
107            Err(RecvTimeoutError::Disconnected) => return,
108        }
109    }
110}
111
112/// The reply and its beat are controlled in the test module.
113#[cfg(test)]
114#[path = "service_reply_tests.rs"]
115mod tests;
git_policy_cli/service/service_reply.rs

git_policy_cli/service/
service_reply.rs
