rust-ipfs 0.15.0

IPFS node implementation
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

use std::io;
use std::iter;

use asynchronous_codec::{FramedRead, FramedWrite};
use connexa::prelude::transport::upgrade::InboundUpgrade;
use connexa::prelude::transport::upgrade::OutboundUpgrade;
use connexa::prelude::transport::upgrade::UpgradeInfo;
use connexa::prelude::StreamProtocol;
use futures::{future::BoxFuture, AsyncRead, AsyncWrite, SinkExt, StreamExt};

use super::{bitswap_pb, message::BitswapMessage};

const PROTOCOL: StreamProtocol = StreamProtocol::new("/ipfs/bitswap/1.2.0");
const MAX_BUF_SIZE: usize = 2_097_152;

#[derive(Debug, Clone, Default)]
pub struct BitswapProtocol;

impl UpgradeInfo for BitswapProtocol {
    type Info = StreamProtocol;
    type InfoIter = iter::Once<Self::Info>;

    fn protocol_info(&self) -> Self::InfoIter {
        iter::once(PROTOCOL)
    }
}

impl<TSocket> InboundUpgrade<TSocket> for BitswapProtocol
where
    TSocket: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
    type Output = BitswapMessage;
    type Error = io::Error;
    type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;

    fn upgrade_inbound(self, socket: TSocket, _: Self::Info) -> Self::Future {
        Box::pin(async move {
            let mut framed = FramedRead::new(
                socket,
                quick_protobuf_codec::Codec::<bitswap_pb::Message>::new(MAX_BUF_SIZE),
            );

            let message = framed
                .next()
                .await
                .ok_or_else(|| std::io::Error::from(io::ErrorKind::UnexpectedEof))??;

            let message = BitswapMessage::from_proto(message).map_err(|e| {
                tracing::error!(error = %e, "unable to parse bitswap message");
                e
            })?;

            Ok(message)
        })
    }
}

impl UpgradeInfo for BitswapMessage {
    type Info = StreamProtocol;
    type InfoIter = iter::Once<Self::Info>;

    fn protocol_info(&self) -> Self::InfoIter {
        iter::once(PROTOCOL)
    }
}

impl<TSocket> OutboundUpgrade<TSocket> for BitswapMessage
where
    TSocket: AsyncRead + AsyncWrite + Send + Unpin + 'static,
{
    type Output = ();
    type Error = io::Error;
    type Future = BoxFuture<'static, Result<Self::Output, Self::Error>>;

    #[inline]
    fn upgrade_outbound(self, socket: TSocket, _info: Self::Info) -> Self::Future {
        Box::pin(async move {
            let mut framed = FramedWrite::new(
                socket,
                quick_protobuf_codec::Codec::<bitswap_pb::Message>::new(MAX_BUF_SIZE),
            );

            let message = self.into_proto()?;

            framed.send(message).await?;
            framed.close().await?;
            Ok(())
        })
    }
}

#[derive(Debug)]
pub enum Message {
    Receive { message: BitswapMessage },
    Sent,
}

impl From<BitswapMessage> for Message {
    #[inline]
    fn from(message: BitswapMessage) -> Self {
        Message::Receive { message }
    }
}

impl From<()> for Message {
    #[inline]
    fn from(_: ()) -> Self {
        Message::Sent
    }
}