Implement bitswap in the network behaviour using libp2p_bitswap.

bin/node-template/node/src/service.rs

@@ -87,6 +87,7 @@ pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &config,
 			client: client.clone(),
+			backend: backend.clone(),
 			transaction_pool: transaction_pool.clone(),
 			spawn_handle: task_manager.spawn_handle(),
 			import_queue,
@@ -242,6 +243,7 @@ pub fn new_light(config: Configuration) -> Result<TaskManager, ServiceError> {
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &config,
 			client: client.clone(),
+			backend: backend.clone(),
 			transaction_pool: transaction_pool.clone(),
 			spawn_handle: task_manager.spawn_handle(),
 			import_queue,

bin/node/cli/src/service.rs

@@ -169,6 +169,7 @@ pub fn new_full_base(
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &config,
 			client: client.clone(),
+			backend: backend.clone(),
 			transaction_pool: transaction_pool.clone(),
 			spawn_handle: task_manager.spawn_handle(),
 			import_queue,
@@ -387,6 +388,7 @@ pub fn new_light_base(config: Configuration) -> Result<(
 		sc_service::build_network(sc_service::BuildNetworkParams {
 			config: &config,
 			client: client.clone(),
+			backend: backend.clone(),
 			transaction_pool: transaction_pool.clone(),
 			spawn_handle: task_manager.spawn_handle(),
 			import_queue,

client/network-gossip/src/bridge.rs

@@ -210,7 +210,8 @@ impl<B: BlockT> Future for GossipEngine<B> {
 
 								this.forwarding_state = ForwardingState::Busy(to_forward.into());
 							},
-							Event::Dht(_) => {}
+							Event::Dht(_) => {},
+							Event::Bitswap(_) => {},
 						}
 						// The network event stream closed. Do the same for [`GossipValidator`].
 						Poll::Ready(None) => return Poll::Ready(()),

client/network/Cargo.toml

@@ -59,6 +59,7 @@ unsigned-varint = { version = "0.4.0", features = ["futures", "futures-codec"] }
 void = "1.0.2"
 wasm-timer = "0.2"
 zeroize = "1.0.0"
+libp2p-bitswap = "0.5.1"
 
 [dependencies.libp2p]
 version = "0.22.0"

client/network/src/behaviour.rs

@@ -50,6 +50,8 @@ pub struct Behaviour<B: BlockT, H: ExHashT> {
 	peer_info: peer_info::PeerInfoBehaviour,
 	/// Discovers nodes of the network.
 	discovery: DiscoveryBehaviour,
+	/// Exchanges blocks of data with other nodes.
+	pub bitswap: libp2p_bitswap::Bitswap,
 	/// Block request handling.
 	block_requests: block_requests::BlockRequests<B>,
 	/// Finality proof request handling.
@@ -148,6 +150,9 @@ pub enum BehaviourOut<B: BlockT> {
 	/// Events generated by a DHT as a response to get_value or put_value requests as well as the
 	/// request duration.
 	Dht(DhtEvent, Duration),
+
+	/// Event generated by bitswap.
+	Bitswap(libp2p_bitswap::BitswapEvent)
 }
 
 impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
@@ -170,6 +175,7 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 			finality_proof_requests,
 			light_client_handler,
 			events: VecDeque::new(),
+			bitswap: libp2p_bitswap::Bitswap::new(),
 			role,
 		}
 	}
@@ -260,6 +266,11 @@ impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 	pub fn light_client_request(&mut self, r: light_client_handler::Request<B>) -> Result<(), light_client_handler::Error> {
 		self.light_client_handler.request(r)
 	}
+
+	fn add_discovered_node(&mut self, peer_id: PeerId) {
+		self.substrate.add_discovered_nodes(iter::once(peer_id.clone()));
+		self.bitswap.connect(peer_id);
+	}
 }
 
 fn reported_roles_to_observed_role(local_role: &Role, remote: &PeerId, roles: Roles) -> ObservedRole {
@@ -438,7 +449,7 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<peer_info::PeerInfoEven
 		for addr in listen_addrs {
 			self.discovery.add_self_reported_address(&peer_id, protocols.iter(), addr);
 		}
-		self.substrate.add_discovered_nodes(iter::once(peer_id));
+		self.add_discovered_node(peer_id);
 	}
 }
 
@@ -453,7 +464,7 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<DiscoveryOut>
 				// implementation for `PeerInfoEvent`.
 			}
 			DiscoveryOut::Discovered(peer_id) => {
-				self.substrate.add_discovered_nodes(iter::once(peer_id));
+				self.add_discovered_node(peer_id);
 			}
 			DiscoveryOut::ValueFound(results, duration) => {
 				self.events.push_back(BehaviourOut::Dht(DhtEvent::ValueFound(results), duration));
@@ -476,6 +487,13 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<DiscoveryOut>
 	}
 }
 
+impl<B: BlockT, H: ExHashT> NetworkBehaviourEventProcess<libp2p_bitswap::BitswapEvent>
+	for Behaviour<B, H> {
+	fn inject_event(&mut self, event: libp2p_bitswap::BitswapEvent) {
+		self.events.push_back(BehaviourOut::Bitswap(event));
+	}
+}
+
 impl<B: BlockT, H: ExHashT> Behaviour<B, H> {
 	fn poll<TEv>(&mut self, _: &mut Context, _: &mut impl PollParameters) -> Poll<NetworkBehaviourAction<TEv, BehaviourOut<B>>> {
 		if let Some(event) = self.events.pop_front() {

client/network/src/lib.rs

@@ -268,6 +268,7 @@ pub use protocol::sync::SyncState;
 pub use libp2p::{Multiaddr, PeerId};
 #[doc(inline)]
 pub use libp2p::multiaddr;
+pub use libp2p_bitswap::BitswapEvent;
 
 pub use sc_peerset::ReputationChange;
 use sp_runtime::traits::{Block as BlockT, NumberFor};

client/network/src/protocol/event.rs

@@ -45,6 +45,8 @@ pub enum DhtEvent {
 pub enum Event {
 	/// Event generated by a DHT.
 	Dht(DhtEvent),
+	/// Event generated by Bitswap.
+	Bitswap(libp2p_bitswap::BitswapEvent),
 
 	/// Opened a substream with the given node with the given notifications protocol.
 	///