xstack 0.2.17

libp2p stack rewritten from ground up for rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

use std::future::Future;

use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use protobuf::Message;
use rand::{thread_rng, RngCore};

use crate::{proto::identity::Identity, Error, Result};

/// An extension trait for libp2p rpc calls.
pub trait XStackRpc: AsyncRead + AsyncWrite + Unpin {
    /// make a rpc call.
    ///
    /// # Parameters
    /// - `message`, a protobuf [`Message`].
    ///
    /// On success, returns one same type [`Message`] as inputs one.
    fn xstack_call<M>(mut self, message: &M, max_recv_len: usize) -> impl Future<Output = Result<M>>
    where
        Self: Sized,
        M: Message,
    {
        async move {
            XStackRpc::xstack_send(&mut self, message).await?;

            let body_len = unsigned_varint::aio::read_usize(&mut self).await?;

            if body_len > max_recv_len {
                return Err(Error::Overflow(max_recv_len));
            }

            let mut buf = vec![0u8; body_len];

            self.read_exact(&mut buf).await?;

            Ok(M::parse_from_bytes(&buf)?)
        }
    }

    /// receive one protobuf message.
    ///
    /// # Parameters
    /// - `max_recv_len`, the maximum length of the receiving packet.
    ///
    /// On success, returns one same type [`Message`] as inputs one.
    fn xstack_recv<M>(mut self, max_recv_len: usize) -> impl Future<Output = Result<M>>
    where
        Self: Sized,
        M: Message,
    {
        async move {
            let body_len = unsigned_varint::aio::read_usize(&mut self).await?;

            if body_len > max_recv_len {
                return Err(Error::Overflow(max_recv_len));
            }

            let mut buf = vec![0u8; body_len];

            self.read_exact(&mut buf).await?;

            Ok(M::parse_from_bytes(&buf)?)
        }
    }

    /// make a rpc call. unlike [`xstack_call`](XStackRpc), this function doesn't wait a response.
    ///
    /// # Parameters
    /// - `message`, a protobuf [`Message`].
    fn xstack_send<M>(mut self, message: &M) -> impl Future<Output = Result<()>>
    where
        Self: Sized,
        M: Message,
    {
        // let mut message = HopMessage::new();

        // message.type_ = Some(circuit::hop_message::Type::RESERVE.into());

        async move {
            let buf = message.write_to_bytes()?;

            let mut payload_len = unsigned_varint::encode::usize_buffer();

            self.write_all(unsigned_varint::encode::usize(buf.len(), &mut payload_len))
                .await?;

            self.write_all(buf.as_slice()).await?;

            Ok(())
        }
    }

    /// Make a ping test via the stream.
    fn xstack_ping(mut self) -> impl Future<Output = Result<()>>
    where
        Self: Sized,
    {
        async move {
            let mut buf = vec![0u8; 32];

            thread_rng().fill_bytes(&mut buf);

            self.write_all(&buf).await?;

            let mut echo = vec![0u8; 32];

            self.read_exact(&mut echo).await?;

            if echo != buf {
                return Err(Error::Ping);
            }

            Ok(())
        }
    }

    fn xstack_recv_identity(
        mut self,
        max_packet_size: usize,
    ) -> impl Future<Output = Result<Identity>>
    where
        Self: Sized,
    {
        async move {
            log::trace!("identity_request: read varint length");

            let body_len = unsigned_varint::aio::read_usize(&mut self).await?;

            log::trace!("identity_request: read varint length");

            if max_packet_size < body_len {
                return Err(Error::Overflow(body_len));
            }

            log::trace!("identity_request recv body: {}", body_len);

            let mut buf = vec![0; body_len];

            self.read_exact(&mut buf).await?;

            Ok(Identity::parse_from_bytes(&buf)?)
        }
    }
}

impl<S> XStackRpc for S where S: AsyncRead + AsyncWrite + Unpin {}