refactor: breaking refactor of domo_proto and domo_node
This commit is contained in:
parent
be4425aff9
commit
15971aa4fa
31 changed files with 902 additions and 393 deletions
2
domo_node/Cargo.lock
generated
2
domo_node/Cargo.lock
generated
|
|
@ -91,7 +91,7 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "domo_proto"
|
||||
version = "0.1.0"
|
||||
version = "0.2.0"
|
||||
dependencies = [
|
||||
"crc32fast",
|
||||
"rand",
|
||||
|
|
|
|||
|
|
@ -1,3 +1,3 @@
|
|||
[node.type.relay]
|
||||
[node.type.bridge]
|
||||
bind = "127.0.0.1:4481"
|
||||
master_address = "127.0.0.1:4480"
|
||||
|
|
@ -2,4 +2,4 @@
|
|||
bind = "127.0.0.1:4480"
|
||||
|
||||
[node]
|
||||
device_id = "000000ff"
|
||||
device_id = "ffffffff"
|
||||
|
|
@ -1,19 +1,18 @@
|
|||
use serde::{Deserialize, Serialize};
|
||||
use std::fmt::{Display, Formatter};
|
||||
use std::net::SocketAddr;
|
||||
use domo_proto::packet::identifier::Identifier;
|
||||
use domo_proto::identifier::Identifier;
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, Clone)]
|
||||
pub enum NodeType {
|
||||
#[serde(rename = "master")]
|
||||
Master {
|
||||
bind: SocketAddr,
|
||||
identifier: String
|
||||
},
|
||||
#[serde(rename = "relay")]
|
||||
Relay {
|
||||
Bridge {
|
||||
bind: SocketAddr,
|
||||
master_address: String
|
||||
master_address: SocketAddr
|
||||
},
|
||||
#[serde(rename = "subnet")]
|
||||
Subnet {
|
||||
|
|
@ -25,7 +24,7 @@ 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::Bridge { .. } => write!(f, "relay"),
|
||||
NodeType::Subnet { .. } => write!(f, "subnet"),
|
||||
}
|
||||
}
|
||||
|
|
@ -35,14 +34,23 @@ impl Default for NodeType {
|
|||
fn default() -> Self {
|
||||
Self::Master {
|
||||
bind: SocketAddr::from(([0,0,0,0], 4480)),
|
||||
identifier: hex::encode(Identifier::random().to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, Default, Clone)]
|
||||
#[derive(Debug, Serialize, Deserialize, Clone)]
|
||||
pub struct NodeConfig {
|
||||
#[serde(rename = "type")]
|
||||
pub node_type: NodeType,
|
||||
pub device_id: Option<String>
|
||||
#[serde(default)]
|
||||
pub device_id: String
|
||||
}
|
||||
|
||||
impl Default for NodeConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
node_type: NodeType::default(),
|
||||
device_id: Identifier::random().to_string()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1 @@
|
|||
pub mod server;
|
||||
57
domo_node/src/connection/server.rs
Normal file
57
domo_node/src/connection/server.rs
Normal file
|
|
@ -0,0 +1,57 @@
|
|||
use tokio::net::UdpSocket;
|
||||
use domo_proto::identifier::Identifier;
|
||||
use std::collections::HashMap;
|
||||
use std::net::SocketAddr;
|
||||
use std::io;
|
||||
use domo_proto::packet::{FULL_PACKET_SIZE, Packet};
|
||||
use std::io::ErrorKind;
|
||||
|
||||
pub struct Device {
|
||||
/// Device identifier
|
||||
pub device_id: Identifier,
|
||||
/// Device socket address
|
||||
pub socket_addr: SocketAddr,
|
||||
}
|
||||
|
||||
pub struct Server {
|
||||
sock: UdpSocket,
|
||||
device_id: Identifier,
|
||||
devices: HashMap<Identifier, Device>
|
||||
}
|
||||
|
||||
impl Server {
|
||||
pub async fn new(socket_addr: SocketAddr, device_id: Option<Identifier>) -> io::Result<Self> {
|
||||
Ok(Self {
|
||||
sock: UdpSocket::bind(socket_addr).await?,
|
||||
device_id: device_id.unwrap_or(Identifier::random()),
|
||||
devices: HashMap::new()
|
||||
})
|
||||
}
|
||||
|
||||
/// Receive 1 packet.
|
||||
pub async fn recv_packet(&mut self) -> (io::Result<Packet>, Option<SocketAddr>) {
|
||||
let mut buf = vec![0; FULL_PACKET_SIZE];
|
||||
let (size, socket_addr) = match self.sock.recv_from(&mut buf).await {
|
||||
Ok(v) => v,
|
||||
Err(e) => return (Err(e), None)
|
||||
};
|
||||
|
||||
(
|
||||
Packet::try_from(buf[..size].to_vec())
|
||||
.map_err(|_| io::Error::new(ErrorKind::InvalidData, "Received invalid data")),
|
||||
Some(socket_addr)
|
||||
)
|
||||
}
|
||||
|
||||
pub fn device_id(&self) -> Identifier {
|
||||
self.device_id.clone()
|
||||
}
|
||||
|
||||
pub fn sock(&self) -> &UdpSocket {
|
||||
&self.sock
|
||||
}
|
||||
|
||||
pub fn devices(& self) -> &HashMap<Identifier, Device> {
|
||||
&self.devices
|
||||
}
|
||||
}
|
||||
|
|
@ -1,14 +1,16 @@
|
|||
use crate::config::node::NodeType;
|
||||
use log::{debug, error, info, trace};
|
||||
use log::{debug, error, info, trace, warn};
|
||||
use std::{env, io};
|
||||
|
||||
|
||||
use std::cell::RefCell;
|
||||
use std::future::Future;
|
||||
use std::net::SocketAddr;
|
||||
use std::str::FromStr;
|
||||
use std::error::Error;
|
||||
use tokio::net::UdpSocket;
|
||||
use domo_proto::packet::{FULL_PACKET_SIZE, ToPacket};
|
||||
use connection::server::Server;
|
||||
use domo_proto::commands::node_management::NodeManagementCommand;
|
||||
use domo_proto::identifier::Identifier;
|
||||
use domo_proto::packet::raw_packet::RawPacket;
|
||||
|
||||
pub mod prelude;
|
||||
pub mod config;
|
||||
pub mod connection;
|
||||
|
||||
|
|
@ -38,25 +40,89 @@ fn main() -> io::Result<()> {
|
|||
|
||||
|
||||
match CONFIG.with_borrow(|c| c.node.node_type.clone()) {
|
||||
NodeType::Master { bind, identifier } => runtime.block_on(async {
|
||||
let sock = UdpSocket::bind(bind).await.unwrap();
|
||||
info!("bound to: {}", sock.local_addr().unwrap());
|
||||
NodeType::Master { bind } => runtime.block_on(async {
|
||||
let mut server = Server::new(
|
||||
bind,
|
||||
Some(
|
||||
Identifier::from(
|
||||
hex::decode(CONFIG.with_borrow(|c| c.node.device_id.clone()))
|
||||
.unwrap()
|
||||
)
|
||||
),
|
||||
).await.unwrap();
|
||||
|
||||
info!("bound server on {} as {}", server.sock().local_addr().unwrap(), server.device_id());
|
||||
|
||||
loop {
|
||||
let mut buf = vec![0; domo_proto::packet::FULL_PACKET_SIZE];
|
||||
match sock.recv_from(&mut buf).await {
|
||||
Ok((size, addr)) => {
|
||||
trace!("recv[{addr} -> {size}]: {:?}", &buf[..size]);
|
||||
},
|
||||
Err(e) => {
|
||||
match server.recv_packet().await {
|
||||
(Ok(packet), Some(source)) => {
|
||||
if packet.dest != server.device_id() {
|
||||
if let Some(d) = server.devices().get(&packet.dest) {
|
||||
if let Err(e) = server.sock().send_to(packet.build_full_packet().as_slice(), d.socket_addr).await {
|
||||
error!("Error forwarding to {}: {}", d.device_id, e);
|
||||
continue
|
||||
}
|
||||
trace!("fwd {} -> {}", packet.packet_id, packet.dest);
|
||||
} else {
|
||||
if let Err(e) = server.sock().send_to(
|
||||
prelude::quick_err::net_dest_unreachable(
|
||||
server.device_id(),
|
||||
packet.dest,
|
||||
packet.packet_id
|
||||
).build_full_packet().as_slice(),
|
||||
source
|
||||
).await {
|
||||
error!("Could not send error packet: {e}");
|
||||
};
|
||||
}
|
||||
continue;
|
||||
}
|
||||
match packet.command >> 4 {
|
||||
// Node management
|
||||
0x0 => {
|
||||
if let Ok(nm) = NodeManagementCommand::try_from(RawPacket::from(packet)) {
|
||||
trace!("{nm:?}");
|
||||
}
|
||||
}
|
||||
// Property control
|
||||
0x1 => {}
|
||||
_ => {
|
||||
let device_id = server.device_id();
|
||||
match server.sock().send_to(
|
||||
prelude::quick_err::net_invalid_packet(
|
||||
device_id,
|
||||
packet.src,
|
||||
packet.packet_id,
|
||||
).build_full_packet().as_slice(),
|
||||
source,
|
||||
).await {
|
||||
Ok(_) => {}
|
||||
Err(e) => error!("could not send error packet: {e}")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
(Err(e), Some(source)) => {
|
||||
error!("{source} sent intelligible data: {e}");
|
||||
let device_id = server.device_id();
|
||||
if let Err(e) = server.sock().send_to(
|
||||
prelude::quick_err::net_broken_packet(
|
||||
device_id,
|
||||
Identifier::random(),
|
||||
).build_full_packet().as_slice(),
|
||||
source,
|
||||
).await {
|
||||
error!("could not send error packet: {e}");
|
||||
}
|
||||
}
|
||||
_ => warn!("dropped intelligible packet")
|
||||
}
|
||||
}
|
||||
}),
|
||||
NodeType::Relay { bind, master_address } => runtime.block_on(async {
|
||||
NodeType::Bridge { bind, master_address } => runtime.block_on(async {
|
||||
|
||||
}),
|
||||
NodeType::Subnet { master_address: _ } => {
|
||||
runtime.block_on(async {})
|
||||
}
|
||||
NodeType::Subnet { master_address: _ } => unimplemented!()
|
||||
}
|
||||
|
||||
Ok(())
|
||||
|
|
|
|||
42
domo_node/src/prelude.rs
Normal file
42
domo_node/src/prelude.rs
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
pub mod quick_err {
|
||||
use domo_proto::commands::node_management::{NodeManagementCommand, errors::NodeManagementError};
|
||||
use domo_proto::identifier::Identifier;
|
||||
use domo_proto::packet::{Packet, ToPacket};
|
||||
|
||||
pub fn net_invalid_packet(src: Identifier, dest: Identifier, reply_to: Identifier) -> Packet {
|
||||
NodeManagementCommand::Error {
|
||||
error_code: NodeManagementError::net_invalid_packet.into(),
|
||||
// todo: non rude meta
|
||||
metadata: "don't send me stupid data, stupid".to_string().into_bytes()
|
||||
}.to_packet(
|
||||
src,
|
||||
dest,
|
||||
Identifier::random(),
|
||||
reply_to
|
||||
)
|
||||
}
|
||||
|
||||
pub fn net_broken_packet(src: Identifier, reply_to: Identifier) -> Packet {
|
||||
NodeManagementCommand::Error {
|
||||
error_code: NodeManagementError::net_broken_packet.into(),
|
||||
metadata: "intelligible data, send again.".to_string().into_bytes()
|
||||
}.to_packet(
|
||||
src,
|
||||
Identifier::default(),
|
||||
Identifier::random(),
|
||||
reply_to
|
||||
)
|
||||
}
|
||||
|
||||
pub fn net_dest_unreachable(src: Identifier, dest: Identifier, reply_to: Identifier) -> Packet {
|
||||
NodeManagementCommand::Error {
|
||||
error_code: NodeManagementError::net_dest_unreachable.into(),
|
||||
metadata: "packet could not reach destination".to_string().into_bytes()
|
||||
}.to_packet(
|
||||
src,
|
||||
dest,
|
||||
Identifier::random(),
|
||||
reply_to
|
||||
)
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue