From ff590ac70e7a858a64d7a4835a22f6d94b31b886 Mon Sep 17 00:00:00 2001 From: littleblack111 Date: Mon, 20 Oct 2025 23:05:54 +0800 Subject: init --- src/network.rs | 205 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 src/network.rs (limited to 'src/network.rs') diff --git a/src/network.rs b/src/network.rs new file mode 100644 index 0000000..82190a5 --- /dev/null +++ b/src/network.rs @@ -0,0 +1,205 @@ +use std::{ + net::SocketAddr, + sync::{Arc, Weak}, +}; + +use serde::{Deserialize, Serialize}; + +use futures::{SinkExt, StreamExt}; +use tokio::{ + net::{TcpListener, TcpStream}, + spawn, + sync::Mutex, +}; +use tokio_serde::{Framed, formats::Json}; +use tokio_util::codec::Framed as TokioFramed; +use tokio_util::codec::LengthDelimitedCodec; + +use crate::{ + blocks::{Block, BlockChain}, + client::Client, +}; + +#[derive(Deserialize, Serialize, Clone)] +pub enum Request { + Block(Block), + Ibd(Option), +} + +pub struct Network { + this: Weak>, + me: Client, + listener: TcpListener, + peers: Vec>>, + blockchain: Arc>, +} + +impl Network { + pub async fn new(blockchain: Arc>) -> Arc> { + let this = Arc::new(Mutex::new(Self { + me: Client::default(), + this: Weak::new(), + listener: TcpListener::bind("0.0.0.0:6767") + .await + .unwrap(), + peers: Vec::default(), + blockchain, + })); + this.lock() + .await + .this = Arc::downgrade(&this); + this + } + + async fn try_peer(&mut self) { + self.peers + .push(Arc::new(Mutex::new(Peer::new( + self.this + .clone(), + TcpStream::connect("192.168.1.100") + .await + .unwrap(), + )))) + } + + async fn start(&self) { + let this = self + .this + .clone(); + spawn(async move { + if let Some(parent) = this.upgrade() { + let mut net = parent + .lock() + .await; + net.listen() + .await; + } + }); + } + + async fn listen(&mut self) -> ! { + loop { + let (stream, addr) = self + .listener + .accept() + .await + .unwrap(); + + let peer = Arc::new(Mutex::new(Peer::new( + self.this + .clone(), + stream, + ))); + + self.peers + .push(Arc::clone(&peer)); + + spawn(async move { + peer.lock() + .await + .start() + .await; + }); + } + } + + pub async fn broadcast(&mut self, data: Request) { + for p in &self.peers { + p.lock() + .await + .framed + .send(data.clone()) + .await + .unwrap(); + } + } +} + +struct Peer { + parent: Weak>, + framed: Framed, Request, Request, Json>, +} + +impl Peer { + fn new(parent: Weak>, stream: TcpStream) -> Self { + Self { + parent, + framed: Framed::new(TokioFramed::new(stream, LengthDelimitedCodec::new()), Json::default()), + } + } + + async fn start(&mut self) { + loop { + while let Some(req) = self + .framed + .next() + .await + { + self.handle(req.unwrap()) + .await; + } + } + } + + async fn handle(&mut self, req: Request) { + match req { + Request::Block(mut block) => match block.pow { + Some(_) => self + .parent + .upgrade() + .unwrap() + .lock() + .await + .blockchain + .lock() + .await + .store(block), + None => { + block.calc_set_pow(); + let parent = self + .parent + .upgrade() + .unwrap(); + let mut net = parent + .lock() + .await; + net.broadcast(Request::Block(block)) + .await; + } + }, + Request::Ibd(bc) => match bc { + Some(_bc) => { + let parent = self + .parent + .upgrade() + .unwrap(); + let net = parent + .lock() + .await; + let mut _bc_lock = net + .blockchain + .lock() + .await; + } + None => { + let parent = self + .parent + .upgrade() + .unwrap(); + let mut net = parent + .lock() + .await; + let bc = { + let guard = net + .blockchain + .lock() + .await; + guard.clone() + }; + net.broadcast(Request::Ibd(Some(bc))) + .await; + } + }, + } + } +} -- cgit v1.2.3