@@ -9,19 +9,17 @@ use std::{
99
1010use libipld:: { store:: StoreParams , Cid } ;
1111use libp2p:: {
12- core:: ConnectedPoint ,
12+ core:: { ConnectedPoint , Endpoint } ,
1313 futures:: channel:: oneshot,
1414 multiaddr:: Protocol ,
15- request_response:: handler:: RequestResponseHandlerEvent ,
1615 swarm:: {
17- derive_prelude:: { ConnectionId , FromSwarm } ,
18- ConnectionHandler , IntoConnectionHandler , NetworkBehaviour , NetworkBehaviourAction ,
19- PollParameters ,
16+ derive_prelude:: FromSwarm , ConnectionDenied , ConnectionId , NetworkBehaviour , THandler ,
17+ THandlerInEvent , THandlerOutEvent , ToSwarm ,
2018 } ,
2119 Multiaddr , PeerId ,
2220} ;
2321use libp2p_bitswap:: { Bitswap , BitswapConfig , BitswapEvent , BitswapResponse , BitswapStore } ;
24- use log:: warn ;
22+ use log:: debug ;
2523use prometheus:: Registry ;
2624
2725use crate :: {
@@ -54,6 +52,7 @@ pub enum Event {
5452/// This is only raised if we are tracking rate limits. The service has to
5553/// do the forwarding between the two oneshot channels, and call this module
5654/// back between doing so.
55+ #[ allow( dead_code) ]
5756 BitswapForward {
5857 peer_id : PeerId ,
5958 /// Receive response from the [`Bitswap`] behaviour.
@@ -140,12 +139,14 @@ impl<P: StoreParams> Behaviour<P> {
140139/// The underlying [`libp2p_request_response::RequestResponse`] behaviour
141140/// will initiate connections to the peers which aren't connected at the moment.
142141pub fn resolve ( & mut self , cid : Cid , peers : Vec < PeerId > ) -> QueryId {
142+ debug ! ( "resolving {cid} from {peers:?}" ) ;
143143 stats:: CONTENT_RESOLVE_RUNNING . inc ( ) ;
144144 // Not passing any missing items, which will result in a call to `BitswapStore::missing_blocks`.
145145 self . inner . sync ( cid, peers, [ ] . into_iter ( ) )
146146 }
147147
148148 /// Check whether the peer has already exhaused their rate limit.
149+ #[ allow( dead_code) ]
149150 fn check_rate_limit ( & mut self , peer_id : & PeerId , cid : & Cid ) -> bool {
150151 if let Some ( ref rate_limit) = self . rate_limit {
151152 if let Some ( addr) = self . peer_addresses . get ( peer_id) . cloned ( ) {
@@ -181,17 +182,9 @@ impl<P: StoreParams> Behaviour<P> {
181182
182183impl < P : StoreParams > NetworkBehaviour for Behaviour < P > {
183184 type ConnectionHandler = <Bitswap < P > as NetworkBehaviour >:: ConnectionHandler ;
184- type OutEvent = Event ;
185+ type ToSwarm = Event ;
185186
186- fn new_handler ( & mut self ) -> Self :: ConnectionHandler {
187- self . inner . new_handler ( )
188- }
189-
190- fn addresses_of_peer ( & mut self , peer_id : & PeerId ) -> Vec < Multiaddr > {
191- self . inner . addresses_of_peer ( peer_id)
192- }
193-
194- fn on_swarm_event ( & mut self , event : FromSwarm < Self :: ConnectionHandler > ) {
187+ fn on_swarm_event ( & mut self , event : FromSwarm ) {
195188 // Store the remote address.
196189 match & event {
197190 FromSwarm :: ConnectionEstablished ( c) => {
@@ -226,61 +219,124 @@ impl<P: StoreParams> NetworkBehaviour for Behaviour<P> {
226219 & mut self ,
227220 peer_id : PeerId ,
228221 connection_id : ConnectionId ,
229- event : << Self :: ConnectionHandler as IntoConnectionHandler > :: Handler as ConnectionHandler > :: OutEvent ,
222+ event : THandlerOutEvent < Self > ,
230223 ) {
231- match event {
232- RequestResponseHandlerEvent :: Request {
233- request_id,
234- request,
235- sender,
236- } if self . rate_limit . is_some ( ) => {
237- if !self . check_rate_limit ( & peer_id, & request. cid ) {
238- warn ! ( "rate limiting {peer_id}" ) ;
239- stats:: CONTENT_RATE_LIMITED . inc ( ) ;
240- return ;
241- }
242- // We need to hijack the response channel to record the size, otherwise it goes straight to the handler.
243- let ( tx, rx) = libp2p:: futures:: channel:: oneshot:: channel ( ) ;
244- let event = RequestResponseHandlerEvent :: Request {
245- request_id,
246- request,
247- sender : tx,
248- } ;
249-
250- self . inner
251- . on_connection_handler_event ( peer_id, connection_id, event) ;
252-
253- let forward = Event :: BitswapForward {
254- peer_id,
255- response_rx : rx,
256- response_tx : sender,
257- } ;
258- self . outbox . push_back ( forward) ;
259- }
260- _ => self
261- . inner
262- . on_connection_handler_event ( peer_id, connection_id, event) ,
263- }
224+ // TODO: `request_response::handler` is now private, so we cannot pattern match on the handler event.
225+ // By the looks of the only way to access the request event is to let it go right into the RR protocol
226+ // wrapped by the Bitswap behaviour and let it raise an event, however we will not see that event here.
227+ // I'm not sure what we can do without moving rate limiting into the bitswap library itself, because
228+ // what we did here relied on the ability to redirect the channels inside the request, but if the event
229+ // itself is private to the `request_response` protocol there's nothing I can do.
230+ // match event {
231+
232+ // request_response::handler::Event::Request {
233+ // request_id,
234+ // request,
235+ // sender,
236+ // } if self.rate_limit.is_some() => {
237+ // if !self.check_rate_limit(&peer_id, &request.cid) {
238+ // warn!("rate limiting {peer_id}");
239+ // stats::CONTENT_RATE_LIMITED.inc();
240+ // return;
241+ // }
242+ // // We need to hijack the response channel to record the size, otherwise it goes straight to the handler.
243+ // let (tx, rx) = libp2p::futures::channel::oneshot::channel();
244+ // let event = request_response::Event::Request {
245+ // request_id,
246+ // request,
247+ // sender: tx,
248+ // };
249+
250+ // self.inner
251+ // .on_connection_handler_event(peer_id, connection_id, event);
252+
253+ // let forward = Event::BitswapForward {
254+ // peer_id,
255+ // response_rx: rx,
256+ // response_tx: sender,
257+ // };
258+ // self.outbox.push_back(forward);
259+ // }
260+ // _ => self
261+ // .inner
262+ // .on_connection_handler_event(peer_id, connection_id, event),
263+ // }
264+
265+ // debug!("BITSWAP CONNECTION HANDLER EVENT: {event:?}");
266+
267+ self . inner
268+ . on_connection_handler_event ( peer_id, connection_id, event)
269+ }
270+
271+ fn handle_pending_inbound_connection (
272+ & mut self ,
273+ connection_id : ConnectionId ,
274+ local_addr : & Multiaddr ,
275+ remote_addr : & Multiaddr ,
276+ ) -> Result < ( ) , ConnectionDenied > {
277+ self . inner
278+ . handle_pending_inbound_connection ( connection_id, local_addr, remote_addr)
279+ }
280+
281+ fn handle_established_inbound_connection (
282+ & mut self ,
283+ connection_id : ConnectionId ,
284+ peer : PeerId ,
285+ local_addr : & Multiaddr ,
286+ remote_addr : & Multiaddr ,
287+ ) -> Result < THandler < Self > , ConnectionDenied > {
288+ self . inner . handle_established_inbound_connection (
289+ connection_id,
290+ peer,
291+ local_addr,
292+ remote_addr,
293+ )
294+ }
295+
296+ fn handle_pending_outbound_connection (
297+ & mut self ,
298+ connection_id : ConnectionId ,
299+ maybe_peer : Option < PeerId > ,
300+ addresses : & [ Multiaddr ] ,
301+ effective_role : Endpoint ,
302+ ) -> Result < Vec < Multiaddr > , ConnectionDenied > {
303+ self . inner . handle_pending_outbound_connection (
304+ connection_id,
305+ maybe_peer,
306+ addresses,
307+ effective_role,
308+ )
309+ }
310+
311+ fn handle_established_outbound_connection (
312+ & mut self ,
313+ connection_id : ConnectionId ,
314+ peer : PeerId ,
315+ addr : & Multiaddr ,
316+ role_override : Endpoint ,
317+ ) -> Result < THandler < Self > , ConnectionDenied > {
318+ self . inner
319+ . handle_established_outbound_connection ( connection_id, peer, addr, role_override)
264320 }
265321
266322 fn poll (
267323 & mut self ,
268324 cx : & mut Context < ' _ > ,
269- params : & mut impl PollParameters ,
270- ) -> Poll < NetworkBehaviourAction < Self :: OutEvent , Self :: ConnectionHandler > > {
325+ ) -> Poll < ToSwarm < Self :: ToSwarm , THandlerInEvent < Self > > > {
271326 // Emit own events first.
272327 if let Some ( ev) = self . outbox . pop_front ( ) {
273- return Poll :: Ready ( NetworkBehaviourAction :: GenerateEvent ( ev) ) ;
328+ return Poll :: Ready ( ToSwarm :: GenerateEvent ( ev) ) ;
274329 }
275330 // Poll Bitswap.
276- while let Poll :: Ready ( ev) = self . inner . poll ( cx, params) {
331+ while let Poll :: Ready ( ev) = self . inner . poll ( cx) {
332+ // debug!("BITSWAP POLL: {ev:?}");
277333 match ev {
278- NetworkBehaviourAction :: GenerateEvent ( ev) => match ev {
334+ ToSwarm :: GenerateEvent ( ev) => match ev {
279335 BitswapEvent :: Progress ( _, _) => { }
280336 BitswapEvent :: Complete ( id, result) => {
281337 stats:: CONTENT_RESOLVE_RUNNING . dec ( ) ;
282338 let out = Event :: Complete ( id, result) ;
283- return Poll :: Ready ( NetworkBehaviourAction :: GenerateEvent ( out) ) ;
339+ return Poll :: Ready ( ToSwarm :: GenerateEvent ( out) ) ;
284340 }
285341 } ,
286342 other => {
0 commit comments