Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat(swarm): introduce ToSwarm type alias #3242

Closed
wants to merge 1 commit into from
Closed

feat(swarm): introduce ToSwarm type alias #3242

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 23 additions & 32 deletions protocols/autonat/src/behaviour.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,12 +35,13 @@ use libp2p_request_response::{
self as request_response, ProtocolSupport, RequestId, ResponseChannel,
};
use libp2p_swarm::{
behaviour::ToSwarm,
behaviour::{
AddressChange, ConnectionClosed, ConnectionEstablished, DialFailure, ExpiredExternalAddr,
ExpiredListenAddr, FromSwarm,
},
ConnectionHandler, ExternalAddresses, IntoConnectionHandler, ListenAddresses, NetworkBehaviour,
NetworkBehaviourAction, PollParameters,
PollParameters,
};
use std::{
collections::{HashMap, VecDeque},
Expand Down Expand Up @@ -209,12 +210,7 @@ pub struct Behaviour {

last_probe: Option<Instant>,

pending_actions: VecDeque<
NetworkBehaviourAction<
<Self as NetworkBehaviour>::OutEvent,
<Self as NetworkBehaviour>::ConnectionHandler,
>,
>,
pending_to_swarm: VecDeque<ToSwarm<Behaviour>>,

probe_id: ProbeId,

Expand Down Expand Up @@ -242,7 +238,7 @@ impl Behaviour {
throttled_servers: Vec::new(),
throttled_clients: Vec::new(),
last_probe: None,
pending_actions: VecDeque::new(),
pending_to_swarm: VecDeque::new(),
probe_id: ProbeId(0),
listen_addresses: Default::default(),
external_addresses: Default::default(),
Expand Down Expand Up @@ -339,10 +335,8 @@ impl Behaviour {
role_override: Endpoint::Dialer,
} => {
if let Some(event) = self.as_server().on_outbound_connection(&peer, address) {
self.pending_actions
.push_back(NetworkBehaviourAction::GenerateEvent(Event::InboundProbe(
event,
)));
self.pending_to_swarm
.push_back(ToSwarm::<Self>::GenerateEvent(Event::InboundProbe(event)));
}
}
ConnectedPoint::Dialer {
Expand Down Expand Up @@ -402,10 +396,8 @@ impl Behaviour {
error,
}));
if let Some(event) = self.as_server().on_outbound_dial_error(peer_id, error) {
self.pending_actions
.push_back(NetworkBehaviourAction::GenerateEvent(Event::InboundProbe(
event,
)));
self.pending_to_swarm
.push_back(ToSwarm::<Self>::GenerateEvent(Event::InboundProbe(event)));
}
}

Expand Down Expand Up @@ -438,14 +430,20 @@ impl NetworkBehaviour for Behaviour {
<request_response::Behaviour<AutoNatCodec> as NetworkBehaviour>::ConnectionHandler;
type OutEvent = Event;

fn poll(&mut self, cx: &mut Context<'_>, params: &mut impl PollParameters) -> Poll<Action> {
fn poll(
&mut self,
cx: &mut Context<'_>,
params: &mut impl PollParameters,
) -> Poll<ToSwarm<Behaviour>> {
loop {
if let Some(event) = self.pending_actions.pop_front() {
if let Some(event) = self.pending_to_swarm.pop_front() {
return Poll::Ready(event);
}

match self.inner.poll(cx, params) {
Poll::Ready(NetworkBehaviourAction::GenerateEvent(event)) => {
Poll::Ready(
ToSwarm::<request_response::Behaviour<AutoNatCodec>>::GenerateEvent(event),
) => {
let (events, action) = match event {
request_response::Event::Message {
message: request_response::Message::Response { .. },
Expand All @@ -464,16 +462,16 @@ impl NetworkBehaviour for Behaviour {
request_response::Event::ResponseSent { .. } => (VecDeque::new(), None),
};

self.pending_actions.extend(
self.pending_to_swarm.extend(
events
.into_iter()
.map(NetworkBehaviourAction::GenerateEvent)
.map(ToSwarm::<Self>::GenerateEvent)
.chain(action),
);
continue;
}
Poll::Ready(action) => {
self.pending_actions
self.pending_to_swarm
.push_back(action.map_out(|_| unreachable!()));
continue;
}
Expand All @@ -482,10 +480,8 @@ impl NetworkBehaviour for Behaviour {

match self.as_client().poll_auto_probe(cx) {
Poll::Ready(event) => {
self.pending_actions
.push_back(NetworkBehaviourAction::GenerateEvent(Event::OutboundProbe(
event,
)));
self.pending_to_swarm
.push_back(ToSwarm::<Self>::GenerateEvent(Event::OutboundProbe(event)));
continue;
}
Poll::Pending => {}
Expand Down Expand Up @@ -568,18 +564,13 @@ impl NetworkBehaviour for Behaviour {
}
}

type Action = NetworkBehaviourAction<
<Behaviour as NetworkBehaviour>::OutEvent,
<Behaviour as NetworkBehaviour>::ConnectionHandler,
>;

// Trait implemented for `AsClient` and `AsServer` to handle events from the inner [`request_response::Behaviour`] Protocol.
trait HandleInnerEvent {
fn handle_event(
&mut self,
params: &mut impl PollParameters,
event: request_response::Event<DialRequest, DialResponse>,
) -> (VecDeque<Event>, Option<Action>);
) -> (VecDeque<Event>, Option<ToSwarm<Behaviour>>);
}

trait GlobalIp {
Expand Down
14 changes: 6 additions & 8 deletions protocols/autonat/src/behaviour/as_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,20 +18,18 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.

use crate::ResponseError;
use crate::{Behaviour, ResponseError};

use super::{
Action, AutoNatCodec, Config, DialRequest, DialResponse, Event, HandleInnerEvent, NatStatus,
ProbeId,
AutoNatCodec, Config, DialRequest, DialResponse, Event, HandleInnerEvent, NatStatus, ProbeId,
};
use futures::FutureExt;
use futures_timer::Delay;
use instant::Instant;
use libp2p_core::{connection::ConnectionId, Multiaddr, PeerId};
use libp2p_request_response::{self as request_response, OutboundFailure, RequestId};
use libp2p_swarm::{
AddressScore, ExternalAddresses, ListenAddresses, NetworkBehaviourAction, PollParameters,
};
use libp2p_swarm::behaviour::ToSwarm;
use libp2p_swarm::{AddressScore, ExternalAddresses, ListenAddresses, PollParameters};
use rand::{seq::SliceRandom, thread_rng};
use std::{
collections::{HashMap, VecDeque},
Expand Down Expand Up @@ -109,7 +107,7 @@ impl<'a> HandleInnerEvent for AsClient<'a> {
&mut self,
params: &mut impl PollParameters,
event: request_response::Event<DialRequest, DialResponse>,
) -> (VecDeque<Event>, Option<Action>) {
) -> (VecDeque<Event>, Option<ToSwarm<Behaviour>>) {
let mut events = VecDeque::new();
let mut action = None;
match event {
Expand Down Expand Up @@ -158,7 +156,7 @@ impl<'a> HandleInnerEvent for AsClient<'a> {
.find_map(|r| (r.addr == address).then_some(r.score))
.unwrap_or(AddressScore::Finite(0));
if let AddressScore::Finite(finite_score) = score {
action = Some(NetworkBehaviourAction::ReportObservedAddr {
action = Some(ToSwarm::<Behaviour>::ReportObservedAddr {
address,
score: AddressScore::Finite(finite_score + 1),
});
Expand Down
10 changes: 6 additions & 4 deletions protocols/autonat/src/behaviour/as_server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,17 +19,19 @@
// DEALINGS IN THE SOFTWARE.

use super::{
Action, AutoNatCodec, Config, DialRequest, DialResponse, Event, HandleInnerEvent, ProbeId,
AutoNatCodec, Config, DialRequest, DialResponse, Event, HandleInnerEvent, ProbeId,
ResponseError,
};
use crate::Behaviour;
use instant::Instant;
use libp2p_core::{connection::ConnectionId, multiaddr::Protocol, Multiaddr, PeerId};
use libp2p_request_response::{
self as request_response, InboundFailure, RequestId, ResponseChannel,
};
use libp2p_swarm::behaviour::ToSwarm;
use libp2p_swarm::{
dial_opts::{DialOpts, PeerCondition},
DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters,
DialError, NetworkBehaviour, PollParameters,
};
use std::{
collections::{HashMap, HashSet, VecDeque},
Expand Down Expand Up @@ -98,7 +100,7 @@ impl<'a> HandleInnerEvent for AsServer<'a> {
&mut self,
_params: &mut impl PollParameters,
event: request_response::Event<DialRequest, DialResponse>,
) -> (VecDeque<Event>, Option<Action>) {
) -> (VecDeque<Event>, Option<ToSwarm<Behaviour>>) {
let mut events = VecDeque::new();
let mut action = None;
match event {
Expand Down Expand Up @@ -130,7 +132,7 @@ impl<'a> HandleInnerEvent for AsServer<'a> {
addresses: addrs.clone(),
}));

action = Some(NetworkBehaviourAction::Dial {
action = Some(ToSwarm::<Behaviour>::Dial {
opts: DialOpts::peer_id(peer)
.condition(PeerCondition::Always)
.override_dial_concurrency_factor(NonZeroU8::new(1).expect("1 > 0"))
Expand Down
Loading