response plumbing

[hippotat.git] / server / server.rs

diff --git a/server/server.rs b/server/server.rs
index 022d00bafc6c214b9c670aef5b91bb9375136e04..dcadd54118f83b4b5124579e5da20650fb25d07c 100644 (file)
--- a/server/server.rs
+++ b/server/server.rs
@@ -4,11 +4,12 @@
 
 use hippotat::prelude::*;
 
-mod sclient;
+mod suser;
+mod slocal;
 mod sweb;
 
-pub use sweb::{WebRequest, WebResponse};
-pub use sclient::Client;
+pub use sweb::{WebRequest, WebResponse, WebResponseBody};
+pub use suser::User;
 
 #[derive(StructOpt,Debug)]
 pub struct Opts {
@@ -25,38 +26,69 @@ pub const INTERNAL_QUEUE: usize = 15; // xxx: config
 #[derive(Debug)]
 pub struct Global {
   config: config::InstanceConfigGlobal,
-  all_clients: HashMap<ClientName, Client>,
+  local_rx: mpsc::Sender<RoutedPacket>,
+  all_clients: HashMap<ClientName, User>,
 }
 
-pub type RoutedPacket = Box<[u8]>; // not MIME data
+pub struct RoutedPacket {
+  pub data: RoutedPacketData,
+//  pub source: Option<ClientName>, // for eh, tracing, etc.
+}
+
+// not MIME data, valid SLIP (checked)
+pub type RoutedPacketData = Box<[u8]>;
+
+// loop prevention
+// we don't decrement the ttl (naughty) but loops cannot arise
+// because only the server has any routing code, and server
+// has no internal loops, so worst case is
+//  client if -> client -> server -> client' -> client if'
+// and the ifs will decrement the ttl.
+mod may_route {
+  #[derive(Clone,Debug)]
+  pub struct MayRoute(());
+  impl MayRoute {
+    pub fn came_from_outside_hippotatd() -> Self { Self(()) }
+  }
+}
+pub use may_route::MayRoute;
 
-#[throws(PacketError)]
 pub async fn route_packet(global: &Global,
-                          conn: &str, link: &(dyn Display + Sync),
-                          packet: RoutedPacket, daddr: IpAddr)
+                          transport_conn: &str, source: Option<&ClientName>,
+                          packet: RoutedPacketData, daddr: IpAddr,
+                          _may_route: MayRoute)
 {
   let c = &global.config;
   let len = packet.len();
   let trace = |how: &str, why: &str| {
-    trace!("{} {} {} {} {:?} len={}",
-           conn, link, how, why, daddr, len);
+    trace!("{} to={:?} came=={} user={} len={} {}",
+           how, daddr, transport_conn,
+           match source {
+             Some(s) => (s as &dyn Display),
+             None => &"local",
+           },
+           len, why);
   };
 
   let (dest, why) =
     if daddr == c.vaddr || ! c.vnetwork.iter().any(|n| n.contains(&daddr)) {
-      (None, "ipif-inbound-xxx")
+      (Some(&global.local_rx), "via=local")
     } else if daddr == c.vrelay {
-      (None, "vrelay")
+      (None, " vrelay")
     } else if let Some(client) = global.all_clients.get(&ClientName(daddr)) {
-      (Some(&client.route), "client")
+      (Some(&client.route), "via=client")
     } else {
       (None, "no-client")
     };
 
   let dest = if let Some(d) = dest { d } else {
-    trace("discard", why); return;
+    trace("discard ", why); return;
   };
 
+  let packet = RoutedPacket {
+    data: packet,
+//    source: source.cloned(),
+  };
   match dest.send(packet).await {
     Ok(()) => trace("forward", why),
     Err(_) => trace("task-crashed!", why),
@@ -71,7 +103,7 @@ async fn main() {
     String,
   )> = vec![];
 
-  let (global, ipif) = config::startup(
+  let global = config::startup(
     "hippotatd", LinkEnd::Server,
     &opts.config, &opts.log, |ics|
   {
@@ -97,15 +129,20 @@ async fn main() {
       client_handles_send,
     ).map(|(ic, (web_send, route_send))| {
       (ic.link.client,
-       Client {
+       User {
          ic: ic.clone(),
          web: web_send,
          route: route_send,
        })
     }).collect();
 
+    let (local_rx_send, local_tx_recv) = mpsc::channel(
+      50 // xxx configurable?
+    );
+
     let global = Arc::new(Global {
       config: global_config,
+      local_rx: local_rx_send,
       all_clients,
     });
 
@@ -116,7 +153,7 @@ async fn main() {
       let global_ = global.clone();
       let ic_ = ic.clone();
       tasks.push((tokio::spawn(async move {
-        sclient::run(global_, ic_, web_recv, route_recv)
+        suser::run(global_, ic_, web_recv, route_recv)
           .await.void_unwrap_err()
       }), format!("client {}", &ic)));
     }
@@ -147,8 +184,15 @@ async fn main() {
       });
       tasks.push((task, format!("http server {}", addr)));
     }
-    
-    Ok((global, ipif))
+
+    let global_ = global.clone();
+    let ipif = tokio::task::spawn(async move {
+      slocal::run(global_, local_tx_recv, ipif).await
+        .void_unwrap_err()
+    });
+    tasks.push((ipif, format!("ipif")));
+
+    Ok(global)
   });
 
   let died = future::select_all(
@@ -156,7 +200,5 @@ async fn main() {
   ).await;
   error!("xxx {:?}", &died);
 
-  ipif.quitting(None).await;
-
   dbg!(global);
 }