Encryption and One-Way-Control (#200)

This is a major rewrite of the core networking logic enabling one-way control and encryption through the webrtc-dtls crate.

closes #164 
closes #104

src/dns.rs

@@ -1,63 +1,106 @@
-use local_channel::mpsc::Receiver;
 use std::net::IpAddr;
 
-use hickory_resolver::{error::ResolveError, TokioAsyncResolver};
+use local_channel::mpsc::{channel, Receiver, Sender};
+use tokio::task::{spawn_local, JoinHandle};
+
+use hickory_resolver::{error::ResolveError, TokioAsyncResolver};
+use tokio_util::sync::CancellationToken;
 
-use crate::server::Server;
 use lan_mouse_ipc::ClientHandle;
 
 pub(crate) struct DnsResolver {
+    cancellation_token: CancellationToken,
+    task: Option<JoinHandle<()>>,
+    request_tx: Sender<DnsRequest>,
+    event_rx: Receiver<DnsEvent>,
+}
+
+struct DnsRequest {
+    handle: ClientHandle,
+    hostname: String,
+}
+
+pub(crate) enum DnsEvent {
+    Resolving(ClientHandle),
+    Resolved(ClientHandle, String, Result<Vec<IpAddr>, ResolveError>),
+}
+
+struct DnsTask {
     resolver: TokioAsyncResolver,
-    dns_request: Receiver<ClientHandle>,
+    request_rx: Receiver<DnsRequest>,
+    event_tx: Sender<DnsEvent>,
+    cancellation_token: CancellationToken,
 }
 
 impl DnsResolver {
-    pub(crate) fn new(dns_request: Receiver<ClientHandle>) -> Result<Self, ResolveError> {
+    pub(crate) fn new() -> Result<Self, ResolveError> {
         let resolver = TokioAsyncResolver::tokio_from_system_conf()?;
-        Ok(Self {
+        let (request_tx, request_rx) = channel();
+        let (event_tx, event_rx) = channel();
+        let cancellation_token = CancellationToken::new();
+        let dns_task = DnsTask {
             resolver,
-            dns_request,
+            request_rx,
+            event_tx,
+            cancellation_token: cancellation_token.clone(),
+        };
+        let task = Some(spawn_local(dns_task.run()));
+        Ok(Self {
+            cancellation_token,
+            task,
+            event_rx,
+            request_tx,
         })
     }
 
-    async fn resolve(&self, host: &str) -> Result<Vec<IpAddr>, ResolveError> {
-        let response = self.resolver.lookup_ip(host).await?;
-        for ip in response.iter() {
-            log::info!("{host}: adding ip {ip}");
-        }
-        Ok(response.iter().collect())
+    pub(crate) fn resolve(&self, handle: ClientHandle, hostname: String) {
+        let request = DnsRequest { handle, hostname };
+        self.request_tx.send(request).expect("channel closed");
     }
 
-    pub(crate) async fn run(mut self, server: Server) {
+    pub(crate) async fn event(&mut self) -> DnsEvent {
+        self.event_rx.recv().await.expect("channel closed")
+    }
+
+    pub(crate) async fn terminate(&mut self) {
+        self.cancellation_token.cancel();
+        self.task.take().expect("task").await.expect("join error");
+    }
+}
+
+impl DnsTask {
+    async fn run(mut self) {
+        let cancellation_token = self.cancellation_token.clone();
         tokio::select! {
-            _ = server.cancelled() => {},
-            _ = self.do_dns(&server) => {},
+            _ = self.do_dns() => {},
+            _ = cancellation_token.cancelled() => {},
         }
     }
 
-    async fn do_dns(&mut self, server: &Server) {
-        loop {
-            let handle = self.dns_request.recv().await.expect("channel closed");
+    async fn do_dns(&mut self) {
+        while let Some(dns_request) = self.request_rx.recv().await {
+            let DnsRequest { handle, hostname } = dns_request;
 
-            /* update resolving status */
-            let hostname = match server.get_hostname(handle) {
-                Some(hostname) => hostname,
-                None => continue,
-            };
+            self.event_tx
+                .send(DnsEvent::Resolving(handle))
+                .expect("channel closed");
 
-            log::info!("resolving ({handle}) `{hostname}` ...");
-            server.set_resolving(handle, true);
+            /* spawn task for dns request */
+            let event_tx = self.event_tx.clone();
+            let resolver = self.resolver.clone();
+            let cancellation_token = self.cancellation_token.clone();
 
-            let ips = match self.resolve(&hostname).await {
-                Ok(ips) => ips,
-                Err(e) => {
-                    log::warn!("could not resolve host '{hostname}': {e}");
-                    vec![]
+            tokio::task::spawn_local(async move {
+                tokio::select! {
+                    ips = resolver.lookup_ip(&hostname) => {
+                       let ips = ips.map(|ips| ips.iter().collect::<Vec<_>>());
+                       event_tx
+                           .send(DnsEvent::Resolved(handle, hostname, ips))
+                           .expect("channel closed");
+                    }
+                    _ = cancellation_token.cancelled() => {},
                 }
-            };
-
-            server.update_dns_ips(handle, ips);
-            server.set_resolving(handle, false);
+            });
         }
     }
 }