forest/libp2p_bitswap/internals/
codec.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use std::io;
5
6use async_trait::async_trait;
7use asynchronous_codec::{FramedRead, FramedWrite};
8use futures::{
9    SinkExt, StreamExt,
10    io::{AsyncRead, AsyncWrite},
11};
12use libp2p::request_response;
13use smallvec::SmallVec;
14
15use crate::libp2p_bitswap::{bitswap_pb::mod_Message::BlockPresenceType, prefix::Prefix, *};
16
17// 2MB Block Size according to the specs at https://github.com/ipfs/specs/blob/main/BITSWAP.md
18const MAX_BUF_SIZE: usize = 1024 * 1024 * 2;
19
20/// The payload of a `bitswap` request. Outbound requests always carry exactly one message
21/// (enforced by the `assert_eq!` in `write_request`) and inbound requests are typically
22/// decoded into just a few parts, so the common case is kept off the heap with inline storage.
23pub type BitswapMessages = SmallVec<[BitswapMessage; 1]>;
24
25fn codec() -> quick_protobuf_codec::Codec<bitswap_pb::Message> {
26    quick_protobuf_codec::Codec::<bitswap_pb::Message>::new(MAX_BUF_SIZE)
27}
28
29#[derive(Default, Debug, Clone)]
30pub struct BitswapRequestResponseCodec;
31
32#[async_trait]
33impl request_response::Codec for BitswapRequestResponseCodec {
34    type Protocol = &'static str;
35    type Request = BitswapMessages;
36    type Response = ();
37
38    async fn read_request<T>(&mut self, _: &Self::Protocol, io: &mut T) -> IOResult<Self::Request>
39    where
40        T: AsyncRead + Send + Unpin,
41    {
42        let pb_msg: bitswap_pb::Message = FramedRead::new(io, codec())
43            .next()
44            .await
45            .ok_or(std::io::ErrorKind::UnexpectedEof)??;
46
47        metrics::inbound_stream_count().inc();
48
49        let mut parts = BitswapMessages::new();
50        for entry in pb_msg.wantlist.unwrap_or_default().entries {
51            let cid = Cid::try_from(entry.block).map_err(io::Error::other)?;
52            parts.push(BitswapMessage::Request(BitswapRequest {
53                ty: entry.wantType.into(),
54                cid,
55                send_dont_have: entry.sendDontHave,
56                cancel: entry.cancel,
57            }));
58        }
59
60        for payload in pb_msg.payload {
61            let prefix = Prefix::new(&payload.prefix).map_err(io::Error::other)?;
62            let cid = prefix.to_cid(&payload.data).map_err(io::Error::other)?;
63            parts.push(BitswapMessage::Response(
64                cid,
65                BitswapResponse::Block(payload.data),
66            ));
67        }
68
69        for presence in pb_msg.blockPresences {
70            let cid = Cid::try_from(presence.cid).map_err(io::Error::other)?;
71            let have = presence.type_pb == BlockPresenceType::Have;
72            parts.push(BitswapMessage::Response(cid, BitswapResponse::Have(have)));
73        }
74
75        Ok(parts)
76    }
77
78    /// Just close the outbound stream,
79    /// the actual responses will come from new inbound stream
80    /// and be received in `read_request`
81    async fn read_response<T>(&mut self, _: &Self::Protocol, _: &mut T) -> IOResult<Self::Response>
82    where
83        T: AsyncRead + Send + Unpin,
84    {
85        Ok(())
86    }
87
88    /// Sending both `bitswap` requests and responses
89    async fn write_request<T>(
90        &mut self,
91        _: &Self::Protocol,
92        io: &mut T,
93        mut messages: Self::Request,
94    ) -> IOResult<()>
95    where
96        T: AsyncWrite + Send + Unpin,
97    {
98        assert_eq!(
99            messages.len(),
100            1,
101            "It's only supported to send a single message" // libp2p-bitswap doesn't support batch sending
102        );
103
104        let data = messages.swap_remove(0).into_proto()?;
105        let mut framed = FramedWrite::new(io, codec());
106        framed.send(data).await?;
107        framed.close().await?;
108
109        metrics::outbound_stream_count().inc();
110
111        Ok(())
112    }
113
114    // Sending `FIN` header and close the stream
115    async fn write_response<T>(
116        &mut self,
117        _: &Self::Protocol,
118        _: &mut T,
119        _: Self::Response,
120    ) -> IOResult<()>
121    where
122        T: AsyncWrite + Send + Unpin,
123    {
124        Ok(())
125    }
126}
forest/libp2p_bitswap/internals/codec.rs

forest/libp2p_bitswap/internals/
codec.rs
