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

forest/libp2p_bitswap/internals/
codec.rs
