arti_rpcserver/
codecs.rs

1//! Helper types for framing Json objects into async read/writes
2
3use std::marker::PhantomData;
4
5use bytes::BytesMut;
6use serde::Serialize;
7
8/// As JsonCodec, but only supports encoding, and places a newline after every
9/// object.
10#[derive(Clone)]
11pub(crate) struct JsonLinesEncoder<T> {
12    /// We consume objects of type T.
13    _phantom: PhantomData<fn(T) -> ()>,
14}
15
16impl<T> Default for JsonLinesEncoder<T> {
17    fn default() -> Self {
18        Self {
19            _phantom: PhantomData,
20        }
21    }
22}
23
24impl<T> asynchronous_codec::Encoder for JsonLinesEncoder<T>
25where
26    T: Serialize + 'static,
27{
28    type Item<'a> = T;
29
30    type Error = asynchronous_codec::JsonCodecError;
31
32    fn encode(&mut self, item: Self::Item<'_>, dst: &mut BytesMut) -> Result<(), Self::Error> {
33        use std::fmt::Write as _;
34        let j = serde_json::to_string(&item)?;
35        // The jsonlines format won't work if serde_json starts adding newlines in the middle.
36        debug_assert!(!j.contains('\n'));
37        writeln!(dst, "{}", j).expect("write! of string on BytesMut failed");
38        Ok(())
39    }
40}
41
42#[cfg(test)]
43mod test {
44    // @@ begin test lint list maintained by maint/add_warning @@
45    #![allow(clippy::bool_assert_comparison)]
46    #![allow(clippy::clone_on_copy)]
47    #![allow(clippy::dbg_macro)]
48    #![allow(clippy::mixed_attributes_style)]
49    #![allow(clippy::print_stderr)]
50    #![allow(clippy::print_stdout)]
51    #![allow(clippy::single_char_pattern)]
52    #![allow(clippy::unwrap_used)]
53    #![allow(clippy::unchecked_time_subtraction)]
54    #![allow(clippy::useless_vec)]
55    #![allow(clippy::needless_pass_by_value)]
56    #![allow(clippy::string_slice)] // See arti#2571
57    //! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
58
59    use super::*;
60    use crate::msgs::*;
61    use futures::sink::SinkExt as _;
62    use futures_await_test::async_test;
63    use tor_rpcbase as rpc;
64
65    #[derive(serde::Serialize)]
66    struct Empty {}
67
68    #[async_test]
69    async fn check_sink_basics() {
70        // Sanity-checking for our sink type.
71        let mut buf = Vec::new();
72        let r1 = BoxedResponse {
73            id: Some(RequestId::Int(7)),
74            body: ResponseBody::Update(Box::new(Empty {})),
75        };
76        let r2 = BoxedResponse {
77            id: Some(RequestId::Int(8)),
78            body: ResponseBody::Error(Box::new(rpc::RpcError::from(
79                crate::connection::RequestCancelled,
80            ))),
81        };
82        let r3 = BoxedResponse {
83            id: Some(RequestId::Int(9)),
84            body: ResponseBody::Success(Box::new(Empty {})),
85        };
86
87        // These should get serialized as follows.
88        let mut expect = String::new();
89        expect.extend(serde_json::to_string(&r1));
90        expect.push('\n');
91        expect.extend(serde_json::to_string(&r2));
92        expect.push('\n');
93        expect.extend(serde_json::to_string(&r3));
94        expect.push('\n');
95
96        {
97            let mut sink =
98                asynchronous_codec::FramedWrite::new(&mut buf, JsonLinesEncoder::default());
99            sink.send(r1).await.unwrap();
100            sink.send(r2).await.unwrap();
101            sink.send(r3).await.unwrap();
102        }
103        // Exactly 3 messages means exactly 3 newlines.
104        assert_eq!(buf.iter().filter(|c| **c == b'\n').count(), 3);
105        // Make sure that the output is what we expected.
106        assert_eq!(std::str::from_utf8(&buf).unwrap(), &expect);
107    }
108}
arti_rpcserver/codecs.rs

arti_rpcserver/
codecs.rs
