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

Fix examples and update core-derive. #1317

Merged
merged 3 commits into from
Nov 25, 2019
Merged

Fix examples and update core-derive. #1317

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9018812
Fix examples and update core-derive.
twittner Nov 22, 2019
4b055ab
Simplify.
twittner Nov 22, 2019
4ecdcdb
Remove anyhow dependency.
twittner Nov 23, 2019
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
4 changes: 2 additions & 2 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,9 +45,9 @@ libp2p-tcp = { version = "0.13.0", path = "transports/tcp" }
libp2p-websocket = { version = "0.13.0", path = "transports/websocket", optional = true }

[dev-dependencies]
anyhow = "1.0"
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is an opinionated way to handle errors, I'd be in favour of not using it in examples.

async-std = "1.0"
env_logger = "0.7.1"
tokio = "0.1"
tokio-stdin-stdout = "0.1"

[workspace]
members = [
Expand Down
66 changes: 28 additions & 38 deletions examples/chat.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,20 +49,22 @@
//!
//! The two nodes then connect.

use futures::prelude::*;
use anyhow::{Context as _, Result};
use async_std::{io, task};
use futures::{future, prelude::*};
use libp2p::{
Multiaddr,
PeerId,
Swarm,
NetworkBehaviour,
identity,
tokio_codec::{FramedRead, LinesCodec},
tokio_io::{AsyncRead, AsyncWrite},
floodsub::{self, Floodsub, FloodsubEvent},
mdns::{Mdns, MdnsEvent},
swarm::NetworkBehaviourEventProcess
};
use std::task::{Context, Poll};

fn main() {
fn main() -> Result<()> {
env_logger::init();

// Create a random PeerId
Expand All @@ -71,7 +73,7 @@ fn main() {
println!("Local peer id: {:?}", local_peer_id);

// Set up a an encrypted DNS-enabled TCP Transport over the Mplex and Yamux protocols
let transport = libp2p::build_development_transport(local_key);
let transport = libp2p::build_development_transport(local_key)?;

// Create a Floodsub topic
let floodsub_topic = floodsub::TopicBuilder::new("chat").build();
Expand All @@ -87,18 +89,16 @@ fn main() {
impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<MdnsEvent> for MyBehaviour<TSubstream> {
fn inject_event(&mut self, event: MdnsEvent) {
match event {
MdnsEvent::Discovered(list) => {
MdnsEvent::Discovered(list) =>
for (peer, _) in list {
self.floodsub.add_node_to_partial_view(peer);
}
},
MdnsEvent::Expired(list) => {
MdnsEvent::Expired(list) =>
for (peer, _) in list {
if !self.mdns.has_node(&peer) {
self.floodsub.remove_node_from_partial_view(&peer);
}
}
}
}
}
}
Expand All @@ -114,9 +114,10 @@ fn main() {

// Create a Swarm to manage peers and events
let mut swarm = {
let mdns = task::block_on(Mdns::new()).context("Failed to create mDNS service")?;
let mut behaviour = MyBehaviour {
floodsub: Floodsub::new(local_peer_id.clone()),
mdns: Mdns::new().expect("Failed to create mDNS service"),
mdns
};

behaviour.floodsub.subscribe(floodsub_topic.clone());
Expand All @@ -125,42 +126,32 @@ fn main() {

// Reach out to another node if specified
if let Some(to_dial) = std::env::args().nth(1) {
let dialing = to_dial.clone();
match to_dial.parse() {
Ok(to_dial) => {
match libp2p::Swarm::dial_addr(&mut swarm, to_dial) {
Ok(_) => println!("Dialed {:?}", dialing),
Err(e) => println!("Dial {:?} failed: {:?}", dialing, e)
}
},
Err(err) => println!("Failed to parse address to dial: {:?}", err),
}
let addr: Multiaddr = to_dial.parse().context("Failed to parse address to dial")?;
Swarm::dial_addr(&mut swarm, addr.clone()).with_context(|| format!("Failed to dial {}", addr))?;
println!("Dialed {:?}", addr)
}

// Read full lines from stdin
let stdin = tokio_stdin_stdout::stdin(0);
let mut framed_stdin = FramedRead::new(stdin, LinesCodec::new());
let mut stdin = io::BufReader::new(io::stdin()).lines();

// Listen on all interfaces and whatever port the OS assigns
Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse().unwrap()).unwrap();
Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse()?)?;

// Kick it off
let mut listening = false;
tokio::run(futures::future::poll_fn(move || -> Result<_, ()> {
task::block_on(future::poll_fn(move |cx: &mut Context| {
loop {
match framed_stdin.poll().expect("Error while polling stdin") {
Async::Ready(Some(line)) => swarm.floodsub.publish(&floodsub_topic, line.as_bytes()),
Async::Ready(None) => panic!("Stdin closed"),
Async::NotReady => break,
};
match stdin.try_poll_next_unpin(cx)? {
Poll::Ready(Some(line)) => swarm.floodsub.publish(&floodsub_topic, line.as_bytes()),
Poll::Ready(None) => panic!("Stdin closed"),
Poll::Pending => break
}
}

loop {
match swarm.poll().expect("Error while polling swarm") {
Async::Ready(Some(_)) => {

},
Async::Ready(None) | Async::NotReady => {
match swarm.poll_next_unpin(cx) {
Poll::Ready(Some(event)) => println!("{:?}", event),
Poll::Ready(None) => return Poll::Ready(Ok(())),
Poll::Pending => {
if !listening {
if let Some(a) = Swarm::listeners(&swarm).next() {
println!("Listening on {:?}", a);
Expand All @@ -171,7 +162,6 @@ fn main() {
}
}
}

Ok(Async::NotReady)
}));
Poll::Pending
}))
}
82 changes: 40 additions & 42 deletions examples/distributed-key-value-store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,37 +29,42 @@
//!
//! 4. Close with Ctrl-c.

use anyhow::{Context as _, Result};
use async_std::{io, task};
use futures::prelude::*;
use libp2p::kad::record::store::MemoryStore;
use libp2p::kad::{record::Key, Kademlia, KademliaEvent, PutRecordOk, Quorum, Record};
use libp2p::{
build_development_transport, identity,
NetworkBehaviour,
PeerId,
Swarm,
build_development_transport,
identity,
mdns::{Mdns, MdnsEvent},
swarm::NetworkBehaviourEventProcess,
tokio_codec::{FramedRead, LinesCodec},
tokio_io::{AsyncRead, AsyncWrite},
NetworkBehaviour, PeerId, Swarm,
swarm::NetworkBehaviourEventProcess
};
use std::task::{Context, Poll};

fn main() {
fn main() -> Result<()> {
env_logger::init();

// Create a random key for ourselves.
let local_key = identity::Keypair::generate_ed25519();
let local_peer_id = PeerId::from(local_key.public());

// Set up a an encrypted DNS-enabled TCP Transport over the Mplex protocol.
let transport = build_development_transport(local_key);
let transport = build_development_transport(local_key)?;

// We create a custom network behaviour that combines Kademlia and mDNS.
#[derive(NetworkBehaviour)]
struct MyBehaviour<TSubstream: AsyncRead + AsyncWrite> {
kademlia: Kademlia<TSubstream, MemoryStore>,
mdns: Mdns<TSubstream>,
mdns: Mdns<TSubstream>
}

impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<MdnsEvent>
for MyBehaviour<TSubstream>
impl<T> NetworkBehaviourEventProcess<MdnsEvent> for MyBehaviour<T>
where
T: AsyncRead + AsyncWrite
{
// Called when `mdns` produces an event.
fn inject_event(&mut self, event: MdnsEvent) {
Expand All @@ -71,8 +76,9 @@ fn main() {
}
}

impl<TSubstream: AsyncRead + AsyncWrite> NetworkBehaviourEventProcess<KademliaEvent>
for MyBehaviour<TSubstream>
impl<T> NetworkBehaviourEventProcess<KademliaEvent> for MyBehaviour<T>
where
T: AsyncRead + AsyncWrite
{
// Called when `kademlia` produces an event.
fn inject_event(&mut self, message: KademliaEvent) {
Expand Down Expand Up @@ -108,58 +114,50 @@ fn main() {
// Create a Kademlia behaviour.
let store = MemoryStore::new(local_peer_id.clone());
let kademlia = Kademlia::new(local_peer_id.clone(), store);

let behaviour = MyBehaviour {
kademlia,
mdns: Mdns::new().expect("Failed to create mDNS service"),
};

let mdns = task::block_on(Mdns::new()).context("Failed to create mDNS service")?;
let behaviour = MyBehaviour { kademlia, mdns };
Swarm::new(transport, behaviour, local_peer_id)
};

// Read full lines from stdin.
let stdin = tokio_stdin_stdout::stdin(0);
let mut framed_stdin = FramedRead::new(stdin, LinesCodec::new());
// Read full lines from stdin
let mut stdin = io::BufReader::new(io::stdin()).lines();

// Listen on all interfaces and whatever port the OS assigns.
Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse().unwrap()).unwrap();
Swarm::listen_on(&mut swarm, "/ip4/0.0.0.0/tcp/0".parse()?)?;

// Kick it off.
let mut listening = false;
tokio::run(futures::future::poll_fn(move || {
task::block_on(future::poll_fn(move |cx: &mut Context| {
loop {
match framed_stdin.poll().expect("Error while polling stdin") {
Async::Ready(Some(line)) => {
handle_input_line(&mut swarm.kademlia, line);
}
Async::Ready(None) => panic!("Stdin closed"),
Async::NotReady => break,
};
match stdin.try_poll_next_unpin(cx)? {
Poll::Ready(Some(line)) => handle_input_line(&mut swarm.kademlia, line),
Poll::Ready(None) => panic!("Stdin closed"),
Poll::Pending => break
}
}

loop {
match swarm.poll().expect("Error while polling swarm") {
Async::Ready(Some(_)) => {}
Async::Ready(None) | Async::NotReady => {
match swarm.poll_next_unpin(cx) {
Poll::Ready(Some(event)) => println!("{:?}", event),
Poll::Ready(None) => return Poll::Ready(Ok(())),
Poll::Pending => {
if !listening {
if let Some(a) = Swarm::listeners(&swarm).next() {
println!("Listening on {:?}", a);
listening = true;
}
}
break;
break
}
}
}

Ok(Async::NotReady)
}));
Poll::Pending
}))
}

fn handle_input_line<TSubstream: AsyncRead + AsyncWrite>(
kademlia: &mut Kademlia<TSubstream, MemoryStore>,
line: String,
) {
fn handle_input_line<T>(kademlia: &mut Kademlia<T, MemoryStore>, line: String)
where
T: AsyncRead + AsyncWrite
{
let mut args = line.split(" ");

match args.next() {
Expand Down
Loading