From a6016c7eeb11fc8aeaa1a3b160b970b15362695d Mon Sep 17 00:00:00 2001 From: hozan23 Date: Sun, 19 May 2024 22:20:03 +0200 Subject: add tokio examples to p2p, jsonrpc, and net crates --- net/examples/tcp_codec_tokio/src/main.rs | 57 ++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) create mode 100644 net/examples/tcp_codec_tokio/src/main.rs (limited to 'net/examples/tcp_codec_tokio/src') diff --git a/net/examples/tcp_codec_tokio/src/main.rs b/net/examples/tcp_codec_tokio/src/main.rs new file mode 100644 index 0000000..9865bfa --- /dev/null +++ b/net/examples/tcp_codec_tokio/src/main.rs @@ -0,0 +1,57 @@ +use std::time::Duration; + +use karyon_core::async_util::sleep; + +use karyon_net::{ + codec::{Codec, Decoder, Encoder}, + tcp, ConnListener, Connection, Endpoint, Result, +}; + +#[derive(Clone)] +struct NewLineCodec {} + +impl Codec for NewLineCodec { + type Item = String; +} + +impl Encoder for NewLineCodec { + type EnItem = String; + fn encode(&self, src: &Self::EnItem, dst: &mut [u8]) -> Result { + dst[..src.len()].copy_from_slice(src.as_bytes()); + Ok(src.len()) + } +} + +impl Decoder for NewLineCodec { + type DeItem = String; + fn decode(&self, src: &mut [u8]) -> Result> { + match src.iter().position(|&b| b == b'\n') { + Some(i) => Ok(Some((i + 1, String::from_utf8(src[..i].to_vec()).unwrap()))), + None => Ok(None), + } + } +} + +#[tokio::main] +async fn main() { + let endpoint: Endpoint = "tcp://127.0.0.1:3000".parse().unwrap(); + + let config = tcp::TcpConfig::default(); + + let listener = tcp::listen(&endpoint, config.clone(), NewLineCodec {}) + .await + .unwrap(); + tokio::spawn(async move { + if let Ok(conn) = listener.accept().await { + loop { + let msg = conn.recv().await.unwrap(); + println!("Receive a message: {:?}", msg); + } + }; + }); + + let conn = tcp::dial(&endpoint, config, NewLineCodec {}).await.unwrap(); + conn.send("hello".to_string()).await.unwrap(); + conn.send(" world\n".to_string()).await.unwrap(); + sleep(Duration::from_secs(1)).await; +} -- cgit v1.2.3