fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-05-02 14:12:18 +00:00
Code Issues Projects Releases Wiki Activity

swarm/src/behaviour: Deprecate NetworkBehaviourEventProcess (#2784)

Browse Source 
In preparation for https://github.com/libp2p/rust-libp2p/pull/2751.
This commit is contained in:
Max Inden 2022-08-16 06:58:17 +02:00 committed by GitHub
parent 0e5a25dea8
commit 878c49fa14
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 267 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
examples/chat-tokio.rs
View File

@ -44,7 +44,7 @@ use libp2p::{
mdns::{Mdns, MdnsEvent},
mplex,
noise,
swarm::{dial_opts::DialOpts, NetworkBehaviourEventProcess, SwarmBuilder, SwarmEvent},
swarm::{SwarmBuilder, SwarmEvent},
// `TokioTcpTransport` is available through the `tcp-tokio` feature.
tcp::TokioTcpTransport,
Multiaddr,
@ -82,47 +82,29 @@ async fn main() -> Result<(), Box<dyn Error>> {
// Create a Floodsub topic
let floodsub_topic = floodsub::Topic::new("chat");
// We create a custom network behaviour that combines floodsub and mDNS.
// The derive generates a delegating `NetworkBehaviour` impl which in turn
// requires the implementations of `NetworkBehaviourEventProcess` for
// the events of each behaviour.
// We create a custom behaviour that combines floodsub and mDNS.
// The derive generates a delegating `NetworkBehaviour` impl.
#[derive(NetworkBehaviour)]
#[behaviour(event_process = true)]
#[behaviour(out_event = "MyBehaviourEvent")]
struct MyBehaviour {
floodsub: Floodsub,
mdns: Mdns,
}
impl NetworkBehaviourEventProcess<FloodsubEvent> for MyBehaviour {
// Called when `floodsub` produces an event.
fn inject_event(&mut self, message: FloodsubEvent) {
if let FloodsubEvent::Message(message) = message {
println!(
"Received: '{:?}' from {:?}",
String::from_utf8_lossy(&message.data),
message.source
);
}
enum MyBehaviourEvent {
Floodsub(FloodsubEvent),
Mdns(MdnsEvent),
}
impl From<FloodsubEvent> for MyBehaviourEvent {
fn from(event: FloodsubEvent) -> Self {
MyBehaviourEvent::Floodsub(event)
}
}
impl NetworkBehaviourEventProcess<MdnsEvent> for MyBehaviour {
// Called when `mdns` produces an event.
fn inject_event(&mut self, event: MdnsEvent) {
match event {
MdnsEvent::Discovered(list) => {
for (peer, _) in list {
self.floodsub.add_node_to_partial_view(peer);
}
}
MdnsEvent::Expired(list) => {
for (peer, _) in list {
if !self.mdns.has_node(&peer) {
self.floodsub.remove_node_from_partial_view(&peer);
}
}
}
}
impl From<MdnsEvent> for MyBehaviourEvent {
fn from(event: MdnsEvent) -> Self {
MyBehaviourEvent::Mdns(event)
}
}
@ -166,8 +148,36 @@ async fn main() -> Result<(), Box<dyn Error>> {
swarm.behaviour_mut().floodsub.publish(floodsub_topic.clone(), line.as_bytes());
}
event = swarm.select_next_some() => {
if let SwarmEvent::NewListenAddr { address, .. } = event {
println!("Listening on {:?}", address);
match event {
SwarmEvent::NewListenAddr { address, .. } => {
println!("Listening on {:?}", address);
}
SwarmEvent::Behaviour(MyBehaviourEvent::Floodsub(event)) => {
if let FloodsubEvent::Message(message) = event {
println!(
"Received: '{:?}' from {:?}",
String::from_utf8_lossy(&message.data),
message.source
);
}
}
SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(event)) => {
match event {
MdnsEvent::Discovered(list) => {
for (peer, _) in list {
swarm.behaviour_mut().floodsub.add_node_to_partial_view(peer);
}
}
MdnsEvent::Expired(list) => {
for (peer, _) in list {
if !swarm.behaviour().mdns.has_node(&peer) {
swarm.behaviour_mut().floodsub.remove_node_from_partial_view(&peer);
}
}
}
}
}
_ => {}
}
}
}

4
examples/chat.rs
View File

@ -79,9 +79,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
let floodsub_topic = floodsub::Topic::new("chat");
// We create a custom network behaviour that combines floodsub and mDNS.
// In the future, we want to improve libp2p to make this easier to do.
// Use the derive to generate delegating NetworkBehaviour impl and require the
// NetworkBehaviourEventProcess implementations below.
// Use the derive to generate delegating NetworkBehaviour impl.
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "OutEvent")]
struct MyBehaviour {

94
examples/distributed-key-value-store.rs
View File

@ -50,7 +50,7 @@ use libp2p::kad::{
use libp2p::{
development_transport, identity,
mdns::{Mdns, MdnsConfig, MdnsEvent},
swarm::{NetworkBehaviourEventProcess, SwarmEvent},
swarm::SwarmEvent,
NetworkBehaviour, PeerId, Swarm,
};
use std::error::Error;
@ -68,28 +68,60 @@ async fn main() -> Result<(), Box<dyn Error>> {
// We create a custom network behaviour that combines Kademlia and mDNS.
#[derive(NetworkBehaviour)]
#[behaviour(event_process = true)]
#[behaviour(out_event = "MyBehaviourEvent")]
struct MyBehaviour {
kademlia: Kademlia<MemoryStore>,
mdns: Mdns,
}
impl NetworkBehaviourEventProcess<MdnsEvent> for MyBehaviour {
// Called when `mdns` produces an event.
fn inject_event(&mut self, event: MdnsEvent) {
if let MdnsEvent::Discovered(list) = event {
for (peer_id, multiaddr) in list {
self.kademlia.add_address(&peer_id, multiaddr);
}
}
enum MyBehaviourEvent {
Kademlia(KademliaEvent),
Mdns(MdnsEvent),
}
impl From<KademliaEvent> for MyBehaviourEvent {
fn from(event: KademliaEvent) -> Self {
MyBehaviourEvent::Kademlia(event)
}
}
impl NetworkBehaviourEventProcess<KademliaEvent> for MyBehaviour {
// Called when `kademlia` produces an event.
fn inject_event(&mut self, message: KademliaEvent) {
match message {
KademliaEvent::OutboundQueryCompleted { result, .. } => match result {
impl From<MdnsEvent> for MyBehaviourEvent {
fn from(event: MdnsEvent) -> Self {
MyBehaviourEvent::Mdns(event)
}
}
// Create a swarm to manage peers and events.
let mut swarm = {
// Create a Kademlia behaviour.
let store = MemoryStore::new(local_peer_id);
let kademlia = Kademlia::new(local_peer_id, store);
let mdns = task::block_on(Mdns::new(MdnsConfig::default()))?;
let behaviour = MyBehaviour { kademlia, mdns };
Swarm::new(transport, behaviour, local_peer_id)
};
// Read full lines from stdin
let mut stdin = io::BufReader::new(io::stdin()).lines().fuse();
// Listen on all interfaces and whatever port the OS assigns.
swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
// Kick it off.
loop {
select! {
line = stdin.select_next_some() => handle_input_line(&mut swarm.behaviour_mut().kademlia, line.expect("Stdin not to close")),
event = swarm.select_next_some() => match event {
SwarmEvent::NewListenAddr { address, .. } => {
println!("Listening in {:?}", address);
},
SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(MdnsEvent::Discovered(list))) => {
for (peer_id, multiaddr) in list {
swarm.behaviour_mut().kademlia.add_address(&peer_id, multiaddr);
}
}
SwarmEvent::Behaviour(MyBehaviourEvent::Kademlia(KademliaEvent::OutboundQueryCompleted { result, ..})) => {
match result {
QueryResult::GetProviders(Ok(ok)) => {
for peer in ok.providers {
println!(
@ -137,38 +169,10 @@ async fn main() -> Result<(), Box<dyn Error>> {
eprintln!("Failed to put provider record: {:?}", err);
}
_ => {}
},
_ => {}
}
}
_ => {}
}
}
// Create a swarm to manage peers and events.
let mut swarm = {
// Create a Kademlia behaviour.
let store = MemoryStore::new(local_peer_id);
let kademlia = Kademlia::new(local_peer_id, store);
let mdns = task::block_on(Mdns::new(MdnsConfig::default()))?;
let behaviour = MyBehaviour { kademlia, mdns };
Swarm::new(transport, behaviour, local_peer_id)
};
// Read full lines from stdin
let mut stdin = io::BufReader::new(io::stdin()).lines().fuse();
// Listen on all interfaces and whatever port the OS assigns.
swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
// Kick it off.
loop {
select! {
line = stdin.select_next_some() => handle_input_line(&mut swarm.behaviour_mut().kademlia, line.expect("Stdin not to close")),
event = swarm.select_next_some() => match event {
SwarmEvent::NewListenAddr { address, .. } => {
println!("Listening in {:?}", address);
},
_ => {}
}
}
}
}

141
examples/ipfs-private.rs
View File

@ -41,9 +41,10 @@ use libp2p::{
identify::{Identify, IdentifyConfig, IdentifyEvent},
identity,
multiaddr::Protocol,
noise, ping,
noise,
ping::{self, PingEvent},
pnet::{PnetConfig, PreSharedKey},
swarm::{NetworkBehaviourEventProcess, SwarmEvent},
swarm::SwarmEvent,
tcp::TcpTransport,
yamux::YamuxConfig,
Multiaddr, NetworkBehaviour, PeerId, Swarm, Transport,
@ -157,78 +158,34 @@ async fn main() -> Result<(), Box<dyn Error>> {
// We create a custom network behaviour that combines gossipsub, ping and identify.
#[derive(NetworkBehaviour)]
#[behaviour(event_process = true)]
#[behaviour(out_event = "MyBehaviourEvent")]
struct MyBehaviour {
gossipsub: Gossipsub,
identify: Identify,
ping: ping::Behaviour,
}
impl NetworkBehaviourEventProcess<IdentifyEvent> for MyBehaviour {
// Called when `identify` produces an event.
fn inject_event(&mut self, event: IdentifyEvent) {
println!("identify: {:?}", event);
enum MyBehaviourEvent {
Gossipsub(GossipsubEvent),
Identify(IdentifyEvent),
Ping(PingEvent),
}
impl From<GossipsubEvent> for MyBehaviourEvent {
fn from(event: GossipsubEvent) -> Self {
MyBehaviourEvent::Gossipsub(event)
}
}
impl NetworkBehaviourEventProcess<GossipsubEvent> for MyBehaviour {
// Called when `gossipsub` produces an event.
fn inject_event(&mut self, event: GossipsubEvent) {
match event {
GossipsubEvent::Message {
propagation_source: peer_id,
message_id: id,
message,
} => println!(
"Got message: {} with id: {} from peer: {:?}",
String::from_utf8_lossy(&message.data),
id,
peer_id
),
_ => {}
}
impl From<IdentifyEvent> for MyBehaviourEvent {
fn from(event: IdentifyEvent) -> Self {
MyBehaviourEvent::Identify(event)
}
}
impl NetworkBehaviourEventProcess<ping::Event> for MyBehaviour {
// Called when `ping` produces an event.
fn inject_event(&mut self, event: ping::Event) {
match event {
ping::Event {
peer,
result: Result::Ok(ping::Success::Ping { rtt }),
} => {
println!(
"ping: rtt to {} is {} ms",
peer.to_base58(),
rtt.as_millis()
);
}
ping::Event {
peer,
result: Result::Ok(ping::Success::Pong),
} => {
println!("ping: pong from {}", peer.to_base58());
}
ping::Event {
peer,
result: Result::Err(ping::Failure::Timeout),
} => {
println!("ping: timeout to {}", peer.to_base58());
}
ping::Event {
peer,
result: Result::Err(ping::Failure::Unsupported),
} => {
println!("ping: {} does not support ping protocol", peer.to_base58());
}
ping::Event {
peer,
result: Result::Err(ping::Failure::Other { error }),
} => {
println!("ping: ping::Failure with {}: {}", peer.to_base58(), error);
}
}
impl From<PingEvent> for MyBehaviourEvent {
fn from(event: PingEvent) -> Self {
MyBehaviourEvent::Ping(event)
}
}
@ -282,8 +239,64 @@ async fn main() -> Result<(), Box<dyn Error>> {
}
},
event = swarm.select_next_some() => {
if let SwarmEvent::NewListenAddr { address, .. } = event {
println!("Listening on {:?}", address);
match event {
SwarmEvent::NewListenAddr { address, .. } => {
println!("Listening on {:?}", address);
}
SwarmEvent::Behaviour(MyBehaviourEvent::Identify(event)) => {
println!("identify: {:?}", event);
}
SwarmEvent::Behaviour(MyBehaviourEvent::Gossipsub(GossipsubEvent::Message {
propagation_source: peer_id,
message_id: id,
message,
})) => {
println!(
"Got message: {} with id: {} from peer: {:?}",
String::from_utf8_lossy(&message.data),
id,
peer_id
)
}
SwarmEvent::Behaviour(MyBehaviourEvent::Ping(event)) => {
match event {
ping::Event {
peer,
result: Result::Ok(ping::Success::Ping { rtt }),
} => {
println!(
"ping: rtt to {} is {} ms",
peer.to_base58(),
rtt.as_millis()
);
}
ping::Event {
peer,
result: Result::Ok(ping::Success::Pong),
} => {
println!("ping: pong from {}", peer.to_base58());
}
ping::Event {
peer,
result: Result::Err(ping::Failure::Timeout),
} => {
println!("ping: timeout to {}", peer.to_base58());
}
ping::Event {
peer,
result: Result::Err(ping::Failure::Unsupported),
} => {
println!("ping: {} does not support ping protocol", peer.to_base58());
}
ping::Event {
peer,
result: Result::Err(ping::Failure::Other { error }),
} => {
println!("ping: ping::Failure with {}: {}", peer.to_base58(), error);
}
}
}
_ => {}
}
}
}

78
swarm/CHANGELOG.md
View File

@ -1,8 +1,72 @@
# 0.38.0 [unreleased]
- Update dial address concurrency factor to `8`, thus dialing up to 8 addresses concurrently for a single connection attempt. See `Swarm::dial_concurrency_factor` and [PR 2741].
- Deprecate `NetworkBehaviourEventProcess`. When deriving `NetworkBehaviour` on a custom `struct` users
should either bring their own `OutEvent` via `#[behaviour(out_event = "MyBehaviourEvent")]` or,
when not specified, have the derive macro generate one for the user.
- Update to `libp2p-core` `v0.35.0`.
See [`NetworkBehaviour`
documentation](https://docs.rs/libp2p/latest/libp2p/swarm/trait.NetworkBehaviour.html) and [PR
2784] for details.
Previously
``` rust
#[derive(NetworkBehaviour)]
#[behaviour(event_process = true)]
struct MyBehaviour {
gossipsub: Gossipsub,
mdns: Mdns,
}
impl NetworkBehaviourEventProcess<Gossipsub> for MyBehaviour {
fn inject_event(&mut self, message: GossipsubEvent) {
todo!("Handle event")
}
}
impl NetworkBehaviourEventProcess<MdnsEvent> for MyBehaviour {
fn inject_event(&mut self, message: MdnsEvent) {
todo!("Handle event")
}
}
```
Now
``` rust
#[derive(NetworkBehaviour)]
#[behaviour(out_event = "MyBehaviourEvent")]
struct MyBehaviour {
gossipsub: Gossipsub,
mdns: Mdns,
}
enum MyBehaviourEvent {
Gossipsub(GossipsubEvent),
Mdns(MdnsEvent),
}
impl From<GossipsubEvent> for MyBehaviourEvent {
fn from(event: GossipsubEvent) -> Self {
MyBehaviourEvent::Gossipsub(event)
}
}
impl From<MdnsEvent> for MyBehaviourEvent {
fn from(event: MdnsEvent) -> Self {
MyBehaviourEvent::Mdns(event)
}
}
match swarm.next().await.unwrap() {
SwarmEvent::Behaviour(MyBehaviourEvent::Gossipsub(event)) => {
todo!("Handle event")
}
SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(event)) => {
todo!("Handle event")
}
}
```
- When deriving `NetworkBehaviour` on a custom `struct` where the user does not specify their own
`OutEvent` via `#[behaviour(out_event = "MyBehaviourEvent")]` and where the user does not enable
@ -10,10 +74,16 @@
the user.
See [`NetworkBehaviour`
documentation](https://docs.rs/libp2p/latest/libp2p/swarm/trait.NetworkBehaviour.html) for
details.
documentation](https://docs.rs/libp2p/latest/libp2p/swarm/trait.NetworkBehaviour.html) and [PR
2792] for details.
- Update dial address concurrency factor to `8`, thus dialing up to 8 addresses concurrently for a single connection attempt. See `Swarm::dial_concurrency_factor` and [PR 2741].
- Update to `libp2p-core` `v0.35.0`.
[PR 2741]: https://github.com/libp2p/rust-libp2p/pull/2741/
[PR 2784]: https://github.com/libp2p/rust-libp2p/pull/2784
[PR 2792]: https://github.com/libp2p/rust-libp2p/pull/2792
# 0.37.0

13
swarm/src/behaviour.rs
View File

@ -85,9 +85,9 @@ pub(crate) type THandlerOutEvent<THandler> =
/// "MyCustomOutEvent")]`. If the user does not specify an `out_event`, the derive macro generates
/// the event definition itself, naming it `<STRUCT_NAME>Event`.
///
/// When setting a custom `out_event`, the aforementioned conversion of each of the event types
/// generated by the struct members to the custom `out_event` is handled by [`From`]
/// implementations the user needs to provide.
/// The aforementioned conversion of each of the event types generated by the struct members to the
/// custom `out_event` is handled by [`From`] implementations which the user needs to define in
/// addition to the event `enum` itself.
///
/// ``` rust
/// # use libp2p::identify::{Identify, IdentifyEvent};
@ -326,6 +326,13 @@ pub trait PollParameters {
///
/// You can opt out of this behaviour through `#[behaviour(event_process = false)]`. See the
/// documentation of [`NetworkBehaviour`] for details.
#[deprecated(
since = "0.38.0",
note = "Use `#[behaviour(out_event = \"MyBehaviourEvent\")]` instead. See \
https://github.com/libp2p/rust-libp2p/blob/master/swarm/CHANGELOG.md#0380 \
for instructions on how to migrate. Will be removed with \
https://github.com/libp2p/rust-libp2p/pull/2751."
)]
pub trait NetworkBehaviourEventProcess<TEvent> {
/// Called when one of the fields of the type you're deriving `NetworkBehaviour` on generates
/// an event.

8
swarm/src/behaviour/either.rs
View File

@ -19,10 +19,9 @@
// DEALINGS IN THE SOFTWARE.
use crate::handler::{either::IntoEitherHandler, ConnectionHandler, IntoConnectionHandler};
use crate::{
DialError, NetworkBehaviour, NetworkBehaviourAction, NetworkBehaviourEventProcess,
PollParameters,
};
#[allow(deprecated)]
pub use crate::NetworkBehaviourEventProcess;
use crate::{DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use either::Either;
use libp2p_core::{
connection::ConnectionId, transport::ListenerId, ConnectedPoint, Multiaddr, PeerId,
@ -237,6 +236,7 @@ where
}
}
#[allow(deprecated)]
impl<TEvent, TBehaviourLeft, TBehaviourRight> NetworkBehaviourEventProcess<TEvent>
for Either<TBehaviourLeft, TBehaviourRight>
where

8
swarm/src/behaviour/toggle.rs
View File

@ -23,10 +23,9 @@ use crate::handler::{
KeepAlive, SubstreamProtocol,
};
use crate::upgrade::{InboundUpgradeSend, OutboundUpgradeSend, SendWrapper};
use crate::{
DialError, NetworkBehaviour, NetworkBehaviourAction, NetworkBehaviourEventProcess,
PollParameters,
};
#[allow(deprecated)]
pub use crate::NetworkBehaviourEventProcess;
use crate::{DialError, NetworkBehaviour, NetworkBehaviourAction, PollParameters};
use either::Either;
use libp2p_core::{
connection::ConnectionId,
@ -233,6 +232,7 @@ where
}
}
#[allow(deprecated)]
impl<TEvent, TBehaviour> NetworkBehaviourEventProcess<TEvent> for Toggle<TBehaviour>
where
TBehaviour: NetworkBehaviourEventProcess<TEvent>,

5
swarm/src/lib.rs
View File

@ -63,9 +63,10 @@ pub mod behaviour;
pub mod dial_opts;
pub mod handler;
#[allow(deprecated)]
pub use behaviour::NetworkBehaviourEventProcess;
pub use behaviour::{
CloseConnection, NetworkBehaviour, NetworkBehaviourAction, NetworkBehaviourEventProcess,
NotifyHandler, PollParameters,
CloseConnection, NetworkBehaviour, NetworkBehaviourAction, NotifyHandler, PollParameters,
};
pub use connection::{
ConnectionCounters, ConnectionError, ConnectionLimit, ConnectionLimits, PendingConnectionError,