mwc_libp2p_core/network/
peer.rs

1// Copyright 2018 Parity Technologies (UK) Ltd.
2//
3// Permission is hereby granted, free of charge, to any person obtaining a
4// copy of this software and associated documentation files (the "Software"),
5// to deal in the Software without restriction, including without limitation
6// the rights to use, copy, modify, merge, publish, distribute, sublicense,
7// and/or sell copies of the Software, and to permit persons to whom the
8// Software is furnished to do so, subject to the following conditions:
9//
10// The above copyright notice and this permission notice shall be included in
11// all copies or substantial portions of the Software.
12//
13// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
14// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
19// DEALINGS IN THE SOFTWARE.
20
21use crate::{
22    Multiaddr,
23    Transport,
24    StreamMuxer,
25    connection::{
26        Connected,
27        ConnectedPoint,
28        ConnectionHandler,
29        Connection,
30        ConnectionId,
31        ConnectionLimit,
32        EstablishedConnection,
33        EstablishedConnectionIter,
34        IntoConnectionHandler,
35        PendingConnection,
36        Substream,
37        pool::Pool,
38    },
39    PeerId
40};
41use fnv::FnvHashMap;
42use smallvec::SmallVec;
43use std::{
44    collections::hash_map,
45    error,
46    fmt,
47};
48use super::{Network, DialingOpts};
49
50/// The possible representations of a peer in a [`Network`], as
51/// seen by the local node.
52///
53/// > **Note**: In any state there may always be a pending incoming
54/// > connection attempt from the peer, however, the remote identity
55/// > of a peer is only known once a connection is fully established.
56pub enum Peer<'a, TTrans, TInEvent, TOutEvent, THandler>
57where
58    TTrans: Transport,
59    THandler: IntoConnectionHandler
60{
61    /// At least one established connection exists to the peer.
62    Connected(ConnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>),
63
64    /// There is an ongoing dialing (i.e. outgoing connection) attempt
65    /// to the peer. There may already be other established connections
66    /// to the peer.
67    Dialing(DialingPeer<'a, TTrans, TInEvent, TOutEvent, THandler>),
68
69    /// There exists no established connection to the peer and there is
70    /// currently no ongoing dialing (i.e. outgoing connection) attempt
71    /// in progress.
72    Disconnected(DisconnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>),
73
74    /// The peer represents the local node.
75    Local,
76}
77
78impl<'a, TTrans, TInEvent, TOutEvent, THandler> fmt::Debug for
79    Peer<'a, TTrans, TInEvent, TOutEvent, THandler>
80where
81    TTrans: Transport,
82    THandler: IntoConnectionHandler,
83{
84    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
85        match self {
86            Peer::Connected(p) => {
87                f.debug_struct("Connected")
88                    .field("peer", &p)
89                    .finish()
90            }
91            Peer::Dialing(p) => {
92                f.debug_struct("Dialing")
93                    .field("peer", &p)
94                    .finish()
95            }
96            Peer::Disconnected(p) => {
97                f.debug_struct("Disconnected")
98                    .field("peer", &p)
99                    .finish()
100            }
101            Peer::Local => {
102                f.debug_struct("Local")
103                    .finish()
104            }
105        }
106    }
107}
108
109impl<'a, TTrans, TInEvent, TOutEvent, THandler>
110    Peer<'a, TTrans, TInEvent, TOutEvent, THandler>
111where
112    TTrans: Transport,
113    THandler: IntoConnectionHandler,
114{
115    pub(super) fn new(
116        network: &'a mut Network<TTrans, TInEvent, TOutEvent, THandler>,
117        peer_id: PeerId
118    ) -> Self {
119        if peer_id == network.local_peer_id {
120            return Peer::Local;
121        }
122
123        if network.pool.is_connected(&peer_id) {
124            return Self::connected(network, peer_id)
125        }
126
127        if network.dialing.get_mut(&peer_id).is_some() {
128            return Self::dialing(network, peer_id);
129        }
130
131        Self::disconnected(network, peer_id)
132    }
133
134
135    fn disconnected(
136        network: &'a mut Network<TTrans, TInEvent, TOutEvent, THandler>,
137        peer_id: PeerId
138    ) -> Self {
139        Peer::Disconnected(DisconnectedPeer { network, peer_id })
140    }
141
142    fn connected(
143        network: &'a mut Network<TTrans, TInEvent, TOutEvent, THandler>,
144        peer_id: PeerId
145    ) -> Self {
146        Peer::Connected(ConnectedPeer { network, peer_id })
147    }
148
149    fn dialing(
150        network: &'a mut Network<TTrans, TInEvent, TOutEvent, THandler>,
151        peer_id: PeerId
152    ) -> Self {
153        Peer::Dialing(DialingPeer { network, peer_id })
154    }
155}
156
157impl<'a, TTrans, TMuxer, TInEvent, TOutEvent, THandler>
158    Peer<'a, TTrans, TInEvent, TOutEvent, THandler>
159where
160    TTrans: Transport<Output = (PeerId, TMuxer)> + Clone,
161    TTrans::Error: Send + 'static,
162    TTrans::Dial: Send + 'static,
163    TMuxer: StreamMuxer + Send + Sync + 'static,
164    TMuxer::OutboundSubstream: Send,
165    TInEvent: Send + 'static,
166    TOutEvent: Send + 'static,
167    THandler: IntoConnectionHandler + Send + 'static,
168    THandler::Handler: ConnectionHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent> + Send,
169    <THandler::Handler as ConnectionHandler>::OutboundOpenInfo: Send,
170    <THandler::Handler as ConnectionHandler>::Error: error::Error + Send + 'static,
171{
172    /// Checks whether the peer is currently connected.
173    ///
174    /// Returns `true` iff [`Peer::into_connected`] returns `Some`.
175    pub fn is_connected(&self) -> bool {
176        match self {
177            Peer::Connected(..) => true,
178            Peer::Dialing(peer) => peer.is_connected(),
179            Peer::Disconnected(..) => false,
180            Peer::Local => false
181        }
182    }
183
184    /// Checks whether the peer is currently being dialed.
185    ///
186    /// Returns `true` iff [`Peer::into_dialing`] returns `Some`.
187    pub fn is_dialing(&self) -> bool {
188        match self {
189            Peer::Dialing(_) => true,
190            Peer::Connected(peer) => peer.is_dialing(),
191            Peer::Disconnected(..) => false,
192            Peer::Local => false
193        }
194    }
195
196    /// Checks whether the peer is currently disconnected.
197    ///
198    /// Returns `true` iff [`Peer::into_disconnected`] returns `Some`.
199    pub fn is_disconnected(&self) -> bool {
200        matches!(self, Peer::Disconnected(..))
201    }
202
203    /// Initiates a new dialing attempt to this peer using the given addresses.
204    ///
205    /// The connection ID of the first connection attempt, i.e. to `address`,
206    /// is returned, together with a [`DialingPeer`] for further use. The
207    /// `remaining` addresses are tried in order in subsequent connection
208    /// attempts in the context of the same dialing attempt, if the connection
209    /// attempt to the first address fails.
210    pub fn dial<I>(self, address: Multiaddr, remaining: I, handler: THandler)
211        -> Result<
212            (ConnectionId, DialingPeer<'a, TTrans, TInEvent, TOutEvent, THandler>),
213            ConnectionLimit
214        >
215    where
216        I: IntoIterator<Item = Multiaddr>,
217    {
218        let (peer_id, network) = match self {
219            Peer::Connected(p) => (p.peer_id, p.network),
220            Peer::Dialing(p) => (p.peer_id, p.network),
221            Peer::Disconnected(p) => (p.peer_id, p.network),
222            Peer::Local => return Err(ConnectionLimit { current: 0, limit: 0 })
223        };
224
225        let id = network.dial_peer(DialingOpts {
226            peer: peer_id,
227            handler,
228            address,
229            remaining: remaining.into_iter().collect(),
230        })?;
231
232        Ok((id, DialingPeer { network, peer_id }))
233    }
234
235    /// Converts the peer into a `ConnectedPeer`, if an established connection exists.
236    ///
237    /// Succeeds if the there is at least one established connection to the peer.
238    pub fn into_connected(self) -> Option<
239        ConnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
240    > {
241        match self {
242            Peer::Connected(peer) => Some(peer),
243            Peer::Dialing(peer) => peer.into_connected(),
244            Peer::Disconnected(..) => None,
245            Peer::Local => None,
246        }
247    }
248
249    /// Converts the peer into a `DialingPeer`, if a dialing attempt exists.
250    ///
251    /// Succeeds if the there is at least one pending outgoing connection to the peer.
252    pub fn into_dialing(self) -> Option<
253        DialingPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
254    > {
255        match self {
256            Peer::Dialing(peer) => Some(peer),
257            Peer::Connected(peer) => peer.into_dialing(),
258            Peer::Disconnected(..) => None,
259            Peer::Local => None
260        }
261    }
262
263    /// Converts the peer into a `DisconnectedPeer`, if neither an established connection
264    /// nor a dialing attempt exists.
265    pub fn into_disconnected(self) -> Option<
266        DisconnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
267    > {
268        match self {
269            Peer::Disconnected(peer) => Some(peer),
270            _ => None,
271        }
272    }
273}
274
275/// The representation of a peer in a [`Network`] to whom at least
276/// one established connection exists. There may also be additional ongoing
277/// dialing attempts to the peer.
278pub struct ConnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
279where
280    TTrans: Transport,
281    THandler: IntoConnectionHandler,
282{
283    network: &'a mut Network<TTrans, TInEvent, TOutEvent, THandler>,
284    peer_id: PeerId,
285}
286
287impl<'a, TTrans, TInEvent, TOutEvent, THandler>
288    ConnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
289where
290    TTrans: Transport,
291    THandler: IntoConnectionHandler,
292{
293    pub fn id(&self) -> &PeerId {
294        &self.peer_id
295    }
296
297    /// Returns the `ConnectedPeer` into a `Peer`.
298    pub fn into_peer(self) -> Peer<'a, TTrans, TInEvent, TOutEvent, THandler> {
299        Peer::Connected(self)
300    }
301
302    /// Obtains an established connection to the peer by ID.
303    pub fn connection(&mut self, id: ConnectionId)
304        -> Option<EstablishedConnection<TInEvent>>
305    {
306        self.network.pool.get_established(id)
307    }
308
309    /// The number of established connections to the peer.
310    pub fn num_connections(&self) -> u32 {
311        self.network.pool.num_peer_established(&self.peer_id)
312    }
313
314    /// Checks whether there is an ongoing dialing attempt to the peer.
315    ///
316    /// Returns `true` iff [`ConnectedPeer::into_dialing`] returns `Some`.
317    pub fn is_dialing(&self) -> bool {
318        self.network.dialing.contains_key(&self.peer_id)
319    }
320
321    /// Converts this peer into a [`DialingPeer`], if there is an ongoing
322    /// dialing attempt, `None` otherwise.
323    pub fn into_dialing(self) -> Option<
324        DialingPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
325    > {
326        if self.network.dialing.contains_key(&self.peer_id) {
327            Some(DialingPeer { network: self.network, peer_id: self.peer_id })
328        } else {
329            None
330        }
331    }
332
333    /// Gets an iterator over all established connections to the peer.
334    pub fn connections(&mut self) ->
335        EstablishedConnectionIter<
336            impl Iterator<Item = ConnectionId>,
337            TInEvent,
338            TOutEvent,
339            THandler,
340            TTrans::Error,
341            <THandler::Handler as ConnectionHandler>::Error>
342    {
343        self.network.pool.iter_peer_established(&self.peer_id)
344    }
345
346    /// Obtains some established connection to the peer.
347    pub fn some_connection(&mut self)
348        -> EstablishedConnection<TInEvent>
349    {
350        self.connections()
351            .into_first()
352            .expect("By `Peer::new` and the definition of `ConnectedPeer`.")
353    }
354
355    /// Disconnects from the peer, closing all connections.
356    pub fn disconnect(self)
357        -> DisconnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
358    {
359        self.network.disconnect(&self.peer_id);
360        DisconnectedPeer { network: self.network, peer_id: self.peer_id }
361    }
362}
363
364impl<'a, TTrans, TInEvent, TOutEvent, THandler> fmt::Debug for
365    ConnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
366where
367    TTrans: Transport,
368    THandler: IntoConnectionHandler,
369{
370    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
371        f.debug_struct("ConnectedPeer")
372            .field("peer_id", &self.peer_id)
373            .field("established", &self.network.pool.iter_peer_established_info(&self.peer_id))
374            .field("attempts", &self.network.dialing.get(&self.peer_id))
375            .finish()
376    }
377}
378
379/// The representation of a peer in a [`Network`] to whom a dialing
380/// attempt is ongoing. There may already exist other established
381/// connections to this peer.
382pub struct DialingPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
383where
384    TTrans: Transport,
385    THandler: IntoConnectionHandler,
386{
387    network: &'a mut Network<TTrans, TInEvent, TOutEvent, THandler>,
388    peer_id: PeerId,
389}
390
391impl<'a, TTrans, TInEvent, TOutEvent, THandler>
392    DialingPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
393where
394    TTrans: Transport,
395    THandler: IntoConnectionHandler,
396{
397    pub fn id(&self) -> &PeerId {
398        &self.peer_id
399    }
400
401    /// Returns the `DialingPeer` into a `Peer`.
402    pub fn into_peer(self) -> Peer<'a, TTrans, TInEvent, TOutEvent, THandler> {
403        Peer::Dialing(self)
404    }
405
406    /// Disconnects from this peer, closing all established connections and
407    /// aborting all dialing attempts.
408    pub fn disconnect(self)
409        -> DisconnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
410    {
411        self.network.disconnect(&self.peer_id);
412        DisconnectedPeer { network: self.network, peer_id: self.peer_id }
413    }
414
415    /// Checks whether there is an established connection to the peer.
416    ///
417    /// Returns `true` iff [`DialingPeer::into_connected`] returns `Some`.
418    pub fn is_connected(&self) -> bool {
419        self.network.pool.is_connected(&self.peer_id)
420    }
421
422    /// Converts the peer into a `ConnectedPeer`, if an established connection exists.
423    pub fn into_connected(self)
424        -> Option<ConnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>>
425    {
426        if self.is_connected() {
427            Some(ConnectedPeer { peer_id: self.peer_id, network: self.network })
428        } else {
429            None
430        }
431    }
432
433    /// Obtains a dialing attempt to the peer by connection ID of
434    /// the current connection attempt.
435    pub fn attempt(&mut self, id: ConnectionId)
436        -> Option<DialingAttempt<'_, TInEvent>>
437    {
438        if let hash_map::Entry::Occupied(attempts) = self.network.dialing.entry(self.peer_id) {
439            if let Some(pos) = attempts.get().iter().position(|s| s.current.0 == id) {
440                if let Some(inner) = self.network.pool.get_outgoing(id) {
441                    return Some(DialingAttempt { pos, inner, attempts })
442                }
443            }
444        }
445        None
446    }
447
448    /// Gets an iterator over all dialing (i.e. pending outgoing) connections to the peer.
449    pub fn attempts(&mut self)
450        -> DialingAttemptIter<'_,
451            TInEvent,
452            TOutEvent,
453            THandler,
454            TTrans::Error,
455            <THandler::Handler as ConnectionHandler>::Error>
456    {
457        DialingAttemptIter::new(&self.peer_id, &mut self.network.pool, &mut self.network.dialing)
458    }
459
460    /// Obtains some dialing connection to the peer.
461    ///
462    /// At least one dialing connection is guaranteed to exist on a `DialingPeer`.
463    pub fn some_attempt(&mut self)
464        -> DialingAttempt<'_, TInEvent>
465    {
466        self.attempts()
467            .into_first()
468            .expect("By `Peer::new` and the definition of `DialingPeer`.")
469    }
470}
471
472impl<'a, TTrans, TInEvent, TOutEvent, THandler> fmt::Debug for
473    DialingPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
474where
475    TTrans: Transport,
476    THandler: IntoConnectionHandler,
477{
478    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
479        f.debug_struct("DialingPeer")
480            .field("peer_id", &self.peer_id)
481            .field("established", &self.network.pool.iter_peer_established_info(&self.peer_id))
482            .field("attempts", &self.network.dialing.get(&self.peer_id))
483            .finish()
484    }
485}
486
487/// The representation of a peer to whom the `Network` has currently
488/// neither an established connection, nor an ongoing dialing attempt
489/// initiated by the local peer.
490pub struct DisconnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
491where
492    TTrans: Transport,
493    THandler: IntoConnectionHandler,
494{
495    peer_id: PeerId,
496    network: &'a mut Network<TTrans, TInEvent, TOutEvent, THandler>,
497}
498
499impl<'a, TTrans, TInEvent, TOutEvent, THandler> fmt::Debug for
500    DisconnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
501where
502    TTrans: Transport,
503    THandler: IntoConnectionHandler,
504{
505    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
506        f.debug_struct("DisconnectedPeer")
507            .field("peer_id", &self.peer_id)
508            .finish()
509    }
510}
511
512impl<'a, TTrans, TInEvent, TOutEvent, THandler>
513    DisconnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>
514where
515    TTrans: Transport,
516    THandler: IntoConnectionHandler,
517{
518    pub fn id(&self) -> &PeerId {
519        &self.peer_id
520    }
521
522    /// Returns the `DisconnectedPeer` into a `Peer`.
523    pub fn into_peer(self) -> Peer<'a, TTrans, TInEvent, TOutEvent, THandler> {
524        Peer::Disconnected(self)
525    }
526
527    /// Moves the peer into a connected state by supplying an existing
528    /// established connection.
529    ///
530    /// No event is generated for this action.
531    ///
532    /// # Panics
533    ///
534    /// Panics if `connected.peer_id` does not identify the current peer.
535    pub fn set_connected<TMuxer>(
536        self,
537        connected: Connected,
538        connection: Connection<TMuxer, THandler::Handler>,
539    ) -> Result<
540        ConnectedPeer<'a, TTrans, TInEvent, TOutEvent, THandler>,
541        ConnectionLimit
542    > where
543        TInEvent: Send + 'static,
544        TOutEvent: Send + 'static,
545        THandler: Send + 'static,
546        TTrans::Error: Send + 'static,
547        THandler::Handler: ConnectionHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent> + Send,
548        <THandler::Handler as ConnectionHandler>::OutboundOpenInfo: Send,
549        <THandler::Handler as ConnectionHandler>::Error: error::Error + Send + 'static,
550        TMuxer: StreamMuxer + Send + Sync + 'static,
551        TMuxer::OutboundSubstream: Send,
552    {
553        if connected.peer_id != self.peer_id {
554            panic!("Invalid peer ID given: {:?}. Expected: {:?}", connected.peer_id, self.peer_id)
555        }
556
557        self.network.pool.add(connection, connected)
558            .map(move |_id| ConnectedPeer {
559                network: self.network,
560                peer_id: self.peer_id,
561            })
562    }
563}
564
565/// The (internal) state of a `DialingAttempt`, tracking the
566/// current connection attempt as well as remaining addresses.
567#[derive(Debug, Clone)]
568pub(super) struct DialingState {
569    /// The ID and (remote) address of the current connection attempt.
570    pub(super) current: (ConnectionId, Multiaddr),
571    /// Multiaddresses to attempt if the current one fails.
572    pub(super) remaining: Vec<Multiaddr>,
573}
574
575/// A `DialingAttempt` is an ongoing outgoing connection attempt to
576/// a known / expected remote peer ID and a list of alternative addresses
577/// to connect to, if the current connection attempt fails.
578pub struct DialingAttempt<'a, TInEvent> {
579    /// The underlying pending connection in the `Pool`.
580    inner: PendingConnection<'a, TInEvent>,
581    /// All current dialing attempts of the peer.
582    attempts: hash_map::OccupiedEntry<'a, PeerId, SmallVec<[DialingState; 10]>>,
583    /// The position of the current `DialingState` of this connection in the `attempts`.
584    pos: usize,
585}
586
587impl<'a, TInEvent>
588    DialingAttempt<'a, TInEvent>
589{
590    /// Returns the ID of the current connection attempt.
591    pub fn id(&self) -> ConnectionId {
592        self.inner.id()
593    }
594
595    /// Returns the (expected) peer ID of the dialing attempt.
596    pub fn peer_id(&self) -> &PeerId {
597        self.attempts.key()
598    }
599
600    /// Returns the remote address of the current connection attempt.
601    pub fn address(&self) -> &Multiaddr {
602        match self.inner.endpoint() {
603            ConnectedPoint::Dialer { address } => address,
604            ConnectedPoint::Listener { .. } => unreachable!("by definition of a `DialingAttempt`.")
605        }
606    }
607
608    /// Aborts the dialing attempt.
609    ///
610    /// Aborting a dialing attempt involves aborting the current connection
611    /// attempt and dropping any remaining addresses given to [`Peer::dial()`]
612    /// that have not yet been tried.
613    pub fn abort(mut self) {
614        self.attempts.get_mut().remove(self.pos);
615        if self.attempts.get().is_empty() {
616            self.attempts.remove();
617        }
618        self.inner.abort();
619    }
620
621    /// Adds an address to the end of the remaining addresses
622    /// for this dialing attempt. Duplicates are ignored.
623    pub fn add_address(&mut self, addr: Multiaddr) {
624        let remaining = &mut self.attempts.get_mut()[self.pos].remaining;
625        if remaining.iter().all(|a| a != &addr) {
626            remaining.push(addr);
627        }
628    }
629}
630
631/// An iterator over the ongoing dialing attempts to a peer.
632pub struct DialingAttemptIter<'a, TInEvent, TOutEvent, THandler, TTransErr, THandlerErr> {
633    /// The peer whose dialing attempts are being iterated.
634    peer_id: &'a PeerId,
635    /// The underlying connection `Pool` of the `Network`.
636    pool: &'a mut Pool<TInEvent, TOutEvent, THandler, TTransErr, THandlerErr>,
637    /// The state of all current dialing attempts known to the `Network`.
638    ///
639    /// Ownership of the `OccupiedEntry` for `peer_id` containing all attempts must be
640    /// borrowed to each `DialingAttempt` in order for it to remove the entry if the
641    /// last dialing attempt is aborted.
642    dialing: &'a mut FnvHashMap<PeerId, SmallVec<[DialingState; 10]>>,
643    /// The current position of the iterator in `dialing[peer_id]`.
644    pos: usize,
645    /// The total number of elements in `dialing[peer_id]` to iterate over.
646    end: usize,
647}
648
649// Note: Ideally this would be an implementation of `Iterator`, but that
650// requires GATs (cf. https://github.com/rust-lang/rust/issues/44265) and
651// a different definition of `Iterator`.
652impl<'a, TInEvent, TOutEvent, THandler, TTransErr, THandlerErr>
653    DialingAttemptIter<'a, TInEvent, TOutEvent, THandler, TTransErr, THandlerErr>
654{
655    fn new(
656        peer_id: &'a PeerId,
657        pool: &'a mut Pool<TInEvent, TOutEvent, THandler, TTransErr, THandlerErr>,
658        dialing: &'a mut FnvHashMap<PeerId, SmallVec<[DialingState; 10]>>,
659    ) -> Self {
660        let end = dialing.get(peer_id).map_or(0, |conns| conns.len());
661        Self { pos: 0, end, pool, dialing, peer_id }
662    }
663
664    /// Obtains the next dialing connection, if any.
665    #[allow(clippy::should_implement_trait)]
666    pub fn next(&mut self) -> Option<DialingAttempt<'_, TInEvent>> {
667        // If the number of elements reduced, the current `DialingAttempt` has been
668        // aborted and iteration needs to continue from the previous position to
669        // account for the removed element.
670        let end = self.dialing.get(self.peer_id).map_or(0, |conns| conns.len());
671        if self.end > end {
672            self.end = end;
673            self.pos -= 1;
674        }
675
676        if self.pos == self.end {
677            return None
678        }
679
680        if let hash_map::Entry::Occupied(attempts) = self.dialing.entry(*self.peer_id) {
681            let id = attempts.get()[self.pos].current.0;
682            if let Some(inner) = self.pool.get_outgoing(id) {
683                let conn = DialingAttempt { pos: self.pos, inner, attempts };
684                self.pos += 1;
685                return Some(conn)
686            }
687        }
688
689        None
690    }
691
692    /// Returns the first connection, if any, consuming the iterator.
693    pub fn into_first<'b>(self)
694        -> Option<DialingAttempt<'b, TInEvent>>
695    where 'a: 'b
696    {
697        if self.pos == self.end {
698            return None
699        }
700
701        if let hash_map::Entry::Occupied(attempts) = self.dialing.entry(*self.peer_id) {
702            let id = attempts.get()[self.pos].current.0;
703            if let Some(inner) = self.pool.get_outgoing(id) {
704                return Some(DialingAttempt { pos: self.pos, inner, attempts })
705            }
706        }
707
708        None
709    }
710}
mwc_libp2p_core/network/peer.rs

mwc_libp2p_core/network/
peer.rs
