mirror of
https://github.com/fluencelabs/rust-libp2p
synced 2025-06-23 23:01:33 +00:00
feat(examples): use cbor
codec in file-sharing
example
Remove the use of the core `upgrade::transfer` module in `file-sharing` example in favor of `cbor` codec. Related #4011. Pull-Request: #4036.
This commit is contained in:
2
Cargo.lock
generated
2
Cargo.lock
generated
@ -1534,13 +1534,13 @@ name = "file-sharing"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"async-std",
|
||||
"async-trait",
|
||||
"clap",
|
||||
"either",
|
||||
"env_logger 0.10.0",
|
||||
"futures",
|
||||
"libp2p",
|
||||
"multiaddr",
|
||||
"serde",
|
||||
"void",
|
||||
]
|
||||
|
||||
|
@ -6,12 +6,12 @@ publish = false
|
||||
license = "MIT"
|
||||
|
||||
[dependencies]
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
async-std = { version = "1.12", features = ["attributes"] }
|
||||
async-trait = "0.1"
|
||||
clap = { version = "4.3.1", features = ["derive"] }
|
||||
either = "1.8"
|
||||
env_logger = "0.10"
|
||||
futures = "0.3.28"
|
||||
libp2p = { path = "../../libp2p", features = ["async-std", "dns", "kad", "noise", "macros", "request-response", "tcp", "websocket", "yamux"] }
|
||||
libp2p = { path = "../../libp2p", features = ["async-std", "cbor", "dns", "kad", "noise", "macros", "request-response", "tcp", "websocket", "yamux"] }
|
||||
multiaddr = { version = "0.17.1" }
|
||||
void = "1.0.2"
|
||||
|
@ -1,14 +1,10 @@
|
||||
use async_std::io;
|
||||
use async_trait::async_trait;
|
||||
use either::Either;
|
||||
use futures::channel::{mpsc, oneshot};
|
||||
use futures::prelude::*;
|
||||
|
||||
use libp2p::{
|
||||
core::{
|
||||
upgrade::{read_length_prefixed, write_length_prefixed},
|
||||
Multiaddr,
|
||||
},
|
||||
core::Multiaddr,
|
||||
identity,
|
||||
kad::{
|
||||
record::store::MemoryStore, GetProvidersOk, Kademlia, KademliaEvent, QueryId, QueryResult,
|
||||
@ -22,9 +18,9 @@ use libp2p::{
|
||||
|
||||
use libp2p::core::upgrade::Version;
|
||||
use libp2p::StreamProtocol;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::{hash_map, HashMap, HashSet};
|
||||
use std::error::Error;
|
||||
use std::iter;
|
||||
|
||||
/// Creates the network components, namely:
|
||||
///
|
||||
@ -60,13 +56,12 @@ pub(crate) async fn new(
|
||||
transport,
|
||||
ComposedBehaviour {
|
||||
kademlia: Kademlia::new(peer_id, MemoryStore::new(peer_id)),
|
||||
request_response: request_response::Behaviour::with_codec(
|
||||
FileExchangeCodec(),
|
||||
iter::once((
|
||||
request_response: request_response::cbor::Behaviour::new(
|
||||
[(
|
||||
StreamProtocol::new("/file-exchange/1"),
|
||||
ProtocolSupport::Full,
|
||||
)),
|
||||
Default::default(),
|
||||
)],
|
||||
request_response::Config::default(),
|
||||
),
|
||||
},
|
||||
peer_id,
|
||||
@ -413,7 +408,7 @@ impl EventLoop {
|
||||
#[derive(NetworkBehaviour)]
|
||||
#[behaviour(to_swarm = "ComposedEvent")]
|
||||
struct ComposedBehaviour {
|
||||
request_response: request_response::Behaviour<FileExchangeCodec>,
|
||||
request_response: request_response::cbor::Behaviour<FileRequest, FileResponse>,
|
||||
kademlia: Kademlia<MemoryStore>,
|
||||
}
|
||||
|
||||
@ -474,77 +469,7 @@ pub(crate) enum Event {
|
||||
}
|
||||
|
||||
// Simple file exchange protocol
|
||||
|
||||
#[derive(Clone)]
|
||||
struct FileExchangeCodec();
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
struct FileRequest(String);
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub(crate) struct FileResponse(Vec<u8>);
|
||||
|
||||
#[async_trait]
|
||||
impl request_response::Codec for FileExchangeCodec {
|
||||
type Protocol = StreamProtocol;
|
||||
type Request = FileRequest;
|
||||
type Response = FileResponse;
|
||||
|
||||
async fn read_request<T>(&mut self, _: &StreamProtocol, io: &mut T) -> io::Result<Self::Request>
|
||||
where
|
||||
T: AsyncRead + Unpin + Send,
|
||||
{
|
||||
let vec = read_length_prefixed(io, 1_000_000).await?;
|
||||
|
||||
if vec.is_empty() {
|
||||
return Err(io::ErrorKind::UnexpectedEof.into());
|
||||
}
|
||||
|
||||
Ok(FileRequest(String::from_utf8(vec).unwrap()))
|
||||
}
|
||||
|
||||
async fn read_response<T>(
|
||||
&mut self,
|
||||
_: &StreamProtocol,
|
||||
io: &mut T,
|
||||
) -> io::Result<Self::Response>
|
||||
where
|
||||
T: AsyncRead + Unpin + Send,
|
||||
{
|
||||
let vec = read_length_prefixed(io, 500_000_000).await?; // update transfer maximum
|
||||
|
||||
if vec.is_empty() {
|
||||
return Err(io::ErrorKind::UnexpectedEof.into());
|
||||
}
|
||||
|
||||
Ok(FileResponse(vec))
|
||||
}
|
||||
|
||||
async fn write_request<T>(
|
||||
&mut self,
|
||||
_: &StreamProtocol,
|
||||
io: &mut T,
|
||||
FileRequest(data): FileRequest,
|
||||
) -> io::Result<()>
|
||||
where
|
||||
T: AsyncWrite + Unpin + Send,
|
||||
{
|
||||
write_length_prefixed(io, data).await?;
|
||||
io.close().await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn write_response<T>(
|
||||
&mut self,
|
||||
_: &StreamProtocol,
|
||||
io: &mut T,
|
||||
FileResponse(data): FileResponse,
|
||||
) -> io::Result<()>
|
||||
where
|
||||
T: AsyncWrite + Unpin + Send,
|
||||
{
|
||||
write_length_prefixed(io, data).await?;
|
||||
io.close().await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
@ -14,6 +14,7 @@ categories = ["network-programming", "asynchronous"]
|
||||
full = [
|
||||
"async-std",
|
||||
"autonat",
|
||||
"cbor",
|
||||
"dcutr",
|
||||
"deflate",
|
||||
"dns",
|
||||
@ -49,6 +50,7 @@ full = [
|
||||
|
||||
async-std = ["libp2p-swarm/async-std", "libp2p-mdns?/async-io", "libp2p-tcp?/async-io", "libp2p-dns?/async-std"]
|
||||
autonat = ["dep:libp2p-autonat"]
|
||||
cbor = ["libp2p-request-response?/cbor"]
|
||||
dcutr = ["dep:libp2p-dcutr", "libp2p-metrics?/dcutr"]
|
||||
deflate = ["dep:libp2p-deflate"]
|
||||
dns = ["dep:libp2p-dns"]
|
||||
|
Reference in New Issue
Block a user