尧图网站建设 
rust-libp2p 通用流式协议libp2p-stream完全指南基于 Control 的入站/出站流管理【免费下载链接】rust-libp2pThe Rust Implementation of the libp2p networking stack.项目地址: https://gitcode.com/GitHub_Trending/ru/rust-libp2plibp2p 网络栈的基石是流stream——所有上层协议Kademlia、Gossipsub、Identify 等最终都以流为载体交换数据。rust-libp2p 仓库中的protocols/streamcrate 名为libp2p-stream把这个通用能力抽象成一个独立的NetworkBehaviour应用可以绕过传统的事件驱动模型通过一个可克隆、可跨任务共享的Control句柄直接注册入站协议并异步打开出站流。读完本文你将掌握如何用libp2p-stream在 Swarm 之上搭建自己的自定义协议、正确处理入站流的背压与资源管理并理解其以共享状态为中心的 Handler 分发这一独特设计。为什么需要独立的通用流协议模块在 rust-libp2p 中流的抽象由 libp2p-swarm 提供swarm::Stream它是连接复用muxing之上的双向字节管道。绝大多数协议如 ping、identify通过实现NetworkBehaviour与ConnectionHandler的样板代码来使用流。libp2p-stream的定位恰恰是把这些样板收起来让应用直接以函数调用的方式操作流。从 Cargo.toml 可以看到它的依赖面非常精简futures、libp2p-core、libp2p-identity、libp2p-swarm、tracing与rand这说明它是一个轻量、专注的通用组件。其核心设计可以概括为三点见 README流是 libp2p 的基本原语其他所有协议都由流实现与传统的NetworkBehaviour不同本模块采用不同的设计思路所有交互都通过Control完成Control可以克隆因此可以在整个应用中被安全共享。从源码结构看模块由 5 个部分组成见 lib.rs 的模块声明behaviour.rsBehaviour 实现、control.rs用户操作入口、handler.rs连接级 Handler、shared.rs跨 Handler 的共享状态与upgrade.rs协议协商升级。其中shared是整个设计的枢纽我们稍后详解。快速上手把 Behaviour 装进 Swarmstream::Behaviour实现了NetworkBehaviour并且实现了Default因此它可以像任何 Behaviour 一样直接组装进 Swarm。仓库中的官方示例 examples/stream/src/main.rs 展示了在 tokio QUIC 传输上的标准装配方式let mut swarm libp2p::SwarmBuilder::with_new_identity() .with_tokio() .with_quic() .with_behaviour(|_| stream::Behaviour::new())? .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(10))) .build(); swarm.listen_on(/ip4/127.0.0.1/udp/0/quic-v1.parse()?)?;装配完成后通过swarm.behaviour().new_control()获取第一个Control实例behaviour.rspub fn new_control(self) - Control { Control::new(self.shared.clone()) }Control内部只是对ArcMutexShared的一个封装克隆它只是克隆Arc成本极低因此你可以无负担地把Control传给任意数量的 task。Inbound注册入站协议并消费入站流基本用法要接受某个StreamProtocol的入站流调用Control::acceptuse libp2p_swarm::{Swarm, StreamProtocol}; use libp2p_stream as stream; use futures::StreamExt as _; let mut swarm: Swarmstream::Behaviour todo!(); let mut control swarm.behaviour().new_control(); let mut incoming control.accept(StreamProtocol::new(/my-protocol)).unwrap(); let handler_future async move { while let Some((peer, stream)) incoming.next().await { // 使用 stream 执行你的协议逻辑。 } };accept的语义要点协议命名StreamProtocol::new(/my-protocol)使用 libp2p 惯例的/协议名形式它与远端节点协商时使用的多流协议 ID 一一对应返回IncomingStreams这是一个实现了futures::Stream的类型其Item为(PeerId, Stream)——PeerId告诉你是谁发起了这条流control.rs重复注册会报错如果同一协议已被注册且尚未注销accept返回AlreadyRegistered这与HashMap的键已存在语义一致。在示例 examples/stream/src/main.rs 中入站侧被放入一个独立的tokio::spawn任务逐个处理 echo 流tokio::spawn(async move { while let Some((peer, stream)) incoming_streams.next().await { match echo(stream).await { /* ... */ } } });资源管理惰性流与背压Control::accept返回的IncomingStreams与其他futures::Stream一样是**惰性lazy**的只有被持续poll才会产生进展所以你必须始终驱动它——示例中通过StreamExt::next完成的正是这件事。IncomingStreams背后的通道是无界还是有界的看 shared.rs 中的accept实现let (sender, receiver) mpsc::channel(0); self.supported_inbound_protocols.insert(protocol.clone(), sender);它创建了一个容量为 0 的futures::channel::mpsc通道。这意味着通道本身不缓存流Shared::on_inbound_stream在try_send失败通道满时直接丢弃入站流并记录tracing::debug日志shared.rs。这正是 README 中所述机制的实现来源如果应用在处理入站流上落后即调用.next()的循环不够快内部会丢弃这些流。这是一个面向 DoS 防护的背压策略宁可丢流也不让接收队列无限膨胀耗尽内存。示例代码 examples/stream/src/main.rs 的注释也明确提示了这一设计意图并告诫如果想提高并发而每条流 spawn 一个 task实际上等于无界缓冲——每个 task 都需要内存激进的远端可能借此让你 OOM。注销与资源回收IncomingStreams被 drop 时其持有的mpsc::Receiver关闭通道另一端的Sender::is_closed()变为true。Shared::accept与supported_inbound_protocols在每次调用时会先retain清理已关闭的 sendershared.rs从而完成协议注销与内存回收。README 明确指出注销后的行为一旦 drop 掉IncomingStreams该协议即被注销。此后远端再尝试用该协议打开流将得到一个协商错误negotiation error。这一行为有对应的集成测试 tests/lib.rsdropping_incoming_streams_deregisters验证测试先让 swarm2 注册/test协议并成功完成一次流通信然后 abort 掉消费任务等价于 dropIncomingStreams再发起open_stream断言返回OpenStreamError::UnsupportedProtocol(_)。Outbound主动打开出站流基本用法打开出站流调用Control::open_streamuse libp2p_swarm::{Swarm, StreamProtocol}; use libp2p_stream as stream; use libp2p_identity::PeerId; let mut swarm: Swarmstream::Behaviour todo!(); let peer_id: PeerId todo!(); let mut control swarm.behaviour().new_control(); let protocol_future async move { let stream control.open_stream(peer_id, StreamProtocol::new(/my-protocol)).await.unwrap(); // 使用 stream 执行你的协议逻辑。 };值得注意的是open_stream是async fn且接收mut self——同一个Control同一时刻只能打开一条流。这与下面要讲的背压机制直接相关。自动拨号未连接也能开流open_stream的一个关键行为是当目标 peer 尚未连接时会自动发起拨号见 control.rs 的文档说明。其实现链路是Shared::sender(peer)在connections表中查找该 peer 的现有连接若存在则复用对应的mpsc::SenderNewStream若不存在则在pending_channels中登记一个通道对并向dial_sender投递一条拨号请求shared.rsBehaviour::poll从dial_receiver中取出 peer发出ToSwarm::Dial并附带PeerCondition::DisconnectedAndNotDialing避免重复拨号behaviour.rs连接建立后Handler收到来自Shared的 pending 通道open_stream的等待被唤醒随即发起子流协商。如果拨号失败Shared::on_dial_failure会把通道中所有 pending 的open_stream调用以io::ErrorKind::NotConnected错误唤醒shared.rs。对应的测试 tests/lib.rsdial_errors_are_propagated向一个随机PeerId开流断言返回io::ErrorKind::NotConnected且错误信息为Dial error: no addresses for peer.。错误类型open_stream返回ResultStream, OpenStreamError错误枚举定义在 control.rs变体含义触发场景UnsupportedProtocol(StreamProtocol)远端不支持该协议协商失败NegotiationFailed或对端已注销该协议Io(std::io::Error)握手期间的 I/O 错误拨号失败、连接重置、超时、通道断裂等OpenStreamError实现了std::error::ErrorIo变体提供source()透传底层错误且标记为#[non_exhaustive]便于未来扩展。示例 examples/stream/src/main.rs 展示了生产环境中的错误处理范式let stream match control.open_stream(peer, ECHO_PROTOCOL).await { Ok(stream) stream, Err(error stream::OpenStreamError::UnsupportedProtocol(_)) { tracing::info!(%peer, %error); return; // 协议不支持是永久性错误直接退出 } Err(error) { // 其他错误可能是暂时性的循环重试 tracing::debug!(%peer, %error); continue; } };背压为什么open_stream要求mut selfREADME 和源码注释都强调了这个设计要点Control支持类似有界通道的背压——每个Control内部有一个保证可用的消息槽位一个Control同一时刻只打开一条流这一点由mut self的签名强制执行control.rs。通道容量为 0意味着只有当Handler真正把NewStream消息取走、并最终把协商好的Stream通过oneshot送回时open_stream的.await才完成天然形成逐条串行、前后衔接的流控。但注释同时警告如果过度克隆Control这个背压机制就会被破坏——每个克隆都获得一个独立槽位克隆越多同时 in-flight 的流越多。所以合理做法是为每个需要并发开流的 task 单独克隆一个Control而不是共享同一个。深入原理Shared 状态与 Handler 分发libp2p-stream的设计精髓在于Shared——一个被ArcMutex包裹的共享状态机Behaviour与每个连接的Handler都持有它的引用shared.rs。它维护了 5 张表字段作用supported_inbound_protocols已注册入站协议 →IncomingStreams通道 sender 的映射connectionsConnectionId→PeerId的映射sendersConnectionId→mpsc::SenderNewStream的映射pending_channels拨号中 peer → 待定通道对的映射dial_sender待拨号 peer 的投递通道避免在poll中加锁之所以要用通道而非直接在poll里加锁取数据是为了避免在NetworkBehaviour::poll内持有锁——dial_sender这个设计behaviour.rs正是为规避锁竞争问题。Handler的职责非常纯粹handler.rs入站侧listen_protocol每次从Shared读取当前注册的协议列表构造Upgrade作为SubstreamProtocol入站流协商完成后FullyNegotiatedInbound事件把(stream, protocol)交回Shared::on_inbound_stream投递到对应的IncomingStreams出站侧poll从自己的receiver取到NewStream请求后发出OutboundSubstreamRequestFullyNegotiatedOutbound或DialUpgradeError则通过oneshot::Sender把结果送回open_stream的等待者。升级错误会被映射为OpenStreamError超时 →TimedOut、协商失败 →UnsupportedProtocol、I/O →Io见 handler.rs。Upgrade本身极简upgrade.rs它实现UpgradeInfo/InboundUpgrade/OutboundUpgradeupgrade_inbound/upgrade_outbound直接把(socket, info)原样返回Error Infallible因为流的协商多流协议选择已经由StreamProtocol列表完成无需额外握手数据。整条调用链可总结为open_stream(peer, proto) └─ Shared::sender(peer) ──► 已有连接? 复用通道 / 无连接? 发起自动拨号 └─ Handler::poll ──► OutboundSubstreamRequest(Upgrade{proto}) └─ 协商完成 ──► oneshot 送回 Stream或映射后的 OpenStreamError accept(proto) └─ Shared::accept ──► 注册 mpsc 通道返回 IncomingStreams └─ Handler 入站协商完成 ──► Shared::on_inbound_stream ──► 投递 (PeerId, Stream)一个完整的 echo 协议示例结合上面所有要点这里给出一个可直接运行的最小 echo 服务器/客户端思路完整可运行代码见 examples/stream/src/main.rs。同一份二进制传入目标地址即为拨号方服务端监听方let mut incoming swarm .behaviour() .new_control() .accept(ECHO_PROTOCOL) .unwrap(); tokio::spawn(async move { while let Some((peer, stream)) incoming.next().await { // echo读一块写一块直到 EOF match echo(stream).await { Ok(n) tracing::info!(%peer, Echoed {n} bytes!), Err(e) tracing::warn!(%peer, Echo failed: {e}), } } });客户端拨号方async fn connection_handler(peer: PeerId, mut control: stream::Control) { loop { tokio::time::sleep(Duration::from_secs(1)).await; let stream match control.open_stream(peer, ECHO_PROTOCOL).await { Ok(stream) stream, Err(error stream::OpenStreamError::UnsupportedProtocol(_)) { tracing::info!(%peer, %error); return; } Err(error) { tracing::debug!(%peer, %error); continue; } }; if let Err(e) send(stream).await { tracing::warn!(%peer, Echo protocol failed: {e}); continue; } } }echo函数在 examples/stream/src/main.rs循环read读到 0 字节对端关闭即结束否则write_all回写。send则写入随机字节并read_exact校验回显最后显式stream.close()。测试与验证行为即规范libp2p-stream的集成测试 tests/lib.rs 用libp2p-swarm-testSwarm::new_ephemeral_tokio在两个临时 Swarm 间建立内存传输连接直接验证了本文介绍的三大行为契约dropping_incoming_streams_deregisterstests/lib.rs入站处理任务被 abortIncomingStreams被 drop后再开流返回UnsupportedProtocol——证明drop 即注销dial_errors_are_propagatedtests/lib.rs对无地址的随机 peer 开流返回io::ErrorKind::NotConnected——证明自动拨号失败的错误传播。此外从 CHANGELOG.md 可以看到该模块的演进脉络0.4.0-alpha 起在accept新流时垃圾回收已注销的流即上文retain清理逻辑0.1.0-alpha 初始发布时就已为OpenStreamError实现Errortrait。当前版本为0.5.0-alphaMSRV 1.88.0仍处于 alpha 阶段API 可能随版本演进调整请以当前仓库源码为准。小结libp2p-stream提供了一种与 libp2p 主流事件驱动 Behaviour截然不同的编程模型以可克隆的Control为唯一入口用异步函数直接读写流。其要点可归纳为入站control.accept(StreamProtocol)注册协议并返回IncomingStreams它必须被持续 poll处理不过来时流会被内部丢弃面向 DoS 的背压drop 句柄即注销协议对端将收到协商错误出站control.open_stream(peer, proto)自动处理拨号与协商mut self保证每个Control单飞背压克隆需节制错误分为协议不支持永久与I/O 错误可能暂时两类原理ArcMutexShared集中管理协议注册、连接映射与待拨号队列Handler只做转发与协商Upgrade零成本透传流验证集成测试与仓库示例共同构成了可复现的行为规范。如果你的协议需求只是在 peer 之间传字节流与其手写全套 Behaviour Handler 样板不如直接在 Swarm 中挂载stream::Behaviour用几十行代码把自定义协议跑起来。【免费下载链接】rust-libp2pThe Rust Implementation of the libp2p networking stack.项目地址: https://gitcode.com/GitHub_Trending/ru/rust-libp2p创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 rust-libp2p 通用流式协议（libp2p-stream）完全指南：基于 Control 的入站/出站流管理
rust-libp2p 通用流式协议libp2p-stream完全指南基于 Control 的入站/出站流管理【免费下载链接】rust-libp2pThe Rust Implementation of the libp2p networking stack.项目地址: https://gitcode.com/GitHub_Trending/ru/rust-libp2plibp2p 网络栈的基石是流stream——所有上层协议Kademlia、Gossipsub、Identify 等最终都以流为载体交换数据。rust-libp2p 仓库中的protocols/streamcrate 名为libp2p-stream把这个通用能力抽象成一个独立的NetworkBehaviour应用可以绕过传统的事件驱动模型通过一个可克隆、可跨任务共享的Control句柄直接注册入站协议并异步打开出站流。读完本文你将掌握如何用libp2p-stream在 Swarm 之上搭建自己的自定义协议、正确处理入站流的背压与资源管理并理解其以共享状态为中心的 Handler 分发这一独特设计。为什么需要独立的通用流协议模块在 rust-libp2p 中流的抽象由 libp2p-swarm 提供swarm::Stream它是连接复用muxing之上的双向字节管道。绝大多数协议如 ping、identify通过实现NetworkBehaviour与ConnectionHandler的样板代码来使用流。libp2p-stream的定位恰恰是把这些样板收起来让应用直接以函数调用的方式操作流。从 Cargo.toml 可以看到它的依赖面非常精简futures、libp2p-core、libp2p-identity、libp2p-swarm、tracing与rand这说明它是一个轻量、专注的通用组件。其核心设计可以概括为三点见 README流是 libp2p 的基本原语其他所有协议都由流实现与传统的NetworkBehaviour不同本模块采用不同的设计思路所有交互都通过Control完成Control可以克隆因此可以在整个应用中被安全共享。从源码结构看模块由 5 个部分组成见 lib.rs 的模块声明behaviour.rsBehaviour 实现、control.rs用户操作入口、handler.rs连接级 Handler、shared.rs跨 Handler 的共享状态与upgrade.rs协议协商升级。其中shared是整个设计的枢纽我们稍后详解。快速上手把 Behaviour 装进 Swarmstream::Behaviour实现了NetworkBehaviour并且实现了Default因此它可以像任何 Behaviour 一样直接组装进 Swarm。仓库中的官方示例 examples/stream/src/main.rs 展示了在 tokio QUIC 传输上的标准装配方式let mut swarm libp2p::SwarmBuilder::with_new_identity() .with_tokio() .with_quic() .with_behaviour(|_| stream::Behaviour::new())? .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(10))) .build(); swarm.listen_on(/ip4/127.0.0.1/udp/0/quic-v1.parse()?)?;装配完成后通过swarm.behaviour().new_control()获取第一个Control实例behaviour.rspub fn new_control(self) - Control { Control::new(self.shared.clone()) }Control内部只是对ArcMutexShared的一个封装克隆它只是克隆Arc成本极低因此你可以无负担地把Control传给任意数量的 task。Inbound注册入站协议并消费入站流基本用法要接受某个StreamProtocol的入站流调用Control::acceptuse libp2p_swarm::{Swarm, StreamProtocol}; use libp2p_stream as stream; use futures::StreamExt as _; let mut swarm: Swarmstream::Behaviour todo!(); let mut control swarm.behaviour().new_control(); let mut incoming control.accept(StreamProtocol::new(/my-protocol)).unwrap(); let handler_future async move { while let Some((peer, stream)) incoming.next().await { // 使用 stream 执行你的协议逻辑。 } };accept的语义要点协议命名StreamProtocol::new(/my-protocol)使用 libp2p 惯例的/协议名形式它与远端节点协商时使用的多流协议 ID 一一对应返回IncomingStreams这是一个实现了futures::Stream的类型其Item为(PeerId, Stream)——PeerId告诉你是谁发起了这条流control.rs重复注册会报错如果同一协议已被注册且尚未注销accept返回AlreadyRegistered这与HashMap的键已存在语义一致。在示例 examples/stream/src/main.rs 中入站侧被放入一个独立的tokio::spawn任务逐个处理 echo 流tokio::spawn(async move { while let Some((peer, stream)) incoming_streams.next().await { match echo(stream).await { /* ... */ } } });资源管理惰性流与背压Control::accept返回的IncomingStreams与其他futures::Stream一样是**惰性lazy**的只有被持续poll才会产生进展所以你必须始终驱动它——示例中通过StreamExt::next完成的正是这件事。IncomingStreams背后的通道是无界还是有界的看 shared.rs 中的accept实现let (sender, receiver) mpsc::channel(0); self.supported_inbound_protocols.insert(protocol.clone(), sender);它创建了一个容量为 0 的futures::channel::mpsc通道。这意味着通道本身不缓存流Shared::on_inbound_stream在try_send失败通道满时直接丢弃入站流并记录tracing::debug日志shared.rs。这正是 README 中所述机制的实现来源如果应用在处理入站流上落后即调用.next()的循环不够快内部会丢弃这些流。这是一个面向 DoS 防护的背压策略宁可丢流也不让接收队列无限膨胀耗尽内存。示例代码 examples/stream/src/main.rs 的注释也明确提示了这一设计意图并告诫如果想提高并发而每条流 spawn 一个 task实际上等于无界缓冲——每个 task 都需要内存激进的远端可能借此让你 OOM。注销与资源回收IncomingStreams被 drop 时其持有的mpsc::Receiver关闭通道另一端的Sender::is_closed()变为true。Shared::accept与supported_inbound_protocols在每次调用时会先retain清理已关闭的 sendershared.rs从而完成协议注销与内存回收。README 明确指出注销后的行为一旦 drop 掉IncomingStreams该协议即被注销。此后远端再尝试用该协议打开流将得到一个协商错误negotiation error。这一行为有对应的集成测试 tests/lib.rsdropping_incoming_streams_deregisters验证测试先让 swarm2 注册/test协议并成功完成一次流通信然后 abort 掉消费任务等价于 dropIncomingStreams再发起open_stream断言返回OpenStreamError::UnsupportedProtocol(_)。Outbound主动打开出站流基本用法打开出站流调用Control::open_streamuse libp2p_swarm::{Swarm, StreamProtocol}; use libp2p_stream as stream; use libp2p_identity::PeerId; let mut swarm: Swarmstream::Behaviour todo!(); let peer_id: PeerId todo!(); let mut control swarm.behaviour().new_control(); let protocol_future async move { let stream control.open_stream(peer_id, StreamProtocol::new(/my-protocol)).await.unwrap(); // 使用 stream 执行你的协议逻辑。 };值得注意的是open_stream是async fn且接收mut self——同一个Control同一时刻只能打开一条流。这与下面要讲的背压机制直接相关。自动拨号未连接也能开流open_stream的一个关键行为是当目标 peer 尚未连接时会自动发起拨号见 control.rs 的文档说明。其实现链路是Shared::sender(peer)在connections表中查找该 peer 的现有连接若存在则复用对应的mpsc::SenderNewStream若不存在则在pending_channels中登记一个通道对并向dial_sender投递一条拨号请求shared.rsBehaviour::poll从dial_receiver中取出 peer发出ToSwarm::Dial并附带PeerCondition::DisconnectedAndNotDialing避免重复拨号behaviour.rs连接建立后Handler收到来自Shared的 pending 通道open_stream的等待被唤醒随即发起子流协商。如果拨号失败Shared::on_dial_failure会把通道中所有 pending 的open_stream调用以io::ErrorKind::NotConnected错误唤醒shared.rs。对应的测试 tests/lib.rsdial_errors_are_propagated向一个随机PeerId开流断言返回io::ErrorKind::NotConnected且错误信息为Dial error: no addresses for peer.。错误类型open_stream返回ResultStream, OpenStreamError错误枚举定义在 control.rs变体含义触发场景UnsupportedProtocol(StreamProtocol)远端不支持该协议协商失败NegotiationFailed或对端已注销该协议Io(std::io::Error)握手期间的 I/O 错误拨号失败、连接重置、超时、通道断裂等OpenStreamError实现了std::error::ErrorIo变体提供source()透传底层错误且标记为#[non_exhaustive]便于未来扩展。示例 examples/stream/src/main.rs 展示了生产环境中的错误处理范式let stream match control.open_stream(peer, ECHO_PROTOCOL).await { Ok(stream) stream, Err(error stream::OpenStreamError::UnsupportedProtocol(_)) { tracing::info!(%peer, %error); return; // 协议不支持是永久性错误直接退出 } Err(error) { // 其他错误可能是暂时性的循环重试 tracing::debug!(%peer, %error); continue; } };背压为什么open_stream要求mut selfREADME 和源码注释都强调了这个设计要点Control支持类似有界通道的背压——每个Control内部有一个保证可用的消息槽位一个Control同一时刻只打开一条流这一点由mut self的签名强制执行control.rs。通道容量为 0意味着只有当Handler真正把NewStream消息取走、并最终把协商好的Stream通过oneshot送回时open_stream的.await才完成天然形成逐条串行、前后衔接的流控。但注释同时警告如果过度克隆Control这个背压机制就会被破坏——每个克隆都获得一个独立槽位克隆越多同时 in-flight 的流越多。所以合理做法是为每个需要并发开流的 task 单独克隆一个Control而不是共享同一个。深入原理Shared 状态与 Handler 分发libp2p-stream的设计精髓在于Shared——一个被ArcMutex包裹的共享状态机Behaviour与每个连接的Handler都持有它的引用shared.rs。它维护了 5 张表字段作用supported_inbound_protocols已注册入站协议 →IncomingStreams通道 sender 的映射connectionsConnectionId→PeerId的映射sendersConnectionId→mpsc::SenderNewStream的映射pending_channels拨号中 peer → 待定通道对的映射dial_sender待拨号 peer 的投递通道避免在poll中加锁之所以要用通道而非直接在poll里加锁取数据是为了避免在NetworkBehaviour::poll内持有锁——dial_sender这个设计behaviour.rs正是为规避锁竞争问题。Handler的职责非常纯粹handler.rs入站侧listen_protocol每次从Shared读取当前注册的协议列表构造Upgrade作为SubstreamProtocol入站流协商完成后FullyNegotiatedInbound事件把(stream, protocol)交回Shared::on_inbound_stream投递到对应的IncomingStreams出站侧poll从自己的receiver取到NewStream请求后发出OutboundSubstreamRequestFullyNegotiatedOutbound或DialUpgradeError则通过oneshot::Sender把结果送回open_stream的等待者。升级错误会被映射为OpenStreamError超时 →TimedOut、协商失败 →UnsupportedProtocol、I/O →Io见 handler.rs。Upgrade本身极简upgrade.rs它实现UpgradeInfo/InboundUpgrade/OutboundUpgradeupgrade_inbound/upgrade_outbound直接把(socket, info)原样返回Error Infallible因为流的协商多流协议选择已经由StreamProtocol列表完成无需额外握手数据。整条调用链可总结为open_stream(peer, proto) └─ Shared::sender(peer) ──► 已有连接? 复用通道 / 无连接? 发起自动拨号 └─ Handler::poll ──► OutboundSubstreamRequest(Upgrade{proto}) └─ 协商完成 ──► oneshot 送回 Stream或映射后的 OpenStreamError accept(proto) └─ Shared::accept ──► 注册 mpsc 通道返回 IncomingStreams └─ Handler 入站协商完成 ──► Shared::on_inbound_stream ──► 投递 (PeerId, Stream)一个完整的 echo 协议示例结合上面所有要点这里给出一个可直接运行的最小 echo 服务器/客户端思路完整可运行代码见 examples/stream/src/main.rs。同一份二进制传入目标地址即为拨号方服务端监听方let mut incoming swarm .behaviour() .new_control() .accept(ECHO_PROTOCOL) .unwrap(); tokio::spawn(async move { while let Some((peer, stream)) incoming.next().await { // echo读一块写一块直到 EOF match echo(stream).await { Ok(n) tracing::info!(%peer, Echoed {n} bytes!), Err(e) tracing::warn!(%peer, Echo failed: {e}), } } });客户端拨号方async fn connection_handler(peer: PeerId, mut control: stream::Control) { loop { tokio::time::sleep(Duration::from_secs(1)).await; let stream match control.open_stream(peer, ECHO_PROTOCOL).await { Ok(stream) stream, Err(error stream::OpenStreamError::UnsupportedProtocol(_)) { tracing::info!(%peer, %error); return; } Err(error) { tracing::debug!(%peer, %error); continue; } }; if let Err(e) send(stream).await { tracing::warn!(%peer, Echo protocol failed: {e}); continue; } } }echo函数在 examples/stream/src/main.rs循环read读到 0 字节对端关闭即结束否则write_all回写。send则写入随机字节并read_exact校验回显最后显式stream.close()。测试与验证行为即规范libp2p-stream的集成测试 tests/lib.rs 用libp2p-swarm-testSwarm::new_ephemeral_tokio在两个临时 Swarm 间建立内存传输连接直接验证了本文介绍的三大行为契约dropping_incoming_streams_deregisterstests/lib.rs入站处理任务被 abortIncomingStreams被 drop后再开流返回UnsupportedProtocol——证明drop 即注销dial_errors_are_propagatedtests/lib.rs对无地址的随机 peer 开流返回io::ErrorKind::NotConnected——证明自动拨号失败的错误传播。此外从 CHANGELOG.md 可以看到该模块的演进脉络0.4.0-alpha 起在accept新流时垃圾回收已注销的流即上文retain清理逻辑0.1.0-alpha 初始发布时就已为OpenStreamError实现Errortrait。当前版本为0.5.0-alphaMSRV 1.88.0仍处于 alpha 阶段API 可能随版本演进调整请以当前仓库源码为准。小结libp2p-stream提供了一种与 libp2p 主流事件驱动 Behaviour截然不同的编程模型以可克隆的Control为唯一入口用异步函数直接读写流。其要点可归纳为入站control.accept(StreamProtocol)注册协议并返回IncomingStreams它必须被持续 poll处理不过来时流会被内部丢弃面向 DoS 的背压drop 句柄即注销协议对端将收到协商错误出站control.open_stream(peer, proto)自动处理拨号与协商mut self保证每个Control单飞背压克隆需节制错误分为协议不支持永久与I/O 错误可能暂时两类原理ArcMutexShared集中管理协议注册、连接映射与待拨号队列Handler只做转发与协商Upgrade零成本透传流验证集成测试与仓库示例共同构成了可复现的行为规范。如果你的协议需求只是在 peer 之间传字节流与其手写全套 Behaviour Handler 样板不如直接在 Swarm 中挂载stream::Behaviour用几十行代码把自定义协议跑起来。【免费下载链接】rust-libp2pThe Rust Implementation of the libp2p networking stack.项目地址: https://gitcode.com/GitHub_Trending/ru/rust-libp2p创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考