tsarevvv/rust-ipfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: change the async executor to tokio

Browse Source 
Signed-off-by: ljedrz <ljedrz@gmail.com>
This commit is contained in:
ljedrz 2020-08-13 11:37:21 +02:00
parent 4c90dd15a9
commit 8be75f7a4a
28 changed files with 222 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
Cargo.lock generated
View File

@ -115,16 +115,6 @@ dependencies = [
"syn",
]
[[package]]
name = "async-attributes"
version = "1.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "efd3d156917d94862e779f356c5acae312b08fd3121e792c857d7928c8088423"
dependencies = [
"quote",
"syn",
]
[[package]]
name = "async-channel"
version = "1.4.0"
@ -142,7 +132,6 @@ version = "1.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "00d68a33ebc8b57800847d00787307f84a562224a14db069b0acefe4c2abbf5d"
dependencies = [
"async-attributes",
"async-task",
"crossbeam-utils",
"futures-channel",
@ -255,7 +244,6 @@ checksum = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693"
name = "bitswap"
version = "0.1.0"
dependencies = [
"async-std",
"async-trait",
"cid",
"fnv",
@ -266,6 +254,7 @@ dependencies = [
"prost",
"prost-build",
"thiserror",
"tokio",
"tracing",
"unsigned-varint 0.3.3",
]
@ -1354,7 +1343,6 @@ name = "ipfs"
version = "0.1.0"
dependencies = [
"anyhow",
"async-std",
"async-stream",
"async-trait",
"base64 0.12.3",
@ -1378,6 +1366,7 @@ dependencies = [
"serde_json",
"sha2 0.9.1",
"thiserror",
"tokio",
"tracing",
"tracing-futures",
"tracing-subscriber",
@ -1785,7 +1774,6 @@ version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b1fa2bbad054020cb875546a577a66a65a5bf42eff55ed5265f92ffee3cc052"
dependencies = [
"async-std",
"futures 0.3.5",
"futures-timer",
"get_if_addrs",
@ -1793,6 +1781,7 @@ dependencies = [
"libp2p-core",
"log",
"socket2",
"tokio",
]
[[package]]

4
Cargo.toml
View File

@ -11,7 +11,6 @@ nightly = []
[dependencies]
anyhow = { default-features = false, version = "1.0" }
async-std = { default-features = false, features = ["attributes", "std"], version = "1.6" }
async-stream = { default-features = false, version = "0.3" }
async-trait = { default-features = false, version = "0.1" }
base64 = { default-features = false, features = ["alloc"], version = "0.12" }
@ -24,7 +23,7 @@ domain = { default-features = false, version = "0.5" }
domain-resolv = { default-features = false, version = "0.5" }
futures = { default-features = false, features = ["compat", "io-compat"], version = "0.3.5" }
ipfs-unixfs = { path = "unixfs" }
libp2p = { default-features = false, features = ["floodsub", "identify", "kad", "tcp-async-std", "mdns", "mplex", "noise", "ping", "yamux"], version = "0.23" }
libp2p = { default-features = false, features = ["floodsub", "identify", "kad", "tcp-tokio", "mdns", "mplex", "noise", "ping", "yamux"], version = "0.23" }
multibase = { default-features = false, version = "0.8" }
multihash = { default-features = false, version = "0.11" }
prost = { default-features = false, version = "0.6" }
@ -32,6 +31,7 @@ rand = { default-features = false, features = ["getrandom"], version = "0.7" }
serde = { default-features = false, features = ["derive"], version = "1.0" }
serde_json = { default-features = false, features = ["std"], version = "1.0" }
thiserror = { default-features = false, version = "1.0" }
tokio = { default-features = false, features = ["full"], version = "0.2" }
tracing = { default-features = false, features = ["log"], version = "0.1" }
tracing-futures = { default-features = false, features = ["std", "futures-03"], version = "0.2" }
void = { default-features = false, version = "1.0" }

47
README.md
View File

@ -77,38 +77,37 @@ _Note: binaries available via `cargo install` is coming soon._
## Getting started
```rust,no_run
use async_std::task;
use tokio::task;
use futures::join;
use ipfs::{make_ipld, Ipfs, IpfsPath, Ipld, Types, UninitializedIpfs};
fn main() {
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
task::block_on(async move {
// Start daemon and initialize repo
let (ipfs, fut): (Ipfs<Types>, _) = UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
// Start daemon and initialize repo
let (ipfs, fut): (Ipfs<Types>, _) = UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
// Create a DAG
let f1 = ipfs.put_dag(make_ipld!("block1"));
let f2 = ipfs.put_dag(make_ipld!("block2"));
let (res1, res2) = join!(f1, f2);
let root = make_ipld!([res1.unwrap(), res2.unwrap()]);
let cid = ipfs.put_dag(root).await.unwrap();
let path = IpfsPath::from(cid);
// Create a DAG
let f1 = ipfs.put_dag(make_ipld!("block1"));
let f2 = ipfs.put_dag(make_ipld!("block2"));
let (res1, res2) = join!(f1, f2);
let root = make_ipld!([res1.unwrap(), res2.unwrap()]);
let cid = ipfs.put_dag(root).await.unwrap();
let path = IpfsPath::from(cid);
// Query the DAG
let path1 = path.sub_path("0").unwrap();
let path2 = path.sub_path("1").unwrap();
let f1 = ipfs.get_dag(path1);
let f2 = ipfs.get_dag(path2);
let (res1, res2) = join!(f1, f2);
println!("Received block with contents: {:?}", res1.unwrap());
println!("Received block with contents: {:?}", res2.unwrap());
// Query the DAG
let path1 = path.sub_path("0").unwrap();
let path2 = path.sub_path("1").unwrap();
let f1 = ipfs.get_dag(path1);
let f2 = ipfs.get_dag(path2);
let (res1, res2) = join!(f1, f2);
println!("Received block with contents: {:?}", res1.unwrap());
println!("Received block with contents: {:?}", res2.unwrap());
// Exit
ipfs.exit_daemon();
});
// Exit
ipfs.exit_daemon();
}
```

2
bitswap/Cargo.toml
View File

@ -8,7 +8,6 @@ version = "0.1.0"
prost-build = { default-features = false, version = "0.6" }
[dependencies]
async-std = { default-features = false, version = "1.6" }
async-trait = { default-features = false, version = "0.1" }
cid = { default-features = false, version = "0.5" }
fnv = { default-features = false, version = "1.0" }
@ -18,5 +17,6 @@ libp2p-swarm = { default-features = false, version = "0.20" }
multihash = { default-features = false, version = "0.11" }
prost = { default-features = false, version = "0.6" }
thiserror = { default-features = false, version = "1.0" }
tokio = { default-features = false, features = ["full"], version = "0.2" }
tracing = { default-features = false, version = "0.1" }
unsigned-varint = { default-features = false, version = "0.3" }

25
examples/client1.rs
View File

@ -1,22 +1,21 @@
use async_std::task;
use tokio::task;
use futures::join;
use ipfs::{make_ipld, Ipfs, TestTypes, UninitializedIpfs};
fn main() {
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
task::block_on(async move {
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
let f1 = ipfs.put_dag(make_ipld!("block1"));
let f2 = ipfs.put_dag(make_ipld!("block2"));
let (res1, res2) = join!(f1, f2);
let f1 = ipfs.put_dag(make_ipld!("block1"));
let f2 = ipfs.put_dag(make_ipld!("block2"));
let (res1, res2) = join!(f1, f2);
let root = make_ipld!([res1.unwrap(), res2.unwrap()]);
ipfs.put_dag(root).await.unwrap();
let root = make_ipld!([res1.unwrap(), res2.unwrap()]);
ipfs.put_dag(root).await.unwrap();
ipfs.exit_daemon().await;
});
ipfs.exit_daemon().await;
}

25
examples/client2.rs
View File

@ -1,25 +1,24 @@
use async_std::task;
use tokio::task;
use futures::join;
use ipfs::{Ipfs, IpfsPath, TestTypes, UninitializedIpfs};
use std::str::FromStr;
fn main() {
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
let path =
IpfsPath::from_str("/ipfs/zdpuB1caPcm4QNXeegatVfLQ839Lmprd5zosXGwRUBJHwj66X").unwrap();
task::block_on(async move {
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
let f1 = ipfs.get_dag(path.sub_path("0").unwrap());
let f2 = ipfs.get_dag(path.sub_path("1").unwrap());
let (res1, res2) = join!(f1, f2);
println!("Received block with contents: {:?}", res1.unwrap());
println!("Received block with contents: {:?}", res2.unwrap());
let f1 = ipfs.get_dag(path.sub_path("0").unwrap());
let f2 = ipfs.get_dag(path.sub_path("1").unwrap());
let (res1, res2) = join!(f1, f2);
println!("Received block with contents: {:?}", res1.unwrap());
println!("Received block with contents: {:?}", res2.unwrap());
ipfs.exit_daemon().await;
});
ipfs.exit_daemon().await;
}

67
examples/fetch_and_cat.rs
View File

@ -1,15 +1,16 @@
#![recursion_limit = "512"]
use cid::Cid;
use futures::io::AsyncWriteExt;
use futures::pin_mut;
use futures::stream::StreamExt; // needed for StreamExt::next
use ipfs::{Ipfs, TestTypes, UninitializedIpfs};
use tokio::io::AsyncWriteExt;
use std::convert::TryFrom;
use std::env;
use std::process::exit;
fn main() {
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
// this example will wait forever attempting to fetch a CID provided at command line. It is
@ -36,45 +37,43 @@ fn main() {
}
};
async_std::task::block_on(async move {
// Start daemon and initialize repo
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
async_std::task::spawn(fut);
// Start daemon and initialize repo
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
tokio::task::spawn(fut);
let (public_key, addresses) = ipfs.identity().await.unwrap();
assert!(!addresses.is_empty(), "Zero listening addresses");
let (public_key, addresses) = ipfs.identity().await.unwrap();
assert!(!addresses.is_empty(), "Zero listening addresses");
eprintln!("Please connect an ipfs node having {} to:\n", cid);
eprintln!("Please connect an ipfs node having {} to:\n", cid);
let peer_id = public_key.into_peer_id().to_string();
let peer_id = public_key.into_peer_id().to_string();
for address in addresses {
eprintln!(" - {}/p2p/{}", address, peer_id);
}
for address in addresses {
eprintln!(" - {}/p2p/{}", address, peer_id);
}
eprintln!();
eprintln!();
let stream = ipfs.cat_unixfs(cid, None).await.unwrap_or_else(|e| {
eprintln!("Error: {}", e);
exit(1);
});
// The stream needs to be pinned on the stack to be used with StreamExt::next
pin_mut!(stream);
let mut stdout = async_std::io::stdout();
let stream = ipfs.cat_unixfs(cid, None).await.unwrap_or_else(|e| {
eprintln!("Error: {}", e);
exit(1);
});
// The stream needs to be pinned on the stack to be used with StreamExt::next
pin_mut!(stream);
let mut stdout = tokio::io::stdout();
loop {
// This could be made more performant by polling the stream while writing to stdout.
match stream.next().await {
Some(Ok(bytes)) => {
stdout.write_all(&bytes).await.unwrap();
}
Some(Err(e)) => {
eprintln!("Error: {}", e);
exit(1);
}
None => break,
loop {
// This could be made more performant by polling the stream while writing to stdout.
match stream.next().await {
Some(Ok(bytes)) => {
stdout.write_all(&bytes).await.unwrap();
}
Some(Err(e)) => {
eprintln!("Error: {}", e);
exit(1);
}
None => break,
}
})
}
}

79
examples/ipfs_bitswap_test.rs
View File

@ -1,65 +1,64 @@
#![recursion_limit = "512"]
use async_std::task;
use tokio::{io::{AsyncBufReadExt, BufReader, stdin}, task};
use cid::{Cid, Codec};
use ipfs::{Block, Ipfs, TestTypes, UninitializedIpfs};
use multihash::Sha2_256;
fn main() {
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
// this example demonstrates
// - block building
// - local swarm communication with go-ipfs
task::block_on(async move {
// Start daemon and initialize repo
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
// Start daemon and initialize repo
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
let data = b"block-want\n".to_vec().into_boxed_slice();
let wanted = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data));
let data = b"block-want\n".to_vec().into_boxed_slice();
let wanted = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data));
let (public_key, addresses) = ipfs.identity().await.unwrap();
assert!(!addresses.is_empty(), "Zero listening addresses");
let (public_key, addresses) = ipfs.identity().await.unwrap();
assert!(!addresses.is_empty(), "Zero listening addresses");
eprintln!("Please connect an ipfs node having {} to:\n", wanted);
eprintln!("Please connect an ipfs node having {} to:\n", wanted);
let peer_id = public_key.into_peer_id().to_string();
let peer_id = public_key.into_peer_id().to_string();
for address in addresses {
eprintln!(" - {}/p2p/{}", address, peer_id);
}
for address in addresses {
eprintln!(" - {}/p2p/{}", address, peer_id);
}
eprintln!();
eprintln!("The block wanted in this example can be created on the other node:");
eprintln!(" echo block-want | ipfs block put -f raw");
eprintln!();
eprintln!();
eprintln!("The block wanted in this example can be created on the other node:");
eprintln!(" echo block-want | ipfs block put -f raw");
eprintln!();
// Create a Block
let data = b"block-provide\n".to_vec().into_boxed_slice();
let cid = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data));
let provided = ipfs.put_block(Block::new(data, cid)).await.unwrap();
// Create a Block
let data = b"block-provide\n".to_vec().into_boxed_slice();
let cid = Cid::new_v1(Codec::Raw, Sha2_256::digest(&data));
let provided = ipfs.put_block(Block::new(data, cid)).await.unwrap();
eprintln!(
"After connecting the node, it can be used to get block: {}",
provided
);
eprintln!("This should print out \"block-provide\\n\":");
eprintln!(" ipfs block get {}", provided);
eprintln!();
eprintln!(
"After connecting the node, it can be used to get block: {}",
provided
);
eprintln!("This should print out \"block-provide\\n\":");
eprintln!(" ipfs block get {}", provided);
eprintln!();
// Retrive a Block
let block = ipfs.get_block(&wanted).await.unwrap();
let contents = std::str::from_utf8(block.data()).unwrap();
eprintln!("Block retrieved: {:?}", contents);
// Retrive a Block
let block = ipfs.get_block(&wanted).await.unwrap();
let contents = std::str::from_utf8(block.data()).unwrap();
eprintln!("Block retrieved: {:?}", contents);
eprintln!();
eprintln!("Press enter or CTRL-C to exit this example.");
eprintln!();
eprintln!("Press enter or CTRL-C to exit this example.");
let _ = async_std::io::stdin().read_line(&mut String::new()).await;
let _ = BufReader::new(stdin()).read_line(&mut String::new()).await;
ipfs.exit_daemon().await;
});
ipfs.exit_daemon().await;
}

53
examples/ipfs_ipns_test.rs
View File

@ -1,37 +1,36 @@
use async_std::task;
use tokio::task;
use ipfs::{Ipfs, IpfsPath, PeerId, TestTypes, UninitializedIpfs};
use std::str::FromStr;
fn main() {
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
task::block_on(async move {
// Start daemon and initialize repo
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
// Start daemon and initialize repo
let (ipfs, fut): (Ipfs<TestTypes>, _) =
UninitializedIpfs::default().await.start().await.unwrap();
task::spawn(fut);
// Create a Block
let cid = ipfs.put_dag("block v0".into()).await.unwrap();
let ipfs_path = IpfsPath::from(cid);
// Publish a Block
let ipns_path = ipfs
.publish_ipns(&PeerId::random(), &ipfs_path)
.await
.unwrap();
// Create a Block
let cid = ipfs.put_dag("block v0".into()).await.unwrap();
let ipfs_path = IpfsPath::from(cid);
// Publish a Block
let ipns_path = ipfs
.publish_ipns(&PeerId::random(), &ipfs_path)
.await
.unwrap();
// Resolve a Block
let new_ipfs_path = ipfs.resolve_ipns(&ipns_path).await.unwrap();
assert_eq!(ipfs_path, new_ipfs_path);
// Resolve a Block
let new_ipfs_path = ipfs.resolve_ipns(&ipns_path).await.unwrap();
assert_eq!(ipfs_path, new_ipfs_path);
// Resolve dnslink
let ipfs_path = IpfsPath::from_str("/ipns/ipfs.io").unwrap();
println!("Resolving {:?}", ipfs_path.to_string());
let ipfs_path = ipfs.resolve_ipns(&ipfs_path).await.unwrap();
println!("Resolved stage 1: {:?}", ipfs_path.to_string());
let ipfs_path = ipfs.resolve_ipns(&ipfs_path).await.unwrap();
println!("Resolved stage 2: {:?}", ipfs_path.to_string());
// Resolve dnslink
let ipfs_path = IpfsPath::from_str("/ipns/ipfs.io").unwrap();
println!("Resolving {:?}", ipfs_path.to_string());
let ipfs_path = ipfs.resolve_ipns(&ipfs_path).await.unwrap();
println!("Resolved stage 1: {:?}", ipfs_path.to_string());
let ipfs_path = ipfs.resolve_ipns(&ipfs_path).await.unwrap();
println!("Resolved stage 2: {:?}", ipfs_path.to_string());
ipfs.exit_daemon().await;
});
ipfs.exit_daemon().await;
}

10
src/dag.rs
View File

@ -95,7 +95,7 @@ mod tests {
use super::*;
use crate::{make_ipld, Node};
#[async_std::test]
#[tokio::test]
async fn test_resolve_root_cid() {
let Node { ipfs, bg_task: _bt } = Node::new("test_node").await;
let dag = IpldDag::new(ipfs);
@ -105,7 +105,7 @@ mod tests {
assert_eq!(res, data);
}
#[async_std::test]
#[tokio::test]
async fn test_resolve_array_elem() {
let Node { ipfs, bg_task: _bt } = Node::new("test_node").await;
let dag = IpldDag::new(ipfs);
@ -118,7 +118,7 @@ mod tests {
assert_eq!(res, make_ipld!(2));
}
#[async_std::test]
#[tokio::test]
async fn test_resolve_nested_array_elem() {
let Node { ipfs, bg_task: _bt } = Node::new("test_node").await;
let dag = IpldDag::new(ipfs);
@ -131,7 +131,7 @@ mod tests {
assert_eq!(res, make_ipld!(2));
}
#[async_std::test]
#[tokio::test]
async fn test_resolve_object_elem() {
let Node { ipfs, bg_task: _bt } = Node::new("test_node").await;
let dag = IpldDag::new(ipfs);
@ -146,7 +146,7 @@ mod tests {
assert_eq!(res, make_ipld!(false));
}
#[async_std::test]
#[tokio::test]
async fn test_resolve_cid_elem() {
let Node { ipfs, bg_task: _bt } = Node::new("test_node").await;
let dag = IpldDag::new(ipfs);

10
src/ipns/dns.rs
View File

@ -79,19 +79,15 @@ pub async fn resolve(domain: &str) -> Result<IpfsPath, Error> {
mod tests {
use super::*;
#[async_std::test]
#[ignore]
#[tokio::test]
async fn test_resolve1() {
let res = resolve("ipfs.io").await.unwrap().to_string();
assert_eq!(res, "/ipns/website.ipfs.io");
}
#[async_std::test]
#[ignore]
#[tokio::test]
async fn test_resolve2() {
let res = resolve("website.ipfs.io").await.unwrap().to_string();
// FIXME: perhaps this should just be a path to multihash? otherwise it'll
// break every time they update the site.
assert_eq!(res, "/ipfs/QmbV3st6TDZVocp4H2f4KE3tvLP1BEpeRHhZyFL9gD4Ut4");
assert_eq!(res, "/ipfs/bafybeiayvrj27f65vbecspbnuavehcb3znvnt2strop2rfbczupudoizya");
}
}

14
src/lib.rs
View File

@ -8,7 +8,7 @@ extern crate tracing;
pub use crate::ipld::Ipld;
use anyhow::{anyhow, format_err};
use async_std::path::PathBuf;
use std::path::PathBuf;
pub use bitswap::{BitswapEvent, Block, Stats};
pub use cid::Cid;
use cid::Codec;
@ -1047,7 +1047,7 @@ mod node {
/// easier.
pub struct Node {
pub ipfs: Ipfs<TestTypes>,
pub bg_task: async_std::task::JoinHandle<()>,
pub bg_task: tokio::task::JoinHandle<()>,
}
impl Node {
@ -1069,7 +1069,7 @@ mod node {
.await
.unwrap();
let bg_task = async_std::task::spawn(fut.in_current_span());
let bg_task = tokio::task::spawn(fut.in_current_span());
Node { ipfs, bg_task }
}
@ -1124,7 +1124,7 @@ mod node {
pub async fn shutdown(self) {
self.ipfs.exit_daemon().await;
self.bg_task.await;
let _ = self.bg_task.await;
}
}
@ -1260,7 +1260,7 @@ mod tests {
));
}
#[async_std::test]
#[tokio::test]
async fn test_put_and_get_block() {
let ipfs = Node::new("test_node").await;
@ -1273,7 +1273,7 @@ mod tests {
assert_eq!(block, new_block);
}
#[async_std::test]
#[tokio::test]
async fn test_put_and_get_dag() {
let ipfs = Node::new("test_node").await;
@ -1283,7 +1283,7 @@ mod tests {
assert_eq!(data, new_data);
}
#[async_std::test]
#[tokio::test]
async fn test_pin_and_unpin() {
let ipfs = Node::new("test_node").await;

2
src/p2p/behaviour.rs
View File

@ -5,7 +5,7 @@ use crate::repo::BlockPut;
use crate::subscription::{SubscriptionFuture, SubscriptionRegistry};
use crate::{Ipfs, IpfsTypes};
use anyhow::anyhow;
use async_std::task;
use tokio::task;
use bitswap::{Bitswap, BitswapEvent};
use cid::Cid;
use libp2p::core::{Multiaddr, PeerId};

2
src/p2p/mod.rs
View File

@ -74,6 +74,6 @@ impl libp2p::core::Executor for SpannedExecutor {
future: std::pin::Pin<Box<dyn std::future::Future<Output = ()> + 'static + Send>>,
) {
use tracing_futures::Instrument;
async_std::task::spawn(future.instrument(self.0.clone()));
tokio::task::spawn(future.instrument(self.0.clone()));
}
}

2
src/p2p/swarm.rs
View File

@ -339,7 +339,7 @@ mod tests {
));
}
#[async_std::test]
#[tokio::test]
async fn swarm_api() {
let (peer1_id, trans) = mk_transport();
let mut swarm1 = Swarm::new(trans, SwarmApi::default(), peer1_id);

4
src/p2p/transport.rs
View File

@ -5,7 +5,7 @@ use libp2p::core::upgrade::SelectUpgrade;
use libp2p::identity;
use libp2p::mplex::MplexConfig;
use libp2p::noise::{self, NoiseConfig};
use libp2p::tcp::TcpConfig;
use libp2p::tcp::TokioTcpConfig;
use libp2p::yamux::Config as YamuxConfig;
use libp2p::{PeerId, Transport};
use std::io::{Error, ErrorKind};
@ -23,7 +23,7 @@ pub fn build_transport(keypair: identity::Keypair) -> TTransport {
.unwrap();
let noise_config = NoiseConfig::xx(xx_keypair).into_authenticated();
TcpConfig::new()
TokioTcpConfig::new()
.nodelay(true)
.upgrade(Version::V1)
.authenticate(noise_config)

11
src/repo/fs.rs
View File

@ -1,9 +1,8 @@
//! Persistent fs backed repo
use crate::error::Error;
use crate::repo::{BlockPut, BlockStore};
use async_std::fs;
use async_std::path::PathBuf;
use async_std::prelude::*;
use tokio::{fs, io::{AsyncReadExt, AsyncWriteExt}};
use std::path::PathBuf;
use async_trait::async_trait;
use bitswap::Block;
use cid::Cid;
@ -145,7 +144,7 @@ mod tests {
use multihash::Sha2_256;
use std::env::temp_dir;
#[async_std::test]
#[tokio::test]
async fn test_fs_blockstore() {
let mut tmp = temp_dir();
tmp.push("blockstore1");
@ -183,7 +182,7 @@ mod tests {
std::fs::remove_dir_all(tmp).ok();
}
#[async_std::test]
#[tokio::test]
async fn test_fs_blockstore_open() {
let mut tmp = temp_dir();
tmp.push("blockstore2");
@ -208,7 +207,7 @@ mod tests {
std::fs::remove_dir_all(&tmp).ok();
}
#[async_std::test]
#[tokio::test]
async fn test_fs_blockstore_list() {
let mut tmp = temp_dir();
tmp.push("blockstore_list");

8
src/repo/mem.rs
View File

@ -1,7 +1,7 @@
//! Volatile memory backed repo
use crate::error::Error;
use crate::repo::{BlockPut, BlockStore, Column, DataStore};
use async_std::path::PathBuf;
use std::path::PathBuf;
use async_trait::async_trait;
use bitswap::Block;
use cid::Cid;
@ -154,7 +154,7 @@ mod tests {
use multihash::Sha2_256;
use std::env::temp_dir;
#[async_std::test]
#[tokio::test]
async fn test_mem_blockstore() {
let tmp = temp_dir();
let store = MemBlockStore::new(tmp.into());
@ -187,7 +187,7 @@ mod tests {
assert_eq!(get.await.unwrap(), None);
}
#[async_std::test]
#[tokio::test]
async fn test_mem_blockstore_list() {
let tmp = temp_dir();
let mem_store = MemBlockStore::new(tmp.into());
@ -210,7 +210,7 @@ mod tests {
}
}
#[async_std::test]
#[tokio::test]
async fn test_mem_datastore() {
let tmp = temp_dir();
let store = MemDataStore::new(tmp.into());

4
src/repo/mod.rs
View File

@ -3,7 +3,7 @@ use crate::error::Error;
use crate::path::IpfsPath;
use crate::subscription::{RequestKind, SubscriptionFuture, SubscriptionRegistry};
use crate::IpfsOptions;
use async_std::path::PathBuf;
use std::path::PathBuf;
use async_trait::async_trait;
use bitswap::Block;
use cid::{self, Cid};
@ -370,7 +370,7 @@ pub(crate) mod tests {
Repo::new(options)
}
#[async_std::test]
#[tokio::test]
async fn test_repo() {
let (repo, _) = create_mock_repo();
repo.init().await.unwrap();

16
src/subscription.rs
View File

@ -4,8 +4,8 @@
//! sharing the same unique numeric identifier, the `SubscriptionId`.
use crate::{p2p::ConnectionTarget, RepoEvent};
use async_std::future::Future;
use async_std::task::{Context, Poll, Waker};
use futures::future::Future;
use std::task::{Context, Poll, Waker};
use cid::Cid;
use core::fmt::Debug;
use core::hash::Hash;
@ -443,7 +443,7 @@ mod tests {
}
}
#[async_std::test]
#[tokio::test]
async fn subscription_basics() {
let registry = SubscriptionRegistry::<u32, ()>::default();
let s1 = registry.create_subscription(0.into(), None);
@ -455,7 +455,7 @@ mod tests {
assert_eq!(s3.await.unwrap(), 10);
}
#[async_std::test]
#[tokio::test]
async fn subscription_cancelled_on_dropping_registry() {
let registry = SubscriptionRegistry::<u32, ()>::default();
let s1 = registry.create_subscription(0.into(), None);
@ -463,7 +463,7 @@ mod tests {
assert_eq!(s1.await, Err(SubscriptionErr::Cancelled));
}
#[async_std::test]
#[tokio::test]
async fn subscription_cancelled_on_shutdown() {
let registry = SubscriptionRegistry::<u32, ()>::default();
let s1 = registry.create_subscription(0.into(), None);
@ -471,7 +471,7 @@ mod tests {
assert_eq!(s1.await, Err(SubscriptionErr::Cancelled));
}
#[async_std::test]
#[tokio::test]
async fn new_subscriptions_cancelled_after_shutdown() {
let registry = SubscriptionRegistry::<u32, ()>::default();
registry.shutdown();
@ -479,9 +479,9 @@ mod tests {
assert_eq!(s1.await, Err(SubscriptionErr::Cancelled));
}
#[async_std::test]
#[tokio::test]
async fn dropping_subscription_future_after_registering() {
use async_std::future::timeout;
use tokio::time::timeout;
use std::time::Duration;
let registry = SubscriptionRegistry::<u32, ()>::default();

6
tests/bitswap_cleanup.rs
View File

@ -1,12 +1,12 @@
use async_std::task;
use tokio::time;
use ipfs::Node;
async fn wait(millis: u64) {
task::spawn(task::sleep(std::time::Duration::from_millis(millis))).await;
time::delay_for(std::time::Duration::from_millis(millis)).await;
}
// Ensure that the Bitswap object doesn't leak.
#[async_std::test]
#[tokio::test]
async fn check_bitswap_cleanups() {
// create a few nodes
let node_a = Node::new("a").await;

2
tests/bitswap_stress_test.rs
View File

@ -12,7 +12,7 @@ fn filter(i: usize) -> bool {
// testing the bitswap protocol (though it would be advised to uncomment
// the tracing_subscriber for stress-testing purposes)
#[ignore]
#[async_std::test]
#[tokio::test]
async fn bitswap_stress_test() {
tracing_subscriber::fmt::init();

14
tests/connect_two.rs
View File

@ -1,9 +1,9 @@
use async_std::future::timeout;
use tokio::time::timeout;
use ipfs::{ConnectionTarget, Node};
use std::time::Duration;
// Make sure two instances of ipfs can be connected by `Multiaddr`.
#[async_std::test]
#[tokio::test]
async fn connect_two_nodes_by_addr() {
let node_a = Node::new("a").await;
let node_b = Node::new("b").await;
@ -22,7 +22,7 @@ async fn connect_two_nodes_by_addr() {
// order to connect by PeerId) already performs a dial to the
// given peer within Pubsub::add_node_to_partial_view it calls
#[ignore]
#[async_std::test]
#[tokio::test]
async fn connect_two_nodes_by_peer_id() {
let node_a = Node::new("a").await;
let node_b = Node::new("b").await;
@ -40,7 +40,7 @@ async fn connect_two_nodes_by_peer_id() {
}
// Make sure two instances of ipfs can be connected with a multiaddr+peer combo.
#[async_std::test]
#[tokio::test]
async fn connect_two_nodes_by_addr_and_peer() {
let node_a = Node::new("a").await;
let node_b = Node::new("b").await;
@ -59,7 +59,7 @@ async fn connect_two_nodes_by_addr_and_peer() {
}
// Ensure that duplicate connection attempts don't cause hangs.
#[async_std::test]
#[tokio::test]
async fn connect_duplicate_multiaddr() {
let node_a = Node::new("a").await;
let node_b = Node::new("b").await;
@ -77,7 +77,7 @@ async fn connect_duplicate_multiaddr() {
}
// Ensure that duplicate connection attempts don't cause hangs.
#[async_std::test]
#[tokio::test]
async fn connect_duplicate_peer_id() {
let node_a = Node::new("a").await;
let node_b = Node::new("b").await;
@ -98,7 +98,7 @@ async fn connect_duplicate_peer_id() {
// More complicated one to the above; first node will have two listening addresses and the second
// one should dial both of the addresses, resulting in two connections.
#[async_std::test]
#[tokio::test]
async fn connect_two_nodes_with_two_connections_doesnt_panic() {
let node_a = Node::new("a").await;
let node_b = Node::new("b").await;

4
tests/exchange_block.rs
View File

@ -1,10 +1,10 @@
use async_std::future::timeout;
use tokio::time::timeout;
use cid::{Cid, Codec};
use ipfs::{Block, Node};
use multihash::Sha2_256;
use std::time::Duration;
#[async_std::test]
#[tokio::test]
async fn exchange_block() {
tracing_subscriber::fmt::init();

6
tests/kademlia.rs
View File

@ -1,10 +1,10 @@
use async_std::future::timeout;
use tokio::time::timeout;
use cid::Cid;
use ipfs::{IpfsOptions, Node};
use libp2p::{Multiaddr, PeerId};
use std::time::Duration;
#[async_std::test]
#[tokio::test]
async fn kademlia_local_peer_discovery() {
const BOOTSTRAPPER_COUNT: usize = 20;
@ -54,7 +54,7 @@ async fn kademlia_local_peer_discovery() {
}
#[ignore = "targets an actual bootstrapper, so random failures can happen"]
#[async_std::test]
#[tokio::test]
async fn kademlia_popular_content_discovery() {
let (bootstrapper_id, bootstrapper_addr): (PeerId, Multiaddr) = (
"QmaCpDMGvV2BGHeYERUEnRQAwe3N8SzbUtfsmvsqQLuvuJ"

12
tests/multiple_listening_addresses.rs
View File

@ -1,4 +1,4 @@
#[async_std::test]
#[tokio::test]
async fn multiple_consecutive_ephemeral_listening_addresses() {
let node = ipfs::Node::new("test_node").await;
@ -12,7 +12,7 @@ async fn multiple_consecutive_ephemeral_listening_addresses() {
assert_ne!(first, second);
}
#[async_std::test]
#[tokio::test]
async fn multiple_concurrent_ephemeral_listening_addresses_on_same_ip() {
let node = ipfs::Node::new("test_node").await;
@ -41,7 +41,7 @@ async fn multiple_concurrent_ephemeral_listening_addresses_on_same_ip() {
);
}
#[async_std::test]
#[tokio::test]
#[cfg(not(target_os = "macos"))]
async fn multiple_concurrent_ephemeral_listening_addresses_on_different_ip() {
let node = ipfs::Node::new("test_node").await;
@ -59,7 +59,7 @@ async fn multiple_concurrent_ephemeral_listening_addresses_on_different_ip() {
second.unwrap();
}
#[async_std::test]
#[tokio::test]
async fn adding_unspecified_addr_resolves_with_first() {
let node = ipfs::Node::new("test_node").await;
// there is no test in trying to match this with others as ... that would be quite
@ -69,7 +69,7 @@ async fn adding_unspecified_addr_resolves_with_first() {
.unwrap();
}
#[async_std::test]
#[tokio::test]
async fn listening_for_multiple_unspecified_addresses() {
let node = ipfs::Node::new("test_node").await;
// there is no test in trying to match this with others as ... that would be quite
@ -93,7 +93,7 @@ async fn listening_for_multiple_unspecified_addresses() {
);
}
#[async_std::test]
#[tokio::test]
async fn remove_listening_address() {
let node = ipfs::Node::new("test_node").await;

13
tests/pubsub.rs
View File

@ -1,16 +1,17 @@
use async_std::future::{pending, timeout};
use futures::future::pending;
use futures::stream::StreamExt;
use ipfs::{Node, PeerId};
use tokio::time::timeout;
use std::time::Duration;
#[async_std::test]
#[tokio::test]
async fn subscribe_only_once() {
let a = Node::new("test_node").await;
let _stream = a.pubsub_subscribe("some_topic".into()).await.unwrap();
a.pubsub_subscribe("some_topic".into()).await.unwrap_err();
}
#[async_std::test]
#[tokio::test]
async fn resubscribe_after_unsubscribe() {
let a = Node::new("test_node").await;
@ -22,7 +23,7 @@ async fn resubscribe_after_unsubscribe() {
drop(a.pubsub_subscribe("topic".into()).await.unwrap());
}
#[async_std::test]
#[tokio::test]
async fn unsubscribe_via_drop() {
let a = Node::new("test_node").await;
@ -35,7 +36,7 @@ async fn unsubscribe_via_drop() {
assert_eq!(a.pubsub_subscribed().await.unwrap(), empty);
}
#[async_std::test]
#[tokio::test]
async fn can_publish_without_subscribing() {
let a = Node::new("test_node").await;
a.pubsub_publish("topic".into(), b"foobar".to_vec())
@ -43,7 +44,7 @@ async fn can_publish_without_subscribing() {
.unwrap()
}
#[async_std::test]
#[tokio::test]
#[allow(clippy::mutable_key_type)] // clippy doesn't like Vec inside HashSet
async fn publish_between_two_nodes() {
use futures::stream::StreamExt;

9
tests/wantlist_and_cancellation.rs
View File

@ -1,9 +1,6 @@
use async_std::{
future::{pending, timeout},
task,
};
use tokio::{task, time::timeout};
use cid::Cid;
use futures::future::{select, Either, FutureExt};
use futures::future::{pending, select, Either, FutureExt};
use futures::future::{AbortHandle, Abortable};
use ipfs::Node;
@ -50,7 +47,7 @@ async fn check_cid_subscriptions(ipfs: &Node, cid: &Cid, expected_count: usize)
}
/// Check if canceling a Cid affects the wantlist.
#[async_std::test]
#[tokio::test]
async fn wantlist_cancellation() {
tracing_subscriber::fmt::init();