fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-05-30 19:21:19 +00:00
Code Issues Projects Releases Wiki Activity

Merge branch 'master' into rand-feature

Browse Source
This commit is contained in:
Ivan Boldyrev 2023-10-05 19:02:35 +04:00
commit 719a6e0cf1
12 changed files with 92 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
.github/workflows/interop-test.yml vendored
View File

@ -18,9 +18,23 @@ jobs:
flavour: [chromium, native]
steps:
- uses: actions/checkout@v4
- uses: docker/setup-buildx-action@v3
# This uses the same S3 cache as all test-plans images. Because we use `cargo-chef` in the Dockerfile, we have a layer available with all dependencies built.
- name: Build ${{ matrix.flavour }} image
run: docker buildx build --load -t ${{ matrix.flavour }}-rust-libp2p-head . -f interop-tests/Dockerfile.${{ matrix.flavour }}
run: |
docker buildx build \
--load \
--cache-to type=s3,mode=max,bucket=libp2p-by-tf-aws-bootstrap,region=us-east-1,prefix=buildCache,name=${{ matrix.flavour }}-rust-libp2p-head \
--cache-from type=s3,mode=max,bucket=libp2p-by-tf-aws-bootstrap,region=us-east-1,prefix=buildCache,name=${{ matrix.flavour }}-rust-libp2p-head \
-t ${{ matrix.flavour }}-rust-libp2p-head \
. \
-f interop-tests/Dockerfile.${{ matrix.flavour }}
env:
AWS_ACCESS_KEY_ID: ${{ vars.TEST_PLANS_BUILD_CACHE_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.TEST_PLANS_BUILD_CACHE_KEY }}
- name: Run ${{ matrix.flavour }} tests
uses: libp2p/test-plans/.github/actions/run-interop-ping-test@master
with:

4
Cargo.lock generated
View File

@ -2252,23 +2252,23 @@ dependencies = [
name = "ipfs-kad-example"
version = "0.1.0"
dependencies = [
"async-std",
"async-trait",
"env_logger 0.10.0",
"futures",
"libp2p",
"tokio",
]
[[package]]
name = "ipfs-private-example"
version = "0.1.0"
dependencies = [
"async-std",
"async-trait",
"either",
"env_logger 0.10.0",
"futures",
"libp2p",
"tokio",
]
[[package]]

52
examples/dcutr/src/main.rs
View File

@ -110,7 +110,6 @@ fn main() -> Result<(), Box<dyn Error>> {
};
#[derive(NetworkBehaviour)]
#[behaviour(to_swarm = "Event")]
struct Behaviour {
relay_client: relay::client::Behaviour,
ping: ping::Behaviour,
@ -118,39 +117,6 @@ fn main() -> Result<(), Box<dyn Error>> {
dcutr: dcutr::Behaviour,
}
#[derive(Debug)]
#[allow(clippy::large_enum_variant)]
enum Event {
Ping(ping::Event),
Identify(identify::Event),
Relay(relay::client::Event),
Dcutr(dcutr::Event),
}
impl From<ping::Event> for Event {
fn from(e: ping::Event) -> Self {
Event::Ping(e)
}
}
impl From<identify::Event> for Event {
fn from(e: identify::Event) -> Self {
Event::Identify(e)
}
}
impl From<relay::client::Event> for Event {
fn from(e: relay::client::Event) -> Self {
Event::Relay(e)
}
}
impl From<dcutr::Event> for Event {
fn from(e: dcutr::Event) -> Self {
Event::Dcutr(e)
}
}
let behaviour = Behaviour {
relay_client: client,
ping: ping::Behaviour::new(ping::Config::new()),
@ -207,12 +173,14 @@ fn main() -> Result<(), Box<dyn Error>> {
SwarmEvent::NewListenAddr { .. } => {}
SwarmEvent::Dialing { .. } => {}
SwarmEvent::ConnectionEstablished { .. } => {}
SwarmEvent::Behaviour(Event::Ping(_)) => {}
SwarmEvent::Behaviour(Event::Identify(identify::Event::Sent { .. })) => {
SwarmEvent::Behaviour(BehaviourEvent::Ping(_)) => {}
SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Sent {
..
})) => {
info!("Told relay its public address.");
told_relay_observed_addr = true;
}
SwarmEvent::Behaviour(Event::Identify(identify::Event::Received {
SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received {
info: identify::Info { observed_addr, .. },
..
})) => {
@ -252,22 +220,22 @@ fn main() -> Result<(), Box<dyn Error>> {
SwarmEvent::NewListenAddr { address, .. } => {
info!("Listening on {:?}", address);
}
SwarmEvent::Behaviour(Event::Relay(
SwarmEvent::Behaviour(BehaviourEvent::RelayClient(
relay::client::Event::ReservationReqAccepted { .. },
)) => {
assert!(opts.mode == Mode::Listen);
info!("Relay accepted our reservation request.");
}
SwarmEvent::Behaviour(Event::Relay(event)) => {
SwarmEvent::Behaviour(BehaviourEvent::RelayClient(event)) => {
info!("{:?}", event)
}
SwarmEvent::Behaviour(Event::Dcutr(event)) => {
SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => {
info!("{:?}", event)
}
SwarmEvent::Behaviour(Event::Identify(event)) => {
SwarmEvent::Behaviour(BehaviourEvent::Identify(event)) => {
info!("{:?}", event)
}
SwarmEvent::Behaviour(Event::Ping(_)) => {}
SwarmEvent::Behaviour(BehaviourEvent::Ping(_)) => {}
SwarmEvent::ConnectionEstablished {
peer_id, endpoint, ..
} => {

27
examples/distributed-key-value-store/src/main.rs
View File

@ -49,37 +49,18 @@ async fn main() -> Result<(), Box<dyn Error>> {
// We create a custom network behaviour that combines Kademlia and mDNS.
#[derive(NetworkBehaviour)]
#[behaviour(to_swarm = "MyBehaviourEvent")]
struct MyBehaviour {
struct Behaviour {
kademlia: kad::Behaviour<MemoryStore>,
mdns: mdns::async_io::Behaviour,
}
#[allow(clippy::large_enum_variant)]
enum MyBehaviourEvent {
Kademlia(kad::Event),
Mdns(mdns::Event),
}
impl From<kad::Event> for MyBehaviourEvent {
fn from(event: kad::Event) -> Self {
MyBehaviourEvent::Kademlia(event)
}
}
impl From<mdns::Event> for MyBehaviourEvent {
fn from(event: mdns::Event) -> 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 = kad::Behaviour::new(local_peer_id, store);
let mdns = mdns::async_io::Behaviour::new(mdns::Config::default(), local_peer_id)?;
let behaviour = MyBehaviour { kademlia, mdns };
let behaviour = Behaviour { kademlia, mdns };
SwarmBuilder::with_async_std_executor(transport, behaviour, local_peer_id).build()
};
@ -99,12 +80,12 @@ async fn main() -> Result<(), Box<dyn Error>> {
SwarmEvent::NewListenAddr { address, .. } => {
println!("Listening in {address:?}");
},
SwarmEvent::Behaviour(MyBehaviourEvent::Mdns(mdns::Event::Discovered(list))) => {
SwarmEvent::Behaviour(BehaviourEvent::Mdns(mdns::Event::Discovered(list))) => {
for (peer_id, multiaddr) in list {
swarm.behaviour_mut().kademlia.add_address(&peer_id, multiaddr);
}
}
SwarmEvent::Behaviour(MyBehaviourEvent::Kademlia(kad::Event::OutboundQueryProgressed { result, ..})) => {
SwarmEvent::Behaviour(BehaviourEvent::Kademlia(kad::Event::OutboundQueryProgressed { result, ..})) => {
match result {
kad::QueryResult::GetProviders(Ok(kad::GetProvidersOk::FoundProviders { key, providers, .. })) => {
for peer in providers {

43
examples/file-sharing/src/network.rs
View File

@ -51,7 +51,7 @@ pub(crate) async fn new(
// higher layer network behaviour logic.
let mut swarm = SwarmBuilder::with_async_std_executor(
transport,
ComposedBehaviour {
Behaviour {
kademlia: kad::Behaviour::new(peer_id, kad::record::store::MemoryStore::new(peer_id)),
request_response: request_response::cbor::Behaviour::new(
[(
@ -171,7 +171,7 @@ impl Client {
}
pub(crate) struct EventLoop {
swarm: Swarm<ComposedBehaviour>,
swarm: Swarm<Behaviour>,
command_receiver: mpsc::Receiver<Command>,
event_sender: mpsc::Sender<Event>,
pending_dial: HashMap<PeerId, oneshot::Sender<Result<(), Box<dyn Error + Send>>>>,
@ -183,7 +183,7 @@ pub(crate) struct EventLoop {
impl EventLoop {
fn new(
swarm: Swarm<ComposedBehaviour>,
swarm: Swarm<Behaviour>,
command_receiver: mpsc::Receiver<Command>,
event_sender: mpsc::Sender<Event>,
) -> Self {
@ -213,10 +213,10 @@ impl EventLoop {
async fn handle_event(
&mut self,
event: SwarmEvent<ComposedEvent, Either<void::Void, io::Error>>,
event: SwarmEvent<BehaviourEvent, Either<void::Void, io::Error>>,
) {
match event {
SwarmEvent::Behaviour(ComposedEvent::Kademlia(
SwarmEvent::Behaviour(BehaviourEvent::Kademlia(
kad::Event::OutboundQueryProgressed {
id,
result: kad::QueryResult::StartProviding(_),
@ -229,7 +229,7 @@ impl EventLoop {
.expect("Completed query to be previously pending.");
let _ = sender.send(());
}
SwarmEvent::Behaviour(ComposedEvent::Kademlia(
SwarmEvent::Behaviour(BehaviourEvent::Kademlia(
kad::Event::OutboundQueryProgressed {
id,
result:
@ -252,7 +252,7 @@ impl EventLoop {
.finish();
}
}
SwarmEvent::Behaviour(ComposedEvent::Kademlia(
SwarmEvent::Behaviour(BehaviourEvent::Kademlia(
kad::Event::OutboundQueryProgressed {
result:
kad::QueryResult::GetProviders(Ok(
@ -261,8 +261,8 @@ impl EventLoop {
..
},
)) => {}
SwarmEvent::Behaviour(ComposedEvent::Kademlia(_)) => {}
SwarmEvent::Behaviour(ComposedEvent::RequestResponse(
SwarmEvent::Behaviour(BehaviourEvent::Kademlia(_)) => {}
SwarmEvent::Behaviour(BehaviourEvent::RequestResponse(
request_response::Event::Message { message, .. },
)) => match message {
request_response::Message::Request {
@ -287,7 +287,7 @@ impl EventLoop {
.send(Ok(response.0));
}
},
SwarmEvent::Behaviour(ComposedEvent::RequestResponse(
SwarmEvent::Behaviour(BehaviourEvent::RequestResponse(
request_response::Event::OutboundFailure {
request_id, error, ..
},
@ -298,7 +298,7 @@ impl EventLoop {
.expect("Request to still be pending.")
.send(Err(Box::new(error)));
}
SwarmEvent::Behaviour(ComposedEvent::RequestResponse(
SwarmEvent::Behaviour(BehaviourEvent::RequestResponse(
request_response::Event::ResponseSent { .. },
)) => {}
SwarmEvent::NewListenAddr { address, .. } => {
@ -406,30 +406,11 @@ impl EventLoop {
}
#[derive(NetworkBehaviour)]
#[behaviour(to_swarm = "ComposedEvent")]
struct ComposedBehaviour {
struct Behaviour {
request_response: request_response::cbor::Behaviour<FileRequest, FileResponse>,
kademlia: kad::Behaviour<kad::record::store::MemoryStore>,
}
#[derive(Debug)]
enum ComposedEvent {
RequestResponse(request_response::Event<FileRequest, FileResponse>),
Kademlia(kad::Event),
}
impl From<request_response::Event<FileRequest, FileResponse>> for ComposedEvent {
fn from(event: request_response::Event<FileRequest, FileResponse>) -> Self {
ComposedEvent::RequestResponse(event)
}
}
impl From<kad::Event> for ComposedEvent {
fn from(event: kad::Event) -> Self {
ComposedEvent::Kademlia(event)
}
}
#[derive(Debug)]
enum Command {
StartListening {

4
examples/ipfs-kad/Cargo.toml
View File

@ -6,8 +6,8 @@ publish = false
license = "MIT"
[dependencies]
async-std = { version = "1.12", features = ["attributes"] }
tokio = { version = "1.12", features = ["rt-multi-thread", "macros"] }
async-trait = "0.1"
env_logger = "0.10"
futures = "0.3.28"
libp2p = { path = "../../libp2p", features = ["async-std", "dns", "kad", "noise", "tcp", "websocket", "yamux", "rsa"] }
libp2p = { path = "../../libp2p", features = ["tokio", "dns", "kad", "noise", "tcp", "websocket", "yamux", "rsa"] }

10
examples/ipfs-kad/src/main.rs
View File

@ -24,9 +24,9 @@ use futures::StreamExt;
use libp2p::kad;
use libp2p::kad::record::store::MemoryStore;
use libp2p::{
development_transport, identity,
identity,
swarm::{SwarmBuilder, SwarmEvent},
PeerId,
tokio_development_transport, PeerId,
};
use std::{env, error::Error, time::Duration};
@ -37,7 +37,7 @@ const BOOTNODES: [&str; 4] = [
"QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
];
#[async_std::main]
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
env_logger::init();
@ -46,7 +46,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
let local_peer_id = PeerId::from(local_key.public());
// Set up a an encrypted DNS-enabled TCP Transport over the yamux protocol
let transport = development_transport(local_key).await?;
let transport = tokio_development_transport(local_key)?;
// Create a swarm to manage peers and events.
let mut swarm = {
@ -63,7 +63,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
behaviour.add_address(&peer.parse()?, "/dnsaddr/bootstrap.libp2p.io".parse()?);
}
SwarmBuilder::with_async_std_executor(transport, behaviour, local_peer_id).build()
SwarmBuilder::with_tokio_executor(transport, behaviour, local_peer_id).build()
};
// Order Kademlia to search for a peer.

4
examples/ipfs-private/Cargo.toml
View File

@ -6,9 +6,9 @@ publish = false
license = "MIT"
[dependencies]
async-std = { version = "1.12", features = ["attributes"] }
tokio = { version = "1.32", features = ["rt-multi-thread", "macros", "io-std"] }
async-trait = "0.1"
either = "1.9"
env_logger = "0.10"
futures = "0.3.28"
libp2p = { path = "../../libp2p", features = ["async-std", "gossipsub", "dns", "identify", "kad", "macros", "noise", "ping", "pnet", "tcp", "websocket", "yamux"] }
libp2p = { path = "../../libp2p", features = ["tokio", "gossipsub", "dns", "identify", "kad", "macros", "noise", "ping", "pnet", "tcp", "websocket", "yamux"] }

16
examples/ipfs-private/src/main.rs
View File

@ -20,9 +20,8 @@
#![doc = include_str!("../README.md")]
use async_std::io;
use either::Either;
use futures::{prelude::*, select};
use futures::prelude::*;
use libp2p::{
core::{muxing::StreamMuxerBox, transport, transport::upgrade::Version},
gossipsub, identify, identity,
@ -33,6 +32,7 @@ use libp2p::{
tcp, yamux, Multiaddr, PeerId, Transport,
};
use std::{env, error::Error, fs, path::Path, str::FromStr, time::Duration};
use tokio::{io, io::AsyncBufReadExt, select};
/// Builds the transport that serves as a common ground for all connections.
pub fn build_transport(
@ -42,7 +42,7 @@ pub fn build_transport(
let noise_config = noise::Config::new(&key_pair).unwrap();
let yamux_config = yamux::Config::default();
let base_transport = tcp::async_io::Transport::new(tcp::Config::default().nodelay(true));
let base_transport = tcp::tokio::Transport::new(tcp::Config::default().nodelay(true));
let maybe_encrypted = match psk {
Some(psk) => Either::Left(
base_transport.and_then(move |socket, _| PnetConfig::new(psk).handshake(socket)),
@ -108,7 +108,7 @@ fn parse_legacy_multiaddr(text: &str) -> Result<Multiaddr, Box<dyn Error>> {
Ok(res)
}
#[async_std::main]
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
env_logger::init();
@ -186,7 +186,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
println!("Subscribing to {gossipsub_topic:?}");
behaviour.gossipsub.subscribe(&gossipsub_topic).unwrap();
SwarmBuilder::with_async_std_executor(transport, behaviour, local_peer_id).build()
SwarmBuilder::with_tokio_executor(transport, behaviour, local_peer_id).build()
};
// Reach out to other nodes if specified
@ -197,7 +197,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
}
// Read full lines from stdin
let mut stdin = io::BufReader::new(io::stdin()).lines().fuse();
let mut stdin = io::BufReader::new(io::stdin()).lines();
// Listen on all interfaces and whatever port the OS assigns
swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
@ -205,11 +205,11 @@ async fn main() -> Result<(), Box<dyn Error>> {
// Kick it off
loop {
select! {
line = stdin.select_next_some() => {
Ok(Some(line)) = stdin.next_line() => {
if let Err(e) = swarm
.behaviour_mut()
.gossipsub
.publish(gossipsub_topic.clone(), line.expect("Stdin not to close").as_bytes())
.publish(gossipsub_topic.clone(), line.as_bytes())
{
println!("Publish error: {e:?}");
}

36
interop-tests/Dockerfile.chromium
View File

@ -1,28 +1,26 @@
FROM rust:1.67.0 as builder
# Run with access to the target cache to speed up builds
WORKDIR /workspace
ADD . .
# syntax=docker/dockerfile:1.5-labs
FROM rust:1.67.0 as chef
RUN rustup target add wasm32-unknown-unknown
RUN wget -q -O- https://github.com/rustwasm/wasm-pack/releases/download/v0.12.1/wasm-pack-v0.12.1-x86_64-unknown-linux-musl.tar.gz | tar -zx -C /usr/local/bin --strip-components 1 --wildcards "wasm-pack-*/wasm-pack"
RUN wget -q -O- https://github.com/WebAssembly/binaryen/releases/download/version_115/binaryen-version_115-x86_64-linux.tar.gz | tar -zx -C /usr/local/bin --strip-components 2 --wildcards "binaryen-version_*/bin/wasm-opt"
RUN wget -q -O- https://github.com/LukeMathWalker/cargo-chef/releases/download/v0.1.62/cargo-chef-x86_64-unknown-linux-gnu.tar.gz | tar -zx -C /usr/local/bin
WORKDIR /app
RUN --mount=type=cache,target=./target \
--mount=type=cache,target=/usr/local/cargo/registry \
wasm-pack build --target web interop-tests
FROM chef AS planner
COPY . .
RUN cargo chef prepare --recipe-path recipe.json
RUN --mount=type=cache,target=./target \
--mount=type=cache,target=/usr/local/cargo/registry \
cargo build --release --package interop-tests --bin wasm_ping
RUN --mount=type=cache,target=./target \
mv ./target/release/wasm_ping /usr/local/bin/testplan
FROM chef AS builder
COPY --from=planner /app/recipe.json recipe.json
# Build dependencies - this is the caching Docker layer!
RUN cargo chef cook --release --package interop-tests --target wasm32-unknown-unknown --recipe-path recipe.json
RUN cargo chef cook --release --package interop-tests --bin wasm_ping --recipe-path recipe.json
# Build application
COPY . .
RUN wasm-pack build --target web interop-tests
RUN cargo build --release --package interop-tests --bin wasm_ping
FROM selenium/standalone-chrome:115.0
COPY --from=builder /usr/local/bin/testplan /usr/local/bin/testplan
COPY --from=builder /app/target/release/wasm_ping /usr/local/bin/testplan
ENV RUST_BACKTRACE=1
ENTRYPOINT ["testplan"]

24
interop-tests/Dockerfile.native
View File

@ -1,17 +1,21 @@
# syntax=docker/dockerfile:1.5-labs
FROM rust:1.67.0 as builder
FROM rust:1.67.0 as chef
RUN wget -q -O- https://github.com/LukeMathWalker/cargo-chef/releases/download/v0.1.62/cargo-chef-x86_64-unknown-linux-gnu.tar.gz | tar -zx -C /usr/local/bin
WORKDIR /app
# Run with access to the target cache to speed up builds
WORKDIR /workspace
ADD . .
RUN --mount=type=cache,target=./target \
--mount=type=cache,target=/usr/local/cargo/registry \
cargo build --release --package interop-tests --bin native_ping
FROM chef AS planner
COPY . .
RUN cargo chef prepare --recipe-path recipe.json
RUN --mount=type=cache,target=./target \
mv ./target/release/native_ping /usr/local/bin/testplan
FROM chef AS builder
COPY --from=planner /app/recipe.json recipe.json
# Build dependencies - this is the caching Docker layer!
RUN cargo chef cook --release --package interop-tests --bin native_ping --recipe-path recipe.json
# Build application
COPY . .
RUN cargo build --release --package interop-tests --bin native_ping
FROM gcr.io/distroless/cc
COPY --from=builder /usr/local/bin/testplan /usr/local/bin/testplan
COPY --from=builder /app/target/release/native_ping /usr/local/bin/testplan
ENV RUST_BACKTRACE=1
ENTRYPOINT ["testplan"]

20
protocols/dcutr/tests/lib.rs
View File

@ -135,30 +135,12 @@ fn build_client() -> Swarm<Client> {
}
#[derive(NetworkBehaviour)]
#[behaviour(to_swarm = "ClientEvent", prelude = "libp2p_swarm::derive_prelude")]
#[behaviour(prelude = "libp2p_swarm::derive_prelude")]
struct Client {
relay: relay::client::Behaviour,
dcutr: dcutr::Behaviour,
}
#[derive(Debug)]
enum ClientEvent {
Relay(relay::client::Event),
Dcutr(dcutr::Event),
}
impl From<relay::client::Event> for ClientEvent {
fn from(event: relay::client::Event) -> Self {
ClientEvent::Relay(event)
}
}
impl From<dcutr::Event> for ClientEvent {
fn from(event: dcutr::Event) -> Self {
ClientEvent::Dcutr(event)
}
}
async fn wait_for_reservation(
client: &mut Swarm<Client>,
client_addr: Multiaddr,