feat: implementation for domo_node

This commit is contained in:
Strix 2023-10-15 18:06:25 +02:00
parent 405cf8b368
commit 3365ca09a5
No known key found for this signature in database
GPG key ID: 49B2E37B8915B774
8 changed files with 445 additions and 72 deletions

View file

@ -0,0 +1,48 @@
use log::LevelFilter;
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize, Clone)]
pub enum LogLevel {
#[serde(rename = "trace")]
Trace,
#[serde(rename = "debug")]
Debug,
#[serde(rename = "Info")]
Info,
#[serde(rename = "warn")]
Warn,
#[serde(rename = "error")]
Error,
}
impl Into<LevelFilter> for LogLevel {
fn into(self) -> LevelFilter {
match self {
LogLevel::Trace => LevelFilter::Trace,
LogLevel::Debug => LevelFilter::Debug,
LogLevel::Info => LevelFilter::Info,
LogLevel::Warn => LevelFilter::Warn,
LogLevel::Error => LevelFilter::Error,
}
}
}
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct LogConfig {
pub stdout_level: LogLevel,
pub file_level: LogLevel,
}
impl Default for LogConfig {
fn default() -> Self {
Self {
// Linter says it's not fine but it is.
#[cfg(not(debug_assertions))]
stdout_level: LogLevel::Info,
#[cfg(debug_assertions)]
stdout_level: LogLevel::Trace,
file_level: LogLevel::Trace,
}
}
}

View file

@ -0,0 +1,30 @@
use log::LogConfig;
use node::NodeConfig;
use serde::{Deserialize, Serialize};
use std::{fs, io};
use ::log::trace;
pub mod log;
pub mod node;
#[derive(Debug, Serialize, Deserialize, Default, Clone)]
#[serde(default)]
pub struct Config {
pub node: NodeConfig,
pub log: LogConfig,
}
impl Config {
pub fn from_path<S: Into<String>>(path: S) -> io::Result<Self> {
let path = path.into().clone();
trace!("importing config from {}...", path);
let s = fs::read_to_string(path.clone())?;
let c: Config = toml::from_str(s.as_str()).map_err(|_| {
io::Error::new(
io::ErrorKind::InvalidData,
format!("{} has invalid data", path),
)
})?;
Ok(c)
}
}

View file

@ -0,0 +1,39 @@
use serde::{Deserialize, Serialize};
use std::fmt::{Display, Formatter};
#[derive(Debug, Serialize, Deserialize, Clone)]
pub enum NodeType {
#[serde(rename = "master")]
Master,
#[serde(rename = "relay")]
Relay {
master_address: String
},
#[serde(rename = "subnet")]
Subnet {
master_address: String
},
}
impl Display for NodeType {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
match self {
NodeType::Master => write!(f, "master"),
NodeType::Relay { .. } => write!(f, "relay"),
NodeType::Subnet { .. } => write!(f, "subnet"),
}
}
}
impl Default for NodeType {
fn default() -> Self {
Self::Master
}
}
#[derive(Debug, Serialize, Deserialize, Default, Clone)]
pub struct NodeConfig {
#[serde(rename = "type")]
pub node_type: NodeType,
pub device_id: Option<String>
}

View file

@ -0,0 +1,77 @@
use tokio::net::{TcpSocket, TcpStream};
use std::io;
use domo_proto::packet::Packet;
use std::net::SocketAddr;
use log::trace;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use domo_proto::packet::identifier::Identifier;
use domo_proto::packet::packet_data::PacketData;
pub struct Client {
socket_addr: SocketAddr,
}
impl Client {
pub fn new(socket_addr: SocketAddr) -> Self {
Self {
socket_addr
}
}
pub async fn client_connection(self) -> io::Result<ClientConnection> {
ClientConnection::new(self).await
}
}
pub struct ClientConnection {
stream: TcpStream,
}
impl ClientConnection {
pub async fn new(c: Client) -> io::Result<Self> {
let sock = TcpSocket::new_v4()?;
let stream = sock.connect(c.socket_addr).await?;
Ok(Self {
stream
})
}
pub async fn send(&mut self, packet: Packet) -> io::Result<Packet> {
self.stream.write(packet.build_full_packet().as_slice()).await?;
from_tcp_stream(&mut self.stream).await
}
}
pub async fn from_tcp_stream(stream: &mut TcpStream) -> io::Result<Packet> {
// Implement the logic to read data from the TcpStream asynchronously and create a Packet instance
match stream.read_u8().await? {
0x01 => {
// Example: Read packet data and create a Packet::V1 instance
let src = Identifier::from(stream.read_u32().await?); // Assuming you have a similar function for Identifier
let dest = Identifier::from(stream.read_u32().await?); // Assuming you have a similar function for Identifier
let packet_id = Identifier::from(stream.read_u32().await?); // Assuming you have a similar function for Identifier
let reply_to = Identifier::from(stream.read_u32().await?); // Assuming you have a similar function for Identifier
let command = stream.read_u8().await?; // Read u8 directly
let data_length = stream.read_u16().await?; // Read u16 directly
trace!("received => src: {src}, dest: {dest}, packet_id: {packet_id}, reply_to: {reply_to}, command: {command}, data_length: {data_length}");
let mut data = vec![0u8; data_length as usize];
stream.read_exact(&mut data).await?;
// Create a Packet::V1 instance
let packet = Packet::V1 {
src,
dest,
packet_id,
reply_to,
command,
data: PacketData::new(data), // Assuming PacketData has a constructor
};
Ok(packet)
}
_ => Err(io::Error::new(io::ErrorKind::InvalidData, "received invalid data."))
}
}

View file

@ -1,94 +1,109 @@
use std::io;
use std::time::Duration;
use log::{error, info, LevelFilter, trace};
use tokio::io::AsyncWriteExt;
use tokio::net::TcpSocket;
use tokio::sync::mpsc::Receiver;
use tokio::time::sleep;
use crate::config::node::NodeType;
use log::{debug, info, trace};
use std::{env, io};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use connection::Client;
use domo_proto::commands::node_management::NodeManagement;
use domo_proto::packet::ToPacket;
use domo_proto::packet::identifier::Identifier;
use domo_proto::packet::Packet;
use domo_proto::packet::packet_data::PacketData;
use std::cell::RefCell;
async fn app(mut recv: Receiver<Packet>) -> io::Result<()> {
trace!("Entered runtime!");
pub mod config;
pub mod connection;
let sock = TcpSocket::new_v4()?;
let mut stream = sock.connect("127.0.0.1:2000".parse().unwrap()).await?;
while let Some(p) = recv.recv().await {
let bytes = p.build_full_packet();
match stream.write(bytes.as_slice()).await {
Ok(s) => trace!("Packet [id: {}] was sent. [sent bytes: {}/{}]", p.packet_id(), s, bytes.len()),
Err(e) => error!("Packet [id: {}] was not sent properly: {}", p.packet_id(), e)
}
}
stream.shutdown().await?;
Ok(())
thread_local! {
static CONFIG: RefCell<config::Config> = RefCell::new(config::Config::from_path(
env::var("DOMO_CONFIG_PATH")
.unwrap_or("config.toml".into())
).unwrap_or({ info!("Using default config..."); config::Config::default() }))
}
fn main() {
setup_logging()
.expect("could not setup logging");
trace!("Building runtime...");
setup_logging().expect("could not setup logging");
let (send, recv) = tokio::sync::mpsc::channel::<Packet>(128);
debug!("Building tokio runtime...");
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.unwrap();
runtime.spawn(async move {
sleep(Duration::new(1, 0)).await;
let p =Packet::V1 {
src: Identifier::from(0x00000000),
dest: Identifier::from(0x00000000),
packet_id: Identifier::from(0x00000000),
reply_to: Identifier::from(0x00000000),
command: 0x00,
data: PacketData::new(
vec![0x1,0x11,0xDD]
)
};
info!("{:X?}", p.build_full_packet());
send.send(p).await
});
debug!("Built tokio runtime with all features...");
runtime.block_on(async {
info!("Starting app...");
match app(recv).await {
Ok(()) => info!("App exited."),
Err(e) => error!("App exited with error: {}", e)
info!(
"Staring as {} node!",
CONFIG.with_borrow(|c| c.node.node_type.clone())
);
match CONFIG.with_borrow(|c| c.node.node_type.clone()) {
NodeType::Master => runtime.block_on(async {}),
NodeType::Relay { master_address } => {
let client = Client::new(master_address.parse().unwrap());
runtime.block_on(async {
let mut conn = client.client_connection().await.unwrap();
let device_id = match CONFIG.with_borrow(|c| c.node.device_id.clone()) {
Some(s) => {
let device_id = Identifier::from(domo_proto::prelude::as_u32_be(hex::decode(s)
.expect("Could not decode hex device_id").as_slice()));
let p = conn.send(NodeManagement::RegisterNode {
device_id
}.to_v1_packet(
Identifier::from(0),
Identifier::from(0),
Identifier::random(),
Identifier::from(0)
)).await.unwrap();
if p.command() == 0x0E {
panic!("device id is taken")
}
trace!("server assigned us: {}", device_id);
device_id
},
None => {
trace!("requesting device_id...");
let p = conn.send(NodeManagement::RegisterNode {
device_id: Identifier::from(0)
}.to_v1_packet(
Identifier::from(0),
Identifier::from(0),
Identifier::random(),
Identifier::from(0)
)).await;
Identifier::from(p.unwrap().data().get_data())
}
};
CONFIG.with_borrow_mut(move |c| c.node.device_id = Some(hex::encode(device_id.bytes)));
})
}
});
NodeType::Subnet { master_address } => {
runtime.block_on(async {})
}
}
}
fn setup_logging() -> Result<(), fern::InitError> {
let stdout = fern::Dispatch::new()
.level(LevelFilter::Debug)
.chain(io::stdout());
let file = fern::Dispatch::new()
.level(LevelFilter::Trace)
.chain(fern::log_file("node.log")?);
fern::Dispatch::new()
.format(|out, message, record| {
match record.metadata().target() {
_ => out.finish(format_args!(
"[{} - {}] <{}> {}: {}",
humantime::format_rfc3339(std::time::SystemTime::now()),
record.level(),
record.target(),
record.file().unwrap_or("??"),
message
))
}
.format(|out, message, record| match record.metadata().target() {
_ => out.finish(format_args!(
"[{} - {}] <{}> {}: {}",
humantime::format_rfc3339(std::time::SystemTime::now()),
record.level(),
record.target(),
record.file().unwrap_or("??"),
message
)),
})
.chain({
fern::Dispatch::new()
.level(CONFIG.with_borrow(|c| c.log.stdout_level.clone()).into())
.chain(io::stdout())
})
.chain({
fern::Dispatch::new()
.level(CONFIG.with_borrow(|c| c.log.stdout_level.clone()).into())
.chain(fern::log_file("./node.log")?)
})
.chain(stdout)
.chain(file)
.apply()?;
Ok(())
}
}