add version to debug page

Cargo.toml

@@ -1,39 +1,59 @@
 [package]
 name = "centralex"
-version = "0.1.0"
+version = "1.0.1"
 edition = "2021"
 
 [profile.release]
-debug = true
-
+strip = true
+lto = true
+codegen-units = 1
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-tokio = { version = "1.24.2", features = ["macros", "rt-multi-thread", "net", "io-util", "sync", "time"] }
+tokio = { version = "1.24.2", features = [
+    "macros",
+    "rt-multi-thread",
+    "net",
+    "io-util",
+    "sync",
+    "time",
+] }
 time = { version = "0.3.20", features = ["local-offset", "macros"] }
 bytemuck = { version = "1.13.0", features = ["derive"] }
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = "1.0.91"
-hyper = { version = "0.14.24", optional = true, features = ["server", "http1", "tcp", "stream"] }
+hyper = { version = "1.0.1", optional = true, features = ["server", "http1"] }
 futures = { version = "0.3.27", default-features = false, features = ["std"] }
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.16", features = ["time"] }
-console-subscriber = {  version = "0.1.8", optional = true }
+console-subscriber = { version = "0.2.0", optional = true }
 once_cell = "1.17.1"
 eyre = "0.6.8"
 color-eyre = "0.6.2"
 tracing-error = "0.2.0"
-zerocopy = "0.6.1"
+zerocopy = { version = "0.7.26", features = ["derive"] }
 tokio-stream = { version = "0.1.14", features = ["sync"] }
-minify-html = "0.11.1"
+flate2 = { version = "1.0.26", optional = true }
+bytes = "1.4.0"
+smallvec = { version = "1.10.0", features = ["serde", "const_generics"] }
+http-body-util = "0.1.0"
+pin-project = "1.1.3"
+async-stream = "0.3.5"
 
 [build-dependencies]
 minify-js = { version = "0.5.6", optional = true }
 minify-html = { version = "0.11.1", optional = true }
 css-minify = { version = "0.3.1", optional = true }
+flate2 = { version = "1.0.26", optional = true }
 
 [features]
 default = ["debug_server"]
-debug_server = ["dep:hyper", "minify-html", "dep:minify-js", "dep:css-minify"]
+debug_server = [
+    "dep:hyper",
+    "dep:minify-html",
+    "dep:minify-js",
+    "dep:css-minify",
+    "dep:flate2",
+]
 tokio_console = ["dep:console-subscriber"]

build.rs

@@ -2,13 +2,15 @@ fn main() {
     #[cfg(feature = "debug_server")]
     pack_debug_page().unwrap();
 
-    println!("cargo:rerun-if-changed=main.js");
-    println!("cargo:rerun-if-changed=index.html");
-    println!("cargo:rerun-if-changed=main.css");
+    println!("cargo:rerun-if-changed=web/main.js");
+    println!("cargo:rerun-if-changed=web/index.html");
+    println!("cargo:rerun-if-changed=web/main.css");
+    println!("cargo:rerun-if-changed=web/connected.svg");
 }
 
 #[cfg(feature = "debug_server")]
 fn pack_debug_page() -> Result<(), Box<dyn std::error::Error>> {
+    use flate2::{write::GzEncoder, Compression};
     use std::io::Write;
 
     use css_minify::optimizations::{Level, Minifier};
@@ -16,6 +18,7 @@ fn pack_debug_page() -> Result<(), Box<dyn std::error::Error>> {
     let js = std::fs::read_to_string("web/main.js").unwrap();
     let html = std::fs::read_to_string("web/index.html").unwrap();
     let css = std::fs::read_to_string("web/main.css").unwrap();
+    let svg = std::fs::read_to_string("web/connected.svg").unwrap();
 
     let mut out = Vec::new();
     minify_js::minify(
@@ -27,19 +30,36 @@ fn pack_debug_page() -> Result<(), Box<dyn std::error::Error>> {
     .unwrap();
     let js = std::str::from_utf8(&out)?;
 
-    let css = Minifier::default().minify(&css, Level::Three).unwrap();
+    let css = Minifier::default().minify(&css, Level::One).unwrap();
+
+    let (start, end) = html
+        .split_once("<!--INSERT SVG HERE-->")
+        .expect("did not find svg split point in html");
+
+    let html = format!("{start}{svg}{end}");
 
     let (head, body) = html
         .split_once("<!--INSERT HEAD CONTENT HERE-->")
-        .expect("did not find split point in html");
+        .expect("did not find head split point in html");
+
+    let (body_a, body_b) = body
+        .split_once("#INSERT VERSION HERE#")
+        .expect("did not find version split point in html");
+
+    let version = env!("CARGO_PKG_VERSION");
 
     let html = minify_html::minify(
-        format!("{head}<style>{css}</style><script>{js}</script>{body}").as_bytes(),
+        format!("{head}<style>{css}</style><script>{js}</script>{body_a}{version}{body_b}")
+            .as_bytes(),
         &minify_html::Cfg::spec_compliant(),
     );
 
-    std::fs::File::create(std::env::var("OUT_DIR").unwrap() + "/minified.html")?
-        .write_all(&html)?;
+    let mut encoder = GzEncoder::new(
+        std::fs::File::create(std::env::var("OUT_DIR").unwrap() + "/minified.html.gz")?,
+        Compression::best(),
+    );
+
+    encoder.write_all(&html)?;
 
     Ok(())
 }

src/auth.rs

@@ -34,7 +34,7 @@ pub async fn dyn_ip_update(
             kind: PacketKind::DynIpUpdate.raw(),
             length: 8,
         },
-        data: Vec::new(),
+        ..Default::default()
     };
 
     packet.data.resize(packet.header.length as usize, 0);
@@ -55,11 +55,11 @@ pub async fn dyn_ip_update(
     packet.recv_into(&mut reader).await?;
 
     let result = match packet.kind() {
-        PacketKind::DynIpUpdateResponse => Ok(<[u8; 4]>::try_from(packet.data)
-            .map_err(|err| {
+        PacketKind::DynIpUpdateResponse => Ok(<[u8; 4]>::try_from(packet.data.as_slice())
+            .map_err(|_| {
                 eyre!(
                     "too little data for ip address. Need 4 bytes got {}",
-                    err.len()
+                    packet.data.len()
                 )
             })?
             .into()),

src/client.rs

@@ -10,17 +10,25 @@ use tokio::{
     sync::Mutex,
     time::{sleep, timeout},
 };
-use tracing::{info, instrument, trace};
+use tracing::{info, instrument, trace, warn};
 
 use crate::{
     auth::dyn_ip_update,
     constants::{AUTH_TIMEOUT, CALL_ACK_TIMEOUT, CALL_TIMEOUT, PING_TIMEOUT, SEND_PING_INTERVAL},
     http::peer_query,
-    packets::{Header, Packet, PacketKind, RemConnect, REJECT_OOP, REJECT_TIMEOUT},
+    packets::{
+        Header, Packet, PacketKind, RemConnect, REJECT_OOP, REJECT_TIMEOUT, REJECT_UNKNOWN_CLIENT,
+    },
     ports::{PortHandler, PortStatus},
     Config, HandlerMetadata,
 };
 
+pub enum AuthResult {
+    OutOfPorts,
+    UnknownClient,
+    Success { port: u16 },
+}
+
 /// # Errors
 /// - the client authentication fails
 #[instrument(skip(config, port_handler, handler_metadata))]
@@ -30,7 +38,16 @@ async fn authenticate(
     handler_metadata: &mut HandlerMetadata,
     number: u32,
     pin: u16,
-) -> eyre::Result<Option<u16>> {
+) -> eyre::Result<AuthResult> {
+    // TODO: do we want to cache this? If so, for how long?
+    let name = peer_query(&config.dyn_ip_server, number).await?;
+
+    let Some(name) = name else {
+        return Ok(AuthResult::UnknownClient);
+    };
+
+    info!(%name, "found client");
+
     let mut authenticated = false;
     loop {
         let mut updated_server = false;
@@ -41,9 +58,11 @@ async fn authenticate(
             .allocate_port_for_number(config, number);
 
         let Some(port) = port else {
-            return Ok(None);
+            return Ok(AuthResult::OutOfPorts);
         };
 
+        handler_metadata.port = Some(port);
+
         // make sure the client is authenticated before opening any ports
         if !authenticated {
             let _ip = dyn_ip_update(&config.dyn_ip_server, number, pin, port).await?;
@@ -71,27 +90,17 @@ async fn authenticate(
                 let _ip = dyn_ip_update(&config.dyn_ip_server, number, pin, port).await?;
             }
 
-            #[cfg(feature = "debug_server")]
-            let name = peer_query(&config.dyn_ip_server, number).await?;
-
             let mut port_handler = port_handler.lock().await;
-
-            #[cfg(feature = "debug_server")]
-            if let Some(name) = name {
-                info!(%name, "found client name");
-                port_handler.names.insert(number, name);
-            }
-
             port_handler.register_update();
+
+            port_handler.names.insert(number, name);
             port_handler
                 .port_state
                 .entry(port)
                 .or_default()
                 .new_state(PortStatus::Idle);
 
-            handler_metadata.port = Some(port);
-
-            break Ok(Some(port));
+            break Ok(AuthResult::Success { port });
         }
 
         port_handler.lock().await.mark_port_error(number, port);
@@ -152,12 +161,12 @@ async fn idle(
                     break Ok(Some(IdleResult::Disconnect { packet }))
                 }
             },
-            _ = sleep(send_next_ping_in) => {
+            () = sleep(send_next_ping_in) => {
                 trace!("sending ping");
                 writer.write_all(bytemuck::bytes_of(& Header { kind: PacketKind::Ping.raw(), length: 0 })).await?;
                 last_ping_sent_at = Instant::now();
             }
-            _ = sleep(next_ping_expected_in) => {
+            () = sleep(next_ping_expected_in) => {
 
                 writer.write_all(REJECT_TIMEOUT).await?;
                 break Ok(None);
@@ -323,19 +332,33 @@ pub async fn handler(
 
     let mut packet = Packet::default();
 
-    let Ok(res) = timeout(AUTH_TIMEOUT, packet.recv_into_cancelation_safe(&mut reader)).await else {
+    let Ok(res) = timeout(AUTH_TIMEOUT, packet.recv_into_cancelation_safe(&mut reader)).await
+    else {
         writer.write_all(REJECT_TIMEOUT).await?;
         return Ok(());
     };
     res?;
 
+    if packet.kind() != PacketKind::RemConnect {
+        let kind = packet.kind();
+        warn!(%addr, ?kind, "client sent unexpected packet instead of RemConnect");
+        return Ok(());
+    }
+
     let RemConnect { number, pin } = packet.as_rem_connect()?;
 
     handler_metadata.number = Some(number);
 
-    let Some(port) = authenticate(config, port_handler, handler_metadata, number, pin).await? else {
+    let port = match authenticate(config, port_handler, handler_metadata, number, pin).await? {
+        AuthResult::OutOfPorts => {
             writer.write_all(REJECT_OOP).await?;
             return Ok(());
+        }
+        AuthResult::UnknownClient => {
+            writer.write_all(REJECT_UNKNOWN_CLIENT).await?;
+            return Ok(());
+        }
+        AuthResult::Success { port } => port,
     };
 
     info!(%addr, number, port, "authenticated");
@@ -351,16 +374,19 @@ pub async fn handler(
     packet.data.clear();
     packet.send(&mut writer).await?;
 
-    let Some(idle_result) = idle(
-        listener,
-        packet,
-        &mut reader,
-        &mut writer,
-    ).await? else {
+    let Some(idle_result) = idle(listener, packet, &mut reader, &mut writer).await? else {
         return Ok(());
     };
 
-    let Some((mut caller, mut packet)) = notify_or_disconnect(idle_result, handler_metadata, port_handler, port, &mut writer).await? else {
+    let Some((mut caller, mut packet)) = notify_or_disconnect(
+        idle_result,
+        handler_metadata,
+        port_handler,
+        port,
+        &mut writer,
+    )
+    .await?
+    else {
         return Ok(());
     };
 

src/http.rs

@@ -1,27 +1,114 @@
+use bytes::{Bytes, BytesMut};
 use futures::Future;
-use tokio_stream::StreamExt;
-
+use http_body_util::combinators::UnsyncBoxBody;
+use http_body_util::BodyExt;
+use http_body_util::StreamBody;
+use hyper::body::{Body, Frame};
+use hyper::header::{ACCEPT_ENCODING, CACHE_CONTROL, CONTENT_ENCODING, CONTENT_TYPE};
 use hyper::rt::Executor;
-use hyper::service::{make_service_fn, service_fn};
-use hyper::{Body, Method, Response, Server, StatusCode};
-use std::convert::Infallible;
+use hyper::service::service_fn;
+use hyper::{Method, Request, Response, StatusCode};
+use std::io::Read;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::net::TcpListener;
 use tokio::sync::Mutex;
 use tokio_stream::wrappers::{IntervalStream, WatchStream};
+use tokio_stream::StreamExt;
 use tracing::error;
-
-use zerocopy::{AsBytes, FromBytes, LittleEndian, Unaligned};
-
 use tracing::{debug, instrument};
+use zerocopy::{AsBytes, FromBytes, FromZeroes, LittleEndian, Unaligned};
 
 use crate::constants::DEBUG_SERVER_PING_INTERVAL;
 use crate::packets::{Header, Packet};
-
 use crate::ports::PortHandler;
 use crate::spawn;
 
+mod tokio_io {
+    use std::{
+        pin::Pin,
+        task::{Context, Poll},
+    };
+
+    #[derive(Debug)]
+    #[pin_project::pin_project]
+    pub struct TokioIo<T> {
+        #[pin]
+        inner: T,
+    }
+
+    impl<T> TokioIo<T> {
+        pub fn new(inner: T) -> Self {
+            Self { inner }
+        }
+    }
+
+    impl<T> hyper::rt::Read for TokioIo<T>
+    where
+        T: tokio::io::AsyncRead,
+    {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            mut buf: hyper::rt::ReadBufCursor<'_>,
+        ) -> Poll<Result<(), std::io::Error>> {
+            let n = unsafe {
+                let mut temp_buf = tokio::io::ReadBuf::uninit(buf.as_mut());
+                match tokio::io::AsyncRead::poll_read(self.project().inner, cx, &mut temp_buf) {
+                    Poll::Ready(Ok(())) => temp_buf.filled().len(),
+                    other => return other,
+                }
+            };
+
+            unsafe {
+                buf.advance(n);
+            }
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    impl<T> hyper::rt::Write for TokioIo<T>
+    where
+        T: tokio::io::AsyncWrite,
+    {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            buf: &[u8],
+        ) -> Poll<Result<usize, std::io::Error>> {
+            tokio::io::AsyncWrite::poll_write(self.project().inner, cx, buf)
+        }
+
+        fn poll_flush(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<(), std::io::Error>> {
+            tokio::io::AsyncWrite::poll_flush(self.project().inner, cx)
+        }
+
+        fn poll_shutdown(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<(), std::io::Error>> {
+            tokio::io::AsyncWrite::poll_shutdown(self.project().inner, cx)
+        }
+
+        fn is_write_vectored(&self) -> bool {
+            tokio::io::AsyncWrite::is_write_vectored(&self.inner)
+        }
+
+        fn poll_write_vectored(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+            bufs: &[std::io::IoSlice<'_>],
+        ) -> Poll<Result<usize, std::io::Error>> {
+            tokio::io::AsyncWrite::poll_write_vectored(self.project().inner, cx, bufs)
+        }
+    }
+}
+use self::tokio_io::TokioIo;
+
 #[derive(Clone)]
 struct NamedExecutor;
 impl<T: Send + 'static, Fut: Future<Output = T> + Send + 'static> Executor<Fut> for NamedExecutor {
@@ -30,52 +117,123 @@ impl<T: Send + 'static, Fut: Future<Output = T> + Send + 'static> Executor<Fut>
     }
 }
 
-pub async fn debug_server(
-    addr: SocketAddr,
-    port_handler: Arc<Mutex<PortHandler>>,
-    change_receiver: tokio::sync::watch::Receiver<std::time::Instant>,
-) {
-    let server = Server::bind(&addr)
-        .executor(NamedExecutor)
-        .serve(make_service_fn(move |_conn| {
-            let port_handler = port_handler.clone();
-            let change_receiver = change_receiver.clone();
-            async move {
-                Ok::<_, Infallible>(service_fn(move |req| {
-                    let port_handler = port_handler.clone();
-                    let change_receiver = WatchStream::new(change_receiver.clone());
-                    async move {
-                        match (req.method(), req.uri().path()) {
-                            (&Method::GET, "/") => Ok(Response::new(Body::from(include_str!(
-                                concat!(env!("OUT_DIR"), "/minified.html")
-                            )))),
+const COMPRESSED_HTML: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/minified.html.gz"));
 
-                            (&Method::GET, "/data") => {
-                                let res = Response::builder().header("Cache-Control", "no-store");
+type ResponseBody = UnsyncBoxBody<Bytes, hyper::Error>;
+type RouteResponse = Result<Response<ResponseBody>, hyper::http::Error>;
+
+fn full<D, B>(buf: B) -> ResponseBody
+where
+    http_body_util::Full<D>: From<B> + BodyExt + Body<Data = Bytes>,
+    D: Send + 'static,
+{
+    http_body_util::Full::from(buf)
+        .map_err(|_| unreachable!())
+        .boxed_unsync()
+}
+
+fn empty() -> ResponseBody {
+    http_body_util::Empty::new()
+        .map_err(|_| unreachable!())
+        .boxed_unsync()
+}
+
+fn index(req: &Request<hyper::body::Incoming>) -> RouteResponse {
+    let response = Response::builder();
+
+    let accepts_gzip = req
+        .headers()
+        .get(ACCEPT_ENCODING)
+        .map_or(false, |accept_encoding| {
+            accept_encoding
+                .as_bytes()
+                .split(|x| *x == b',')
+                .filter_map(|x| x.split(|x| *x == b';').next())
+                .filter_map(|x| std::str::from_utf8(x).ok())
+                .any(|x| x.trim() == "gzip")
+        });
+
+    if accepts_gzip {
+        response
+            .header(CONTENT_ENCODING, "gzip")
+            .header(CONTENT_TYPE, "text/html")
+            .body(full(COMPRESSED_HTML))
+    } else {
+        let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
+
+        spawn("gunzip task", async move {
+            let mut decoder =
+                flate2::bufread::GzDecoder::new(std::io::Cursor::new(COMPRESSED_HTML));
+
+            let mut done = false;
+            while !done {
+                let mut chunk = BytesMut::zeroed(256);
+
+                let mut i = 0;
+
+                loop {
+                    let dst = &mut chunk.as_bytes_mut()[i..];
+
+                    if dst.is_empty() {
+                        break; // we are done
+                    }
+
+                    match decoder.read(dst) {
+                        Ok(n) => {
+                            if n == 0 {
+                                done = true;
+                                break;
+                            }
+
+                            i += n;
+                        }
+                        Err(err) => unreachable!("failed to read from gzip decode: {err}"),
+                    }
+                }
+
+                chunk.truncate(i);
+
+                if sender.send(chunk.freeze()).await.is_err() {
+                    break;
+                }
+            }
+        });
+
+        response.body(
+            StreamBody::new(async_stream::stream! {
+                while let Some(item) = receiver.recv().await {
+                    yield Ok(Frame::data(item));
+                }
+            })
+            .boxed_unsync(),
+        )
+    }
+}
+
+async fn data(
+    _req: &Request<hyper::body::Incoming>,
+    port_handler: &Mutex<PortHandler>,
+) -> RouteResponse {
+    let res = Response::builder().header(CACHE_CONTROL, "no-store");
 
     match serde_json::to_string(&*port_handler.lock().await) {
         Ok(data) => res
-                                        .header("Content-Type", "application/json")
-                                        .body(Body::from(data)),
+            .header(CONTENT_TYPE, "application/json")
+            .body(full(data)),
         Err(err) => {
             error!(%err, "failed to serialize data for debug server");
-                                        res.status(StatusCode::INTERNAL_SERVER_ERROR)
-                                            .body(Body::from(""))
+            res.status(StatusCode::INTERNAL_SERVER_ERROR).body(empty())
         }
     }
 }
-
-                            (&Method::GET, "/events") => Response::builder()
-                                .status(StatusCode::OK)
-                                .header("Cache-Control", "no-store")
-                                .header("Content-Type", "text/event-stream")
-                                .body(Body::wrap_stream({
-                                    change_receiver
+fn events(
+    _req: &Request<hyper::body::Incoming>,
+    change_receiver: tokio::sync::watch::Receiver<std::time::Instant>,
+) -> RouteResponse {
+    let stream = WatchStream::new(change_receiver)
         .map(|x| ("change", x))
         .merge(
-                                            IntervalStream::new(tokio::time::interval(
-                                                DEBUG_SERVER_PING_INTERVAL,
-                                            ))
+            IntervalStream::new(tokio::time::interval(DEBUG_SERVER_PING_INTERVAL))
                 .map(|x| ("ping", x.into_std())),
         )
         .filter_map(|(kind, time)| {
@@ -84,22 +242,79 @@ pub async fn debug_server(
                 .ok()?
                 .as_secs();
 
-                                            Some(Ok::<_, Infallible>(format!(
+            Some(Ok::<Frame<Bytes>, _>(Frame::data(Bytes::from(format!(
                 "event: {kind}\ndata: {timestamp}\n\n"
-                                            )))
-                                        })
-                                })),
-                            _ => Response::builder()
-                                .status(StatusCode::NOT_FOUND)
-                                .body(Body::empty()),
-                        }
-                    }
-                }))
-            }
-        }));
+            )))))
+        });
 
-    if let Err(error) = server.await {
-        error!(%error, "debug server error");
+    Response::builder()
+        .status(StatusCode::OK)
+        .header(CACHE_CONTROL, "no-store")
+        .header(CONTENT_TYPE, "text/event-stream")
+        .body(StreamBody::new(stream).boxed_unsync())
+}
+
+async fn debug_server_routes(
+    req: Request<hyper::body::Incoming>,
+    port_handler: Arc<Mutex<PortHandler>>,
+    change_receiver: tokio::sync::watch::Receiver<std::time::Instant>,
+) -> RouteResponse {
+    match (req.method(), req.uri().path()) {
+        (&Method::GET, "/") => index(&req),
+        (&Method::GET, "/data") => Ok(data(&req, &port_handler).await?),
+        (&Method::GET, "/events") => events(&req, change_receiver),
+        _ => Ok(Response::builder()
+            .status(StatusCode::NOT_FOUND)
+            .body(empty())?),
+    }
+}
+
+pub async fn debug_server(
+    addr: SocketAddr,
+    port_handler: Arc<Mutex<PortHandler>>,
+    change_receiver: tokio::sync::watch::Receiver<std::time::Instant>,
+) {
+    let listener = TcpListener::bind(addr).await;
+
+    let listener = match listener {
+        Ok(listener) => listener,
+        Err(error) => {
+            error!(%error, %addr, "failed to bind debug server");
+            return;
+        }
+    };
+
+    loop {
+        let conn = listener.accept().await;
+
+        let stream = match conn {
+            Ok((stream, _)) => stream,
+            Err(error) => {
+                error!(%error, "failed accept debug server connection");
+                continue;
+            }
+        };
+
+        let io = TokioIo::new(stream);
+
+        let port_handler = port_handler.clone();
+        let change_receiver = change_receiver.clone();
+
+        tokio::task::spawn(async move {
+            if let Err(error) = hyper::server::conn::http1::Builder::new()
+                .serve_connection(
+                    io,
+                    service_fn(move |req| {
+                        debug_server_routes(req, port_handler.clone(), change_receiver.clone())
+                    }),
+                )
+                .await
+            {
+                if !error.is_incomplete_message() {
+                    error!(%error, "Failed to serve debug server connection");
+                }
+            }
+        });
     }
 }
 
@@ -113,7 +328,7 @@ struct PeerQuery {
     number: U32,
 }
 
-#[derive(FromBytes, Unaligned, Debug)]
+#[derive(FromBytes, FromZeroes, Unaligned, Debug)]
 #[repr(packed)]
 #[allow(dead_code)]
 struct PeerReply {
@@ -138,7 +353,7 @@ pub async fn peer_query(server: &SocketAddr, number: u32) -> eyre::Result<Option
             kind: 3, // Peer Query
             length: 4,
         },
-        data: Vec::new(),
+        ..Default::default()
     };
 
     packet.data.clear();

src/main.rs

@@ -14,6 +14,7 @@ use futures::Future;
 use http::debug_server;
 use packets::{Header, Packet};
 use serde::{Deserialize, Deserializer};
+use smallvec::SmallVec;
 use time::format_description::OwnedFormatItem;
 use tokio::{
     io::AsyncWriteExt,
@@ -255,7 +256,10 @@ async fn connection_handler(
             }
             _ => Some(err.to_string()),
         },
-        Ok(Ok(())) => None,
+        Ok(Ok(())) => {
+            debug!(%addr, "finished handling client");
+            None
+        }
     };
 
     if let Some(error) = error {
@@ -280,10 +284,18 @@ async fn connection_handler(
         let mut port_handler = port_handler.lock().await;
 
         if let Some(port_state) = port_handler.port_state.get_mut(&port) {
+            // the client is known. Mark is as disconnected
             port_state.new_state(PortStatus::Disconnected);
-            port_handler.register_update();
+        } else {
+            // the client is not known. Free its port for realloction
+            assert!(
+                port_handler.free_ports.insert(port),
+                "tried to free up a port that was not allocted"
+            );
         }
 
+        port_handler.register_update();
+
         if let Some(listener) = handler_metadata.listener.take() {
             port_handler.start_rejector(
                 port,
@@ -293,13 +305,13 @@ async fn connection_handler(
                         kind: PacketKind::Reject.raw(),
                         length: 3,
                     },
-                    data: b"nc\0".to_vec(),
+                    data: SmallVec::from_slice(b"nc\0"),
                 },
             );
         }
     }
 
-    sleep(Duration::from_secs(3)).await;
+    sleep(Duration::from_secs(2)).await;
     _ = stream.shutdown().await;
 }
 
@@ -333,7 +345,7 @@ async fn tokio_main(config: Arc<Config>) -> eyre::Result<()> {
     let (change_sender, change_receiver) = tokio::sync::watch::channel(std::time::Instant::now());
 
     let mut port_handler = PortHandler::load_or_default(&cache_path, change_sender);
-    port_handler.update_allowed_ports(&config.allowed_ports);
+    port_handler.update_allowed_ports(config.allowed_ports.clone());
 
     let port_handler = Arc::new(Mutex::new(port_handler));
 

src/packets.rs

@@ -1,8 +1,8 @@
-use std::fmt::Debug;
+use std::fmt::{Debug, Display};
 
 use bytemuck::{Pod, Zeroable};
-use eyre::eyre;
 use serde::Serialize;
+use smallvec::SmallVec;
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::tcp::{ReadHalf, WriteHalf},
@@ -10,6 +10,48 @@ use tokio::{
 
 pub const REJECT_OOP: &[u8; 6] = b"\x04\x04oop\x00";
 pub const REJECT_TIMEOUT: &[u8; 10] = b"\x04\x08timeout\x00";
+pub const REJECT_UNKNOWN_CLIENT: &[u8; 17] = b"\x04\x0funknown client\x00";
+
+#[derive(Debug)]
+pub enum Error {
+    TooLittleData {
+        expected: u8,
+        got: u8,
+    },
+    Unexpected {
+        expected: PacketKind,
+        got: PacketKind,
+    },
+    Io(std::io::Error),
+    Client(Option<String>),
+}
+
+impl std::error::Error for Error {}
+impl Display for Error {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Error::TooLittleData { expected, got } => write!(
+                f,
+                "received too little data: expected {expected} bytes but received only {got}"
+            ),
+            Error::Unexpected { expected, got } => write!(
+                f,
+                "received an unexpected packet: expected {expected:?} but received {got:?}",
+            ),
+            Error::Io(inner) => Display::fmt(inner, f),
+            Error::Client(Some(inner)) => write!(f, "client reported an error: {inner}"),
+            Error::Client(None) => write!(f, "client reported a malformed error",),
+        }
+    }
+}
+
+impl From<std::io::Error> for Error {
+    fn from(value: std::io::Error) -> Self {
+        Self::Io(value)
+    }
+}
+
+type Result<T> = std::result::Result<T, Error>;
 
 #[repr(u8)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -79,7 +121,7 @@ pub struct Header {
 #[derive(Serialize, Default, Clone)]
 pub struct Packet {
     pub header: Header,
-    pub data: Vec<u8>,
+    pub data: SmallVec<[u8; 8]>,
 }
 
 impl Packet {
@@ -150,10 +192,7 @@ impl Packet {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn recv_into_cancelation_safe(
-        &mut self,
-        stream: &mut ReadHalf<'_>,
-    ) -> eyre::Result<()> {
+    pub async fn recv_into_cancelation_safe(&mut self, stream: &mut ReadHalf<'_>) -> Result<()> {
         // Makes sure all data is available before reading
         let header_bytes = bytemuck::bytes_of_mut(&mut self.header);
         stream.peek(header_bytes).await?;
@@ -165,7 +204,7 @@ impl Packet {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn recv_into(&mut self, stream: &mut ReadHalf<'_>) -> eyre::Result<()> {
+    pub async fn recv_into(&mut self, stream: &mut ReadHalf<'_>) -> Result<()> {
         let header_bytes = bytemuck::bytes_of_mut(&mut self.header);
 
         stream.read_exact(header_bytes).await?;
@@ -175,10 +214,7 @@ impl Packet {
         stream.read_exact(&mut self.data).await?;
 
         if self.header.kind == PacketKind::Error.raw() {
-            return Err(eyre!(
-                "client reported error: {:?}",
-                self.as_string().unwrap_or("unknown dyn auth error")
-            ));
+            return Err(Error::Client(self.as_string().map(ToOwned::to_owned)));
         }
 
         Ok(())
@@ -198,24 +234,26 @@ impl Packet {
 
     /// # Errors
     /// the packet must be a `RemConnect` packet and must contain at least 6 bytes of data
-    pub fn as_rem_connect(&self) -> eyre::Result<RemConnect> {
+    #[allow(clippy::missing_panics_doc)]
+    pub fn as_rem_connect(&self) -> Result<RemConnect> {
         if self.kind() != PacketKind::RemConnect {
-            return Err(eyre!(
-                "Unexpected Packet: {:?} expected RemConnect",
-                self.kind()
-            ));
+            return Err(Error::Unexpected {
+                expected: PacketKind::RemConnect,
+                got: self.kind(),
+            });
         }
 
         if self.data.len() < 6 {
-            return Err(eyre!(
-                "Too little data for RemConnect. Need at least 6 Bytes got {}",
-                self.data.len()
-            ));
+            #[allow(clippy::cast_possible_truncation)]
+            return Err(Error::TooLittleData {
+                expected: 6,
+                got: self.data.len() as u8,
+            });
         }
 
         Ok(RemConnect {
-            number: u32::from_le_bytes(self.data[..4].try_into()?),
-            pin: u16::from_le_bytes(self.data[4..6].try_into()?),
+            number: u32::from_le_bytes(self.data[..4].try_into().unwrap()),
+            pin: u16::from_le_bytes(self.data[4..6].try_into().unwrap()),
         })
     }
 }

src/ports.rs

@@ -1,6 +1,6 @@
 use std::{
     collections::{BTreeSet, HashMap, HashSet},
-    fmt::{Debug, Display},
+    fmt::Debug,
     fs::File,
     io::{BufReader, BufWriter},
     ops::RangeInclusive,
@@ -12,16 +12,17 @@ use std::{
 use eyre::eyre;
 use serde::{Deserialize, Serialize, Serializer};
 use tokio::{
+    io::AsyncWriteExt,
     net::TcpListener,
     sync::{watch::Receiver, Mutex},
     task::JoinHandle,
-    time::Instant,
+    time::{sleep, Instant},
 };
 use tracing::{debug, error, info, instrument, warn};
 
 use crate::{
     constants::{CACHE_STORE_INTERVAL, PORT_OWNERSHIP_TIMEOUT, PORT_RETRY_TIME},
-    packets::{Packet, PacketKind},
+    packets::Packet,
     spawn, Config, Number, Port, UnixTimestamp,
 };
 
@@ -40,7 +41,7 @@ pub struct PortHandler {
     allowed_ports: AllowedList,
 
     #[serde(skip)]
-    free_ports: HashSet<Port>,
+    pub free_ports: HashSet<Port>,
 
     errored_ports: BTreeSet<(UnixTimestamp, Port)>,
     allocated_ports: HashMap<Number, Port>,
@@ -102,33 +103,41 @@ pub async fn cache_daemon(
     }
 }
 
-#[derive(Hash, PartialEq, Eq)]
-struct DisplayAsDebug<T: Display>(T);
-impl<T: Display> Debug for DisplayAsDebug<T> {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.0)
-    }
-}
-
 #[derive(Default, Serialize, Deserialize)]
 pub struct PortState {
+    #[serde(deserialize_with = "deserialize_last_change")]
     last_change: UnixTimestamp,
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
     status: PortStatus,
 }
 
+fn deserialize_last_change<'de, D>(deserializer: D) -> Result<UnixTimestamp, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    Ok(match Option::<UnixTimestamp>::deserialize(deserializer)? {
+        Some(timestamp) => timestamp,
+        None => now(),
+    })
+}
+
+fn now() -> UnixTimestamp {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("timestamp overflow")
+        .as_secs()
+}
+
 impl PortState {
     pub fn new_state(&mut self, status: PortStatus) {
-        self.last_change = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .expect("timestamp overflow")
-            .as_secs();
+        self.last_change = now();
 
         self.status = status;
     }
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, PartialOrd, Ord)]
+#[serde(rename_all = "snake_case")]
 pub enum PortStatus {
     InCall,
     Idle,
@@ -156,6 +165,8 @@ impl AllowedList {
 }
 
 impl PortHandler {
+    /// # Panics
+    /// If the the `change_sender` could not be notified
     pub fn register_update(&mut self) {
         let now = std::time::Instant::now();
         self.last_update = Some(now);
@@ -179,6 +190,24 @@ impl PortHandler {
         value_object.remove("rejectors").unwrap();
         value_object.remove("last_update").unwrap();
 
+        value_object
+            .get_mut("port_state")
+            .unwrap()
+            .as_object_mut()
+            .unwrap()
+            .values_mut()
+            .for_each(|value| {
+                let value_object = value.as_object_mut().unwrap();
+
+                // it does not make sense to store when the did anything else other than disconnect
+                // because when we restart the server it will no longer be connected
+                if value_object.get("status").unwrap().as_str().unwrap() != "disconnected" {
+                    *value_object.get_mut("last_change").unwrap() = serde_json::Value::Null;
+                }
+
+                value_object.remove("status").unwrap();
+            });
+
         serde_json::to_writer(BufWriter::new(File::create(&temp_file)?), &value)?;
         std::fs::rename(temp_file, cache)?;
 
@@ -207,10 +236,10 @@ impl PortHandler {
         this
     }
 
-    pub fn update_allowed_ports(&mut self, allowed_ports: &AllowedList) {
+    pub fn update_allowed_ports(&mut self, allowed_ports: AllowedList) {
         self.register_update();
 
-        self.allowed_ports = allowed_ports.clone();
+        self.allowed_ports = allowed_ports;
 
         self.free_ports.clear(); // remove all ports
         self.free_ports
@@ -291,17 +320,10 @@ impl Serialize for Rejector {
         S: Serializer,
     {
         let packet = &self.state.1;
-        let kind = match packet.kind() {
-            PacketKind::End => "end",
-            PacketKind::Reject => "reject",
-            _ => unreachable!(),
-        };
 
         match packet.as_string() {
-            Some(string) if string.chars().all(|c| !c.is_control()) => {
-                (kind, string).serialize(serializer)
-            }
-            _ => (kind, packet.data()).serialize(serializer),
+            Some(string) if string.chars().all(|c| !c.is_control()) => string.serialize(serializer),
+            _ => packet.data().serialize(serializer),
         }
     }
 }
@@ -310,7 +332,7 @@ impl Debug for Rejector {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Rejector")
             .field("message", &self.state.1)
-            .finish()
+            .finish_non_exhaustive()
     }
 }
 
@@ -326,12 +348,22 @@ impl Rejector {
             spawn(&format!("rejector for port {port}",), async move {
                 let (listener, packet) = state.as_ref();
 
+                let packet = Arc::new(packet.clone());
+
                 let listener = listener.lock().await;
 
                 loop {
                     if let Ok((mut socket, _)) = listener.accept().await {
+                        let packet = packet.clone();
+                        tokio::task::spawn(async move {
                             let (_, mut writer) = socket.split();
                             _ = packet.send(&mut writer).await;
+
+                            // wait two seconds before closing the connection
+                            // to accomodate bad itelex tpc stacks
+                            sleep(Duration::from_secs(2)).await;
+                            _ = socket.shutdown().await;
+                        });
                     }
                 }
             })
@@ -445,7 +477,7 @@ impl PortHandler {
             return Some(port);
         }
 
-        None // TODO
+        None // TODO: are there more ways?
     }
 
     #[instrument(skip(self))]

web/connected.svg

@@ -0,0 +1 @@
+<svg version="1.1" viewBox="0 0 60.59 60.734" xmlns="http://www.w3.org/2000/svg"><g transform="translate(-38.21,-51.32)"><path d="m98.8 55.1-6.52 6.52-3.78-3.78 6.52-6.52zm-7.49 20.5-8.5 8.5-16.8-16.8 8.5-8.5zm-16.9 0.0741a11.9 11.9 45 0 1 0.0256-16.8 11.9 11.9 45 0 1 16.8-0.0256 11.9 11.9 45 0 1-0.0256 16.8 11.9 11.9 45 0 1-16.8 0.0256zm-36.2 32.6 6.52-6.52 3.78 3.78-6.52 6.52zm19.7-25.3 8.39-8.39 1.88 1.88-8.39 8.39zm7.46 7.5 8.39-8.39 1.88 1.88-8.39 8.39zm-19.7-2.72 8.5-8.5 16.8 16.8-8.5 8.5zm16.9-0.0741a11.9 11.9 45 0 1-0.0256 16.8 11.9 11.9 45 0 1-16.8 0.0256 11.9 11.9 45 0 1 0.0256-16.8 11.9 11.9 45 0 1 16.8-0.0256z"/></g></svg>

web/index.html

@@ -2,6 +2,7 @@
 <html>
 
 <head>
+    <link rel="icon" href="data:,">
     <meta charset='utf-8'>
     <meta name='viewport' content='width=device-width, initial-scale=1'>
     <title>Centralex State</title>
@@ -9,10 +10,24 @@
 </head>
 
 <body>
+    <div class="content">
 
-    <p id="last_update" />
-    <pre id="data"></pre>
-    <list id="list" />
+        <div id="header">
+            <p id="free_ports" />
+            <div id="connected" class="hidden"><!--INSERT SVG HERE--></div>
+            <p id="last_ping" title="v#INSERT VERSION HERE#" />
+        </div>
+
+        <table id="table">
+            <tr>
+                <th onclick="sort(this,'name')">Name</th>
+                <th onclick="sort(this,'number')">Nummer</th>
+                <th onclick="sort(this,'port')">Port</th>
+                <th onclick="sort(this,'status')">Zustand</th>
+                <th onclick="sort(this,'last_change')">Seit</th>
+            </tr>
+        </table>
+    </div>
 </body>
 
 </html>

web/main.css

@@ -1,3 +1,90 @@
 body {
     background-color: #eee;
 }
+
+.last_change {
+    text-align: right;
+}
+
+.last_change .unit {
+    padding-left: 1ch;
+    min-width: 8ch;
+    display: inline-block;
+    text-align: left;
+}
+
+#last_ping,
+#connected,
+#free_ports {
+    margin-top: 0;
+    margin-bottom: 0.25em;
+}
+
+#connected svg {
+    height: 1em;
+}
+
+#header {
+    display: flex;
+    flex-direction: row;
+    grid-template-columns: auto 1fr;
+    justify-content: space-between;
+}
+
+td,
+th {
+    border: 1px solid black;
+    padding: 0.5em;
+    font-family: monospace;
+}
+
+table {
+    border-spacing: 0;
+    max-width: 100%;
+    margin: 0 auto;
+}
+
+th {
+    cursor: pointer;
+    user-select: none;
+    font-size: inherit;
+}
+
+th::after {
+    font-family: monospace;
+    padding-left: 3px;
+}
+
+th:not(.sort)::after {
+    content: "\a0";
+}
+
+.sort-up::after {
+    content: "\25b2";
+}
+
+.sort-down::after {
+    content: "\25bc";
+}
+
+td {
+    text-align: left;
+}
+
+.port {
+    text-align: left;
+}
+
+.visible {
+    opacity: 0;
+    /*transition: opacity 500ms linear;*/
+}
+
+.hidden {
+    opacity: 1;
+    transition: opacity 5000ms ease-in;
+}
+
+#connected {
+    fill: rgb(224, 32, 6)
+}

web/main.js

@@ -1,28 +1,256 @@
 window.onload = () => {
-    const evtSource = new EventSource("/events");
-    const data = document.getElementById("data");
-    const last_update = document.getElementById("last_update");
-    evtSource.addEventListener("change", event => {
-        console.log(event);
+  const table_elem = document.getElementById("table");
+  const last_change = document.getElementById("last_change");
+  const last_ping = document.getElementById("last_ping");
+  const connected = document.getElementById("connected");
+  const free_ports = document.getElementById("free_ports");
 
-        last_update.innerText = `last update at ${new Date(+event.data * 1000)}`;
+  const timeout_duration = 10 * 1000;
+  const retry_timeout = 5 * 1000;
 
-        const newElement = document.createElement("li");
-        const eventList = document.getElementById("list");
-        newElement.textContent = `change at ${+event.data}`;
-        eventList.appendChild(newElement);
+  let reconnect_timeout;
+  let ping_timeout;
+  let evtSource;
+  let table = [];
 
-        fetch("/data").then(res => res.json().then(res => data.innerText = JSON.stringify(res, null, 1)));
-    });
+  let last_update = new Date();
 
-    evtSource.addEventListener("ping", event => {
-        console.log(event);
+  let direction = -1;
+  let oldkey = "last_change";
 
-        last_update.innerText = `last update at ${new Date(+event.data * 1000)}`;
+  table_elem.firstChild.firstChild.lastChild.className = "sort sort-down";
 
-        const newElement = document.createElement("li");
-        const eventList = document.getElementById("list");
-        newElement.textContent = `ping at ${+event.data}`;
-        eventList.appendChild(newElement);
-    });
+  let time_ago = (ms) => {
+    let value = ms / 1000;
+    // let prev = 0;
+    let unit = 0;
+
+    let factors = [
+      [1, "Sekunde", "n"],
+      [60, "Minute", "n"],
+      [60, "Stunde", "n"],
+      [24, "Tag", "en"],
+      [7, "Woche", "n"],
+      [4.348214, "Monat", "en"],
+      [12, "Jahr", "en"],
+    ];
+
+    for (let i in factors) {
+      let factor = factors[i][0];
+      let new_value = Math.floor(value / factor);
+      if (new_value == 0) break;
+
+      // prev = Math.floor(value % factor);
+      value = new_value;
+      unit = i;
+    }
+
+    let factor = factors[unit];
+
+    return [value, factors[unit][1] + (value == 1 ? "" : factor[2])];
+  };
+
+  sort = (element, key) => {
+    if (key == oldkey) {
+      direction *= -1;
+    } else {
+      oldkey = key;
+      direction = -1;
+    }
+
+    for (let child of table_elem.firstChild.firstChild.children) {
+      child.className = "";
+    }
+
+    element.className = `sort ${direction > 0 ? "sort-up" : "sort-down"}`;
+
+    do_sort(oldkey, direction);
+  };
+
+  let do_sort = (key, direction) => {
+    let is_number = !!~(["port", "number", "last_change"].indexOf(key));
+
+    table = table.sort((a, b) => (direction * (
+      is_number ? a[key] - b[key] : ("" + a[key]).localeCompare(b[key], "de-DE")
+    )));
+
+    populate_table();
+  };
+
+  let fmt = Intl.DateTimeFormat("de-DE", {
+    dateStyle: "medium",
+    timeStyle: "medium",
+  });
+
+  let format_date = (date) => fmt.format(date).replace(", ", " ");
+  let format_time = (date) => fmt.format(date).split(", ", 2)[1];
+
+  let populate_table = () => {
+    // clear everything except for the header row
+    while (table_elem.rows.length > 1) {
+      table_elem.deleteRow(-1);
+    }
+
+    for (let row of table) {
+      let tr = table_elem.insertRow(-1);
+
+      let values = [
+        row.name === null ? "?" : row.name,
+        row.number,
+        row.port,
+        row.status,
+        time_ago(last_update - row.last_change),
+      ];
+
+      let names = [
+        "name",
+        "number",
+        "port",
+        "status",
+        "last_change",
+      ];
+
+      for (let i in values) {
+        let value = values[i];
+        let name = names[i];
+
+        let td = tr.insertCell(-1);
+        td.className = name;
+
+        if (name == "last_change") {
+          let [number, unit] = value;
+
+          let span = document.createElement("span");
+          // span.className = "value";
+          span.innerText = number;
+          td.appendChild(span);
+
+          span = document.createElement("span");
+          span.className = "unit";
+          span.innerText = unit;
+          td.appendChild(span);
+        } else {
+          td.innerText = value;
+        }
+      }
+    }
+  };
+
+  let update_table = (data) => {
+    const allowed_ports = data.allowed_ports.map((x) => x.end - x.start + 1)
+      .reduce((a, b) => a + b, 0);
+    free_ports.innerText = `Freie Ports: ${
+      allowed_ports - Object.keys(data.allocated_ports).length -
+      data.errored_ports.length
+    }`;
+
+    // last_change.innerHTML = `Letzte Änderung: ${format_date(new Date(+data.last_update * 1000))}`;
+
+    table = [];
+
+    for (let number in data.allocated_ports) {
+      try {
+        let port = data.allocated_ports[number];
+        number = +number;
+
+        // allocated port has no state. This means that it is unknown and should not be displayed
+        if (data.port_state[port] == undefined) {
+          continue;
+        }
+
+        let { status, last_change } = data.port_state[port];
+
+        let rejector = data.rejectors[port] || null;
+
+        if (rejector && rejector instanceof Array) {
+          rejector = rejector.map((x) => "0x" + x.toString(16).padStart(2, 0))
+            .join(" ");
+        }
+
+        last_change = new Date(last_change * 1000);
+
+        let name = data.names[number] || null;
+
+        switch (status) {
+          case "disconnected":
+            status = rejector ? `getrennt: ${rejector}` : "getrennt";
+            break;
+          case "idle":
+            status = "bereit";
+            break;
+          case "in_call":
+            status = "anruf";
+            break;
+        }
+
+        table.push({ port, number, status, last_change, rejector, name });
+      } catch (error) {
+        console.error(error);
+      }
+    }
+
+    for (let [timestamp, port] of data.errored_ports) {
+      table.push({
+        port,
+        number: null,
+        status: "Fehler",
+        last_change: new Date(timestamp * 1000),
+        rejector: null,
+        name: "",
+      });
+    }
+
+    do_sort(oldkey, direction);
+
+    populate_table();
+  };
+
+  let format_event = (event, method) =>
+    (method || format_date)(new Date(+event.data * 1000));
+
+  let display_disconnected;
+
+  let connect_event_source = () => {
+    clearTimeout(reconnect_timeout);
+    clearTimeout(ping_timeout);
+    ping_timeout = setTimeout(connect_event_source, timeout_duration);
+
+    evtSource && evtSource.close && evtSource.close();
+
+    evtSource = new EventSource("/events");
+    evtSource.addEventListener("change", (event) => {
+      last_ping.innerText = `Stand: ${format_event(event, format_time)}`;
+
+      last_update = new Date(+event.data * 1000);
+
+      fetch("/data")
+        .then((res) => res.json())
+        .then(update_table);
+    });
+
+    evtSource.addEventListener("ping", (event) => {
+      clearTimeout(ping_timeout);
+      ping_timeout = setTimeout(connect_event_source, timeout_duration);
+
+      last_update = new Date(+event.data * 1000);
+
+      last_ping.innerText = `Stand: ${format_event(event, format_time)}`;
+      connected.className = "visible";
+
+      populate_table();
+
+      clearTimeout(display_disconnected);
+      display_disconnected = setTimeout(
+        () => connected.className = "hidden",
+        5000,
+      );
+    });
+
+    evtSource.onerror = () => {
+      clearTimeout(reconnect_timeout);
+      reconnect_timeout = setTimeout(connect_event_source, retry_timeout);
+    };
+  };
+
+  connect_event_source();
 };