update rt to tokio

Cargo.toml

@@ -23,19 +23,17 @@ panic = 'abort'
 log = "0.4"
 rand = "0.8"
 once_cell = "1.7"
-simplelog = "0.8"
+simplelog = "0.10"
 image = "0.23"
 base64 = "0.13"
 sha2 = "0.9"
 blake3 = "0.3"
 aes-gcm = "0.8"
-async-lock = "2.3"
-async-channel = "1.4"
-async-fs = "1.5"
 serde = { version = "1", features = ["derive"] }
 postcard = { version = "0.5", default-features = false, features = ["alloc"] }
 sysinfo = "0.16"
-tdn = { version = "0.4", default-features = false, features = ["full"] }
+tokio = { version = "1", features = ["full"] }
+tdn = { version = "0.5", default-features = false, features = ["full"] }
 tdn_did = { git = "https://github.com/cypherlink/tdn_did", branch="main" }
 tdn_storage = { git = "https://github.com/cypherlink/tdn_storage", branch="main" }
 group-chat_types = { git = "https://github.com/cympletech/group-chat", branch="main" }

src/apps.rs

@@ -1,13 +1,11 @@
 use std::sync::Arc;
-use tdn::{
-    smol::lock::RwLock,
-    types::{
-        group::GroupId,
-        message::RecvType,
-        primitive::{HandleResult, Result},
-        rpc::RpcHandler,
-    },
+use tdn::types::{
+    group::GroupId,
+    message::RecvType,
+    primitive::{HandleResult, Result},
+    rpc::RpcHandler,
 };
+use tokio::sync::RwLock;
 
 use crate::layer::Layer;
 use crate::rpc::RpcState;

src/apps/chat/layer.rs

@@ -1,16 +1,14 @@
 use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 use std::sync::Arc;
-use tdn::{
-    smol::lock::RwLock,
-    types::{
-        group::{EventId, GroupId},
-        message::{RecvType, SendType},
-        primitive::{new_io_error, DeliveryType, HandleResult, PeerAddr, Result},
-        rpc::RpcError,
-    },
+use tdn::types::{
+    group::{EventId, GroupId},
+    message::{RecvType, SendType},
+    primitive::{new_io_error, DeliveryType, HandleResult, PeerAddr, Result},
+    rpc::RpcError,
 };
 use tdn_did::{user::User, Proof};
+use tokio::sync::RwLock;
 
 use crate::event::InnerEvent;
 use crate::layer::{Layer, Online};

src/apps/chat/rpc.rs

@@ -166,8 +166,7 @@ pub(crate) fn new_rpc_handler(handler: &mut RpcHandler<RpcState>) {
                 let mut addrs: HashMap<PeerAddr, GroupId> = HashMap::new();
                 addrs.insert(faddr, friend.gid);
                 let sender = state.group.read().await.sender();
-                tdn::smol::spawn(sleep_waiting_close_stable(sender, HashMap::new(), addrs))
-                    .detach();
+                tokio::spawn(sleep_waiting_close_stable(sender, HashMap::new(), addrs));
             }
 
             let data = postcard::to_allocvec(&LayerEvent::Close).unwrap_or(vec![]);
@@ -207,8 +206,7 @@ pub(crate) fn new_rpc_handler(handler: &mut RpcHandler<RpcState>) {
                 let mut addrs: HashMap<PeerAddr, GroupId> = HashMap::new();
                 addrs.insert(faddr, friend.gid);
                 let sender = state.group.read().await.sender();
-                tdn::smol::spawn(sleep_waiting_close_stable(sender, HashMap::new(), addrs))
-                    .detach();
+                tokio::spawn(sleep_waiting_close_stable(sender, HashMap::new(), addrs));
             }
 
             let data = postcard::to_allocvec(&LayerEvent::Close).unwrap_or(vec![]);

src/apps/group_chat/layer.rs

@@ -1,13 +1,11 @@
 use std::path::PathBuf;
 use std::sync::Arc;
-use tdn::{
-    smol::lock::RwLock,
-    types::{
-        group::GroupId,
-        message::{RecvType, SendType},
-        primitive::{new_io_error, HandleResult, PeerAddr, Result},
-    },
+use tdn::types::{
+    group::GroupId,
+    message::{RecvType, SendType},
+    primitive::{new_io_error, HandleResult, PeerAddr, Result},
 };
+use tokio::sync::RwLock;
 
 use group_chat_types::{
     ConnectProof, Event, JoinProof, LayerConnect, LayerEvent, LayerResult, PackedEvent,

src/apps/group_chat/mod.rs

@@ -10,7 +10,6 @@ pub(crate) fn add_layer(results: &mut HandleResult, gid: GroupId, msg: SendType)
     results.layers.push((gid, GROUP_ID, msg));
 }
 
-pub(crate) use models::GroupChat;
 pub(crate) mod rpc;
 pub(crate) use layer::group_chat_conn;
 pub(crate) use layer::handle as layer_handle;

src/daemon.rs

@@ -2,7 +2,6 @@
 extern crate log;
 
 use std::env::args;
-use tdn::smol::{self, io::Result};
 
 mod account;
 mod apps;
@@ -18,12 +17,13 @@ mod session;
 mod storage;
 mod utils;
 
-fn main() -> Result<()> {
+#[tokio::main]
+async fn main() {
     let db_path = args().nth(1).unwrap_or("./.tdn".to_owned());
 
     if std::fs::metadata(&db_path).is_err() {
         std::fs::create_dir(&db_path).unwrap();
     }
 
-    smol::block_on(server::start(db_path))
+    let _ = server::start(db_path).await;
 }

src/event.rs

@@ -3,7 +3,6 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
-use tdn::smol::{channel::Sender, lock::RwLock};
 use tdn::types::{
     group::{EventId, GroupId},
     message::{SendMessage, SendType},
@@ -11,12 +10,13 @@ use tdn::types::{
 };
 use tdn_did::user::User;
 use tdn_storage::local::DStorage;
+use tokio::sync::{mpsc::Sender, RwLock};
 
 use crate::account::Account;
 use crate::apps::chat::LayerEvent;
 use crate::consensus::Event;
 use crate::group::{Group, GroupEvent};
-use crate::layer::{Layer, Online};
+use crate::layer::Layer;
 use crate::migrate::consensus::{
     ACCOUNT_TABLE_PATH, FILE_TABLE_PATH, FRIEND_TABLE_PATH, MESSAGE_TABLE_PATH, REQUEST_TABLE_PATH,
 };
@@ -323,14 +323,13 @@ impl InnerEvent {
                         let fgid = f.gid;
                         let sender = group.sender();
                         let layer_event = LayerEvent::Message(hash, m.clone());
-                        tdn::smol::spawn(InnerEvent::direct_layer_session(
+                        tokio::spawn(InnerEvent::direct_layer_session(
                             sender,
                             layer_lock,
                             ggid,
                             fgid,
                             layer_event,
-                        ))
-                        .detach();
+                        ));
                     }
 
                     let (msg, _) = m.handle(is_me, gid, group.base(), &db, f.id, hash)?;
@@ -388,20 +387,18 @@ impl InnerEvent {
                     let layer_lock = layer.clone();
                     let ggid = gid.clone();
                     let sender = group.sender();
-                    tdn::smol::spawn(async move {
+                    tokio::spawn(async move {
                         let online = layer_lock.write().await.remove_online(&ggid, &f.gid);
                         if let Some(faddr) = online {
                             let mut addrs: HashMap<PeerAddr, GroupId> = HashMap::new();
                             addrs.insert(faddr, f.gid);
-                            tdn::smol::spawn(rpc::sleep_waiting_close_stable(
+                            tokio::spawn(rpc::sleep_waiting_close_stable(
                                 sender,
                                 HashMap::new(),
                                 addrs,
-                            ))
-                            .detach();
+                            ));
                         }
-                    })
-                    .detach();
+                    });
                     (FRIEND_TABLE_PATH, rfid)
                 } else {
                     return Ok(());
@@ -418,20 +415,18 @@ impl InnerEvent {
                     let layer_lock = layer.clone();
                     let ggid = gid.clone();
                     let sender = group.sender();
-                    tdn::smol::spawn(async move {
+                    tokio::spawn(async move {
                         let online = layer_lock.write().await.remove_online(&ggid, &f.gid);
                         if let Some(faddr) = online {
                             let mut addrs: HashMap<PeerAddr, GroupId> = HashMap::new();
                             addrs.insert(faddr, f.gid);
-                            tdn::smol::spawn(rpc::sleep_waiting_close_stable(
+                            tokio::spawn(rpc::sleep_waiting_close_stable(
                                 sender,
                                 HashMap::new(),
                                 addrs,
-                            ))
-                            .detach();
+                            ));
                         }
-                    })
-                    .detach();
+                    });
 
                     (FRIEND_TABLE_PATH, rfid)
                 } else {
@@ -498,14 +493,13 @@ impl StatusEvent {
                     let rid = f.id;
                     let ggid = gid.clone();
                     let sender = group.sender();
-                    tdn::smol::spawn(async move {
+                    tokio::spawn(async move {
                         if let Ok(running) = layer_lock.write().await.running_mut(&ggid) {
                             if running.check_offline(&rgid, &addr) {
                                 // TODO
                             }
                         }
-                    })
-                    .detach();
+                    });
                 }
             }
         }
@@ -814,20 +808,18 @@ impl SyncEvent {
                             let ggid = gid.clone();
                             let fgid = friend.gid;
                             let sender = group.sender();
-                            tdn::smol::spawn(async move {
+                            tokio::spawn(async move {
                                 let online = layer_lock.write().await.remove_online(&ggid, &fgid);
                                 if let Some(faddr) = online {
                                     let mut addrs: HashMap<PeerAddr, GroupId> = HashMap::new();
                                     addrs.insert(faddr, fgid);
-                                    tdn::smol::spawn(rpc::sleep_waiting_close_stable(
+                                    tokio::spawn(rpc::sleep_waiting_close_stable(
                                         sender,
                                         HashMap::new(),
                                         addrs,
-                                    ))
-                                    .detach();
+                                    ));
                                 }
-                            })
-                            .detach();
+                            });
                         }
 
                         if friend.is_deleted {

src/group.rs

@@ -2,15 +2,13 @@ use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::Arc;
-use tdn::{
-    smol::{channel::Sender, lock::RwLock},
-    types::{
-        group::{EventId, GroupId},
-        message::{RecvType, SendMessage, SendType},
-        primitive::{new_io_error, HandleResult, PeerAddr, Result},
-    },
+use tdn::types::{
+    group::{EventId, GroupId},
+    message::{RecvType, SendMessage, SendType},
+    primitive::{new_io_error, HandleResult, PeerAddr, Result},
 };
 use tdn_did::{user::User, Proof};
+use tokio::sync::{mpsc::Sender, RwLock};
 
 use crate::account::Account;
 use crate::apps::device::rpc as device_rpc;

src/layer.rs

@@ -2,14 +2,12 @@ use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::Arc;
-use tdn::{
-    smol::lock::RwLock,
-    types::{
-        group::GroupId,
-        message::SendType,
-        primitive::{new_io_error, PeerAddr, Result},
-    },
+use tdn::types::{
+    group::GroupId,
+    message::SendType,
+    primitive::{new_io_error, PeerAddr, Result},
 };
+use tokio::sync::RwLock;
 
 use crate::apps::chat::chat_conn;
 use crate::apps::group_chat::{group_chat_conn, GROUP_ID};

src/lib.rs

@@ -3,7 +3,6 @@ extern crate log;
 
 use std::ffi::CStr;
 use std::os::raw::c_char;
-use tdn::smol;
 
 mod account;
 mod apps;
@@ -46,5 +45,7 @@ pub mod android {
 pub extern "C" fn start(db_path: *const c_char) {
     let c_str = unsafe { CStr::from_ptr(db_path) };
     let s_path = c_str.to_str().unwrap_or("./tdn").to_owned();
-    let _ = smol::block_on(server::start(s_path));
+
+    let rt = tokio::runtime::Runtime::new().unwrap();
+    let _ = rt.block_on(server::start(s_path));
 }

src/rpc.rs

@@ -1,15 +1,15 @@
-use async_lock::RwLock;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
-use tdn::{
-    smol::channel::{SendError, Sender},
-    types::{
-        group::GroupId,
-        message::{NetworkType, SendMessage, SendType, StateRequest, StateResponse},
-        primitive::{new_io_error, HandleResult, PeerAddr, Result},
-        rpc::{json, rpc_response, RpcError, RpcHandler, RpcParam},
-    },
+use tdn::types::{
+    group::GroupId,
+    message::{NetworkType, SendMessage, SendType, StateRequest, StateResponse},
+    primitive::{new_io_error, HandleResult, PeerAddr, Result},
+    rpc::{json, rpc_response, RpcError, RpcHandler, RpcParam},
+};
+use tokio::sync::{
+    mpsc::{self, error::SendError, Sender},
+    RwLock,
 };
 
 use crate::apps::app_rpc_inject;
@@ -140,7 +140,7 @@ pub(crate) async fn sleep_waiting_close_stable(
     groups: HashMap<PeerAddr, ()>,
     layers: HashMap<PeerAddr, GroupId>,
 ) -> std::result::Result<(), SendError<SendMessage>> {
-    tdn::smol::Timer::after(std::time::Duration::from_secs(10)).await;
+    tokio::time::sleep(std::time::Duration::from_secs(10)).await;
     for (addr, _) in groups {
         sender
             .send(SendMessage::Group(
@@ -164,11 +164,7 @@ pub(crate) async fn sleep_waiting_close_stable(
 }
 
 #[inline]
-pub(crate) async fn inner_rpc(
-    uid: u64,
-    method: &str,
-    sender: &async_channel::Sender<SendMessage>,
-) -> Result<()> {
+pub(crate) async fn inner_rpc(uid: u64, method: &str, sender: &Sender<SendMessage>) -> Result<()> {
     // Inner network default rpc method. only use in http-rpc.
     if method == "network-stable" || method == "network-dht" {
         let req = match method {
@@ -177,16 +173,16 @@ pub(crate) async fn inner_rpc(
             _ => return Ok(()),
         };
 
-        let (s, r) = async_channel::unbounded::<StateResponse>();
+        let (s, mut r) = mpsc::channel::<StateResponse>(128);
         let _ = sender
             .send(SendMessage::Network(NetworkType::NetworkState(req, s)))
             .await
             .expect("TDN channel closed");
 
         let param = match r.recv().await {
-            Ok(StateResponse::Stable(peers)) => network_stable(peers),
-            Ok(StateResponse::DHT(peers)) => network_dht(peers),
-            Ok(_) | Err(_) => {
+            Some(StateResponse::Stable(peers)) => network_stable(peers),
+            Some(StateResponse::DHT(peers)) => network_dht(peers),
+            Some(_) | None => {
                 return Ok(());
             }
         };
@@ -301,11 +297,10 @@ fn new_rpc_handler(
                 let sender = group_lock.sender();
                 let msg = group_lock.create_message(&gid, addr)?;
                 drop(group_lock);
-                tdn::smol::spawn(async move {
-                    tdn::smol::Timer::after(std::time::Duration::from_secs(2)).await;
+                tokio::spawn(async move {
+                    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
                     let _ = sender.send(SendMessage::Group(gid, msg)).await;
-                })
-                .detach();
+                });
             }
 
             Ok(results)
@@ -409,7 +404,7 @@ fn new_rpc_handler(
             let groups = group_lock.remove_all_running();
             let sender = group_lock.sender();
             drop(group_lock);
-            tdn::smol::spawn(sleep_waiting_close_stable(sender, groups, layers)).detach();
+            tokio::spawn(sleep_waiting_close_stable(sender, groups, layers));
 
             Ok(results)
         },
@@ -455,7 +450,7 @@ fn new_rpc_handler(
             let groups = group_lock.remove_running(&gid);
             let sender = group_lock.sender();
             drop(group_lock);
-            tdn::smol::spawn(sleep_waiting_close_stable(sender, groups, layers)).detach();
+            tokio::spawn(sleep_waiting_close_stable(sender, groups, layers));
 
             debug!("Account Offline: {}.", gid.to_hex());
             // add Remove Group to TDN.

src/server.rs

@@ -3,14 +3,11 @@ use simplelog::{CombinedLogger, Config as LogConfig, LevelFilter};
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::sync::Arc;
-use tdn::{
-    prelude::*,
-    smol::{
-        channel::{SendError, Sender},
-        io::Result,
-        lock::RwLock,
-    },
-    types::primitive::HandleResult,
+use tdn::{prelude::*, types::primitive::HandleResult};
+use tokio::{
+    io::Result,
+    sync::mpsc::{error::SendError, Sender},
+    sync::RwLock,
 };
 
 use crate::account::Account;
@@ -30,7 +27,7 @@ pub static RPC_WS_UID: OnceCell<u64> = OnceCell::new();
 pub async fn start(db_path: String) -> Result<()> {
     let db_path = PathBuf::from(db_path);
     if !db_path.exists() {
-        tdn::smol::fs::create_dir_all(&db_path).await?;
+        tokio::fs::create_dir_all(&db_path).await?;
     }
 
     init_log(db_path.clone());
@@ -61,7 +58,7 @@ pub async fn start(db_path: String) -> Result<()> {
     }
     config.group_ids = me.keys().cloned().collect();
 
-    let (peer_id, sender, recver) = start_with_config(config).await.unwrap();
+    let (peer_id, sender, mut recver) = start_with_config(config).await.unwrap();
     info!("Network Peer id : {}", peer_id.to_hex());
 
     let group = Arc::new(RwLock::new(
@@ -76,9 +73,9 @@ pub async fn start(db_path: String) -> Result<()> {
     let mut now_rpc_uid = 0;
 
     // running session remain task.
-    tdn::smol::spawn(session_remain(layer.clone(), sender.clone())).detach();
+    tokio::spawn(session_remain(layer.clone(), sender.clone()));
 
-    while let Ok(message) = recver.recv().await {
+    while let Some(message) = recver.recv().await {
         match message {
             ReceiveMessage::Group(fgid, g_msg) => {
                 if let Ok(handle_result) =
@@ -124,7 +121,7 @@ pub async fn start(db_path: String) -> Result<()> {
                     .all_layer_conns()
                     .await
                     .unwrap_or(HashMap::new());
-                tdn::smol::spawn(sleep_waiting_reboot(t_sender, g_conns, l_conns)).detach();
+                tokio::spawn(sleep_waiting_reboot(t_sender, g_conns, l_conns));
             }
         }
     }
@@ -138,7 +135,8 @@ async fn sleep_waiting_reboot(
     groups: HashMap<GroupId, Vec<SendType>>,
     layers: HashMap<GroupId, Vec<(GroupId, SendType)>>,
 ) -> std::result::Result<(), SendError<SendMessage>> {
-    tdn::smol::Timer::after(std::time::Duration::from_secs(10)).await;
+    tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+
     for (gid, conns) in groups {
         for conn in conns {
             sender.send(SendMessage::Group(gid, conn)).await?;
@@ -156,7 +154,7 @@ async fn sleep_waiting_reboot(
 
 async fn session_remain(layer: Arc<RwLock<Layer>>, sender: Sender<SendMessage>) -> Result<()> {
     loop {
-        tdn::smol::Timer::after(std::time::Duration::from_secs(120)).await;
+        tokio::time::sleep(std::time::Duration::from_secs(120)).await;
         if let Some(uid) = RPC_WS_UID.get() {
             let mut layer_lock = layer.write().await;
             let mut rpcs = vec![];
@@ -267,6 +265,7 @@ pub fn init_log(mut db_path: PathBuf) {
         LevelFilter::Debug,
         LogConfig::default(),
         simplelog::TerminalMode::Mixed,
+        simplelog::ColorChoice::Auto,
     )])
     .unwrap();
 

src/storage.rs

@@ -1,8 +1,8 @@
-use async_fs as fs;
 use image::{load_from_memory, DynamicImage, GenericImageView};
 use rand::{distributions::Alphanumeric, thread_rng, Rng};
 use std::path::PathBuf;
 use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::fs;
 
 use tdn::types::{
     group::GroupId,
@@ -84,7 +84,7 @@ pub(crate) fn write_file_sync(
     path.push(gid.to_hex());
     path.push(FILES_DIR);
     path.push(name);
-    tdn::smol::spawn(async move { fs::write(path, bytes).await }).detach();
+    tokio::spawn(async move { fs::write(path, bytes).await });
 
     Ok(name.to_owned())
 }
@@ -130,14 +130,13 @@ pub(crate) fn write_image_sync(base: &PathBuf, gid: &GroupId, bytes: Vec<u8>) ->
     let mut thumb_path = path.clone();
     thumb_path.push(THUMB_DIR);
     thumb_path.push(name.clone());
-    tdn::smol::spawn(async move {
+    tokio::spawn(async move {
         let _ = thumb.save(thumb_path);
-    })
-    .detach();
+    });
 
     path.push(IMAGE_DIR);
     path.push(name.clone());
-    tdn::smol::spawn(async move { fs::write(path, bytes).await }).detach();
+    tokio::spawn(async move { fs::write(path, bytes).await });
 
     Ok(name)
 }
@@ -152,10 +151,9 @@ pub(crate) async fn write_image(base: &PathBuf, gid: &GroupId, bytes: &[u8]) ->
     let mut thumb_path = path.clone();
     thumb_path.push(THUMB_DIR);
     thumb_path.push(name.clone());
-    tdn::smol::spawn(async move {
+    tokio::spawn(async move {
         let _ = thumb.save(thumb_path);
-    })
-    .detach();
+    });
 
     path.push(IMAGE_DIR);
     path.push(name.clone());
@@ -236,7 +234,7 @@ pub(crate) fn write_avatar_sync(
     path.push(gid.to_hex());
     path.push(AVATAR_DIR);
     path.push(avatar_png(remote));
-    tdn::smol::spawn(async move { fs::write(path, bytes).await }).detach();
+    tokio::spawn(async move { fs::write(path, bytes).await });
     Ok(())
 }
 
@@ -258,7 +256,7 @@ pub(crate) fn delete_avatar_sync(base: &PathBuf, gid: &GroupId, remote: &GroupId
     path.push(AVATAR_DIR);
     path.push(avatar_png(remote));
     if path.exists() {
-        tdn::smol::spawn(async move { fs::remove_file(path).await }).detach();
+        tokio::spawn(async move { fs::remove_file(path).await });
     }
     Ok(())
 }
@@ -300,7 +298,7 @@ pub(crate) fn write_record_sync(
     path.push(gid.to_hex());
     path.push(RECORD_DIR);
     path.push(format!("{}_{}.m4a", fid, datetime));
-    tdn::smol::spawn(async move { fs::write(path, bytes).await }).detach();
+    tokio::spawn(async move { fs::write(path, bytes).await });
 
     Ok(format!("{}-{}_{}.m4a", t, fid, datetime))
 }