Rust网络协议库polkadot-node-network-protocol的使用，Polkadot节点网络通信与区块链协议实现

Rust网络协议库polkadot-node-network-protocol的使用，Polkadot节点网络通信与区块链协议实现

安装

在项目目录中运行以下Cargo命令：

cargo add polkadot-node-network-protocol

或者在Cargo.toml中添加以下行：

polkadot-node-network-protocol = "25.0.0"

基本信息

  • 版本: 25.0.0
  • 许可证: GPL-3.0-only
  • 大小: 81.9 KiB

示例代码

以下是使用polkadot-node-network-protocol库的基本示例：

use polkadot_node_network_protocol::{
    request_response::v1::{Request, Response},
    v1 as protocol_v1,
};
use futures::channel::oneshot;
use sc_network::ProtocolName;

// 创建一个简单的网络协议实现
async fn run_network_protocol() {
    // 定义协议名称
    let protocol_name: ProtocolName = "/polkadot/req/1".into();
    
    // 创建请求和响应通道
    let (tx, rx) = oneshot::channel();
    
    // 示例请求
    let request = Request::CollationFetching(protocol_v1::CollationFetchingRequest {
        relay_parent: Default::default(),
        para_id: 100.into(),
    });
    
    // 处理请求（示例）
    tokio::spawn(async move {
        match rx.await {
            Ok(request) => {
                println!("Received request: {:?}", request);
                // 处理请求并发送响应
                let response = Response::CollationFetching(Ok(Default::default()));
                tx.send(response).unwrap();
            }
            Err(e) => println!("Error receiving request: {:?}", e),
        }
    });
    
    // 这里可以添加更多的网络协议处理逻辑
    // ...
}

#[tokio::main]
async fn main() {
    run_network_protocol().await;
}

更完整的示例

以下是一个更完整的Polkadot节点网络通信示例，展示了如何使用该库进行区块链协议实现：

use polkadot_node_network_protocol::{
    v1::{PeerId, AuthorityDiscoveryId},
    request_response::{
        IncomingRequest, OutgoingRequest, RequestResponseConfig,
        v1::{Request, Response},
    },
};
use futures::{
    channel::{mpsc, oneshot},
    StreamExt,
};
use sc_network::{Multiaddr, NetworkService};
use std::sync::Arc;

// 网络服务配置
struct NetworkConfig {
    protocol_name: String,
    max_request_size: u64,
    max_response_size: u64,
    request_timeout: std::time::Duration,
}

impl Default for NetworkConfig {
    fn default() -> Self {
        Self {
            protocol_name: "/polkadot/req/1".into(),
            max_request_size: 1024 * 1024,
            max_response_size: 1024 * 1024,
            request_timeout: std::time::Duration::from_secs(10),
        }
    }
}

// 网络服务实现
struct PolkadotNetworkService {
    network: Arc<NetworkService>,
    request_receiver: mpsc::Receiver<IncomingRequest>,
}

impl PolkadotNetworkService {
    async fn new(network: Arc<NetworkService>, config: NetworkConfig) -> Self {
        let (tx, rx) = mpsc::channel(100);
        
        let cfg = RequestResponseConfig {
            name: config.protocol_name.into(),
            max_request_size: config.max_request_size,
            max_response_size: config.max_response_size,
            request_timeout: config.request_timeout,
            inbound_queue: Some(tx),
        };
        
        network.register_req_protocol(cfg).unwrap();
        
        Self {
            network,
            request_receiver: rx,
        }
    }
    
    async fn run(&mut self) {
        while let Some(request) = self.request_receiver.next().await {
            match request.payload {
                Request::CollationFetching(_) => {
                    // 处理collation fetching请求
                    let response = Response::CollationFetching(Ok(Default::default()));
                    request.pending_response.send(response).unwrap();
                }
                Request::PoVFetching(_) => {
                    // 处理PoV fetching请求
                    let response = Response::PoVFetching(Ok(Default::default()));
                    request.pending_response.send(response).unwrap();
                }
                // 处理其他类型的请求...
            }
        }
    }
    
    async fn send_request(&self, peer: PeerId, request: Request) -> Result<Response, String> {
        let (tx, rx) = oneshot::channel();
        
        let outgoing_request = OutgoingRequest {
            peer,
            payload: request,
            pending_response: tx,
        };
        
        self.network.send_request(outgoing_request);
        
        rx.await.map_err(|e| format!("Request failed: {:?}", e))
    }
}

#[tokio::main]
async fn main() {
    // 初始化网络服务
    let network = Arc::new(NetworkService::new(Default::default()));
    let config = NetworkConfig::default();
    
    // 创建Polkadot网络服务
    let mut polkadot_network = PolkadotNetworkService::new(network, config).await;
    
    // 运行网络服务
    tokio::spawn(async move {
        polkadot_network.run().await;
    });
    
    // 示例：发送请求
    let peer_id = PeerId::random();
    let request = Request::CollationFetching(Default::default());
    
    match polkadot_network.send_request(peer_id, request).await {
        Ok(response) => println!("Received response: {:?}", response),
        Err(e) => println!("Error: {}", e),
    }
}

完整示例代码

以下是一个更完整的Polkadot网络协议实现示例，包含节点发现和请求处理：

use polkadot_node_network_protocol::{
    v1::{PeerId, AuthorityDiscoveryId, OurView, View},
    request_response::{
        IncomingRequest, OutgoingRequest, RequestResponseConfig,
        v1::{Request, Response, CollationFetchingRequest, PoVFetchingRequest},
    },
    peer_set::PeerSet,
};
use futures::{
    channel::{mpsc, oneshot},
    StreamExt, SinkExt,
};
use sc_network::{
    NetworkService, Multiaddr, PeerId as NetworkPeerId, 
    ProtocolName, Event, NetworkEventStream
};
use std::{sync::Arc, time::Duration, collections::HashSet};

// 网络管理器
struct NetworkManager {
    network: Arc<NetworkService>,
    peers: HashSet<PeerId>,
    event_stream: Box<dyn NetworkEventStream>,
}

impl NetworkManager {
    fn new(network: Arc<NetworkService>) -> Self {
        let event_stream = network.event_stream("polkadot_network");
        
        Self {
            network,
            peers: HashSet::new(),
            event_stream,
        }
    }
    
    async fn run(&mut self) {
        while let Some(event) = self.event_stream.next().await {
            match event {
                Event::Connected { peer_id, .. } => {
                    self.peers.insert(peer_id.into());
                    println!("Peer connected: {:?}", peer_id);
                }
                Event::Disconnected { peer_id } => {
                    self.peers.remove(&peer_id.into());
                    println!("Peer disconnected: {:?}", peer_id);
                }
                _ => {}
            }
        }
    }
    
    fn get_peers(&self) -> &HashSet<PeerId> {
        &self.peers
    }
}

// 请求处理器
struct RequestHandler {
    network: Arc<NetworkService>,
    request_receiver: mpsc::Receiver<IncomingRequest>,
}

impl RequestHandler {
    async fn handle_requests(&mut self) {
        while let Some(request) = self.request_receiver.next().await {
            match request.payload {
                Request::CollationFetching(req) => {
                    println!("Received collation fetching request: {:?}", req);
                    let response = Response::CollationFetching(Ok(Default::default()));
                    let _ = request.pending_response.send(response);
                }
                Request::PoVFetching(req) => {
                    println!("Received PoV fetching request: {:?}", req);
                    let response = Response::PoVFetching(Ok(Default::default()));
                    let _ = request.pending_response.send(response);
                }
                // 处理其他请求类型...
            }
        }
    }
}

#[tokio::main]
async fn main() {
    // 初始化网络服务
    let network = Arc::new(NetworkService::new(Default::default()));
    
    // 配置网络协议
    let config = RequestResponseConfig {
        name: "/polkadot/req/1".into(),
        max_request_size: 1024 * 1024,
        max_response_size: 1024 * 1024,
        request_timeout: Duration::from_secs(10),
        inbound_queue: None,
    };
    
    let (request_sender, request_receiver) = mpsc::channel(100);
    let config = RequestResponseConfig {
        inbound_queue: Some(request_sender),
        ..config
    };
    
    // 注册协议
    network.register_req_protocol(config).unwrap();
    
    // 启动网络管理器
    let mut network_manager = NetworkManager::new(network.clone());
    tokio::spawn(async move {
        network_manager.run().await;
    });
    
    // 启动请求处理器
    let mut request_handler = RequestHandler {
        network: network.clone(),
        request_receiver,
    };
    tokio::spawn(async move {
        request_handler.handle_requests().await;
    });
    
    // 示例：广播视图更新
    let our_view = OurView {
        heads: Vec::new(),
        finalized_number: 0,
    };
    
    // 示例：发送请求给随机节点
    if let Some(peer) = network_manager.get_peers().iter().next() {
        let request = Request::CollationFetching(CollationFetchingRequest {
            relay_parent: Default::default(),
            para_id: 100.into(),
        });
        
        let (tx, rx) = oneshot::channel();
        let outgoing_request = OutgoingRequest {
            peer: *peer,
            payload: request,
            pending_response: tx,
        };
        
        network.send_request(outgoing_request);
        
        match rx.await {
            Ok(response) => println!("Received response: {:?}", response),
            Err(e) => println!("Request failed: {:?}", e),
        }
    }
    
    // 保持主线程运行
    tokio::signal::ctrl_c().await.unwrap();
}

注意事项

  1. 该库是Polkadot区块链网络协议的核心实现，主要用于节点间的通信
  2. 使用时需要配合Substrate框架的网络层(sc-network)
  3. 协议涉及多种请求类型，包括CollationFetching、PoVFetching等
  4. 需要正确处理请求超时和错误情况
  5. 实际使用中需要考虑网络拓扑和节点发现机制

以上示例展示了polkadot-node-network-protocol库的基本用法。


1 回复

以下是基于提供的内容整理的完整示例demo，展示了如何使用polkadot-node-network-protocol库实现基本的节点网络通信功能：

use polkadot_node_network_protocol::{
    v1::{BlockAnnounce, Protocol, ProtocolConfig},
    request_response::{v1::DisputeRequest, IncomingRequest, OutgoingRequest},
    PeerId, View, ObservedRole, OurView, PeerSet, UnifiedReputationChange as Rep,
};
use polkadot_primitives::Block;
use futures::channel::oneshot;

#[tokio::main]
async fn main() {
    // 1. 初始化网络协议
    println!("--- 初始化网络协议 ---");
    let view = View::with_head(Block::default());
    let peer_id = PeerId::random();
    let role = ObservedRole::Authority;
    println!("创建网络视图: {:?}", view);
    println!("生成随机PeerId: {}", peer_id);
    println!("观察节点角色: {:?}", role);

    // 2. 处理网络请求
    println!("\n--- 处理网络请求 ---");
    async fn handle_dispute_request(req: IncomingRequest<DisputeRequest>) {
        println!("收到争议请求: {:?}", req.payload);
        let (tx, _rx) = oneshot::channel();
        let outgoing_request = OutgoingRequest {
            payload: req.payload,
            pending_response: tx,
        };
        println!("已创建响应通道: {:?}", outgoing_request);
    }

    // 模拟接收请求
    let dummy_request = IncomingRequest {
        payload: DisputeRequest::default(),
        pending_response: oneshot::channel().0,
    };
    handle_dispute_request(dummy_request).await;

    // 3. 自定义协议行为
    println!("\n--- 自定义协议行为 ---");
    let rep_change = Rep::new_fatal("Bad behavior");
    println!("定义声誉变化: {:?}", rep_change);
    
    let our_view = OurView::new(vec![Default::default()], PeerSet::Collation);
    println!("更新对等节点视图: {:?}", our_view);

    // 4. 发送和接收消息
    println!("\n--- 发送和接收消息 ---");
    let block_announce = BlockAnnounce {
        header: Default::default(),
        state: None,
        data: None,
    };
    let protocol = Protocol::BlockAnnounce(block_announce);
    println!("准备发送协议消息: {:?} 给Peer: {}", protocol, peer_id);

    // 5. 协议配置
    println!("\n--- 协议配置 ---");
    let protocol_config = ProtocolConfig::default();
    println!("默认协议配置: {:?}", protocol_config);
}

示例说明

  1. 初始化网络协议：创建基础网络视图、随机PeerId和节点角色
  2. 请求处理：模拟处理DisputeRequest类型请求并创建响应通道
  3. 自定义行为：定义节点声誉变化和更新对等节点视图
  4. 消息传输：创建并准备发送区块公告消息
  5. 协议配置：使用默认协议配置

运行说明

  1. 需要添加以下依赖到Cargo.toml：
[dependencies]
polkadot-node-network-protocol = { git = "https://github.com/paritytech/polkadot", branch = "master" }
futures = "0.3"
tokio = { version = "1.0", features = ["full"] }
  1. 此示例展示了基本用法，实际生产环境需要：
  • 处理网络错误
  • 实现完整的事件循环
  • 集成实际网络层实现
  1. 高级功能如自定义协议扩展可以参考Polkadot节点源码实现。
回到顶部