Inital ntp support

Cargo.toml

@@ -9,7 +9,9 @@ exclude = ["/.vscode"]
 
 [dependencies]
 clap = { version = "4.3.0", features = ["derive"] }
+lazy_static = "1.4.0"
 mpvipc = "1.2.2"
+rsntp = "3.0.2"
 tempfile = "3.5.0"
 tokio = { version = "1.28.1", features = ["full"] }
 url = "2.3.1"

src/main.rs

@@ -7,7 +7,9 @@ use client_message_handler::*;
 use mpv_event_handler::*;
 use mpvipc::*;
 use proto::*;
+use rsntp::SntpClient;
 use std::net::SocketAddr;
+use std::time::{SystemTime, UNIX_EPOCH};
 use std::vec;
 use std::{collections::HashMap, process::Command, sync::Arc};
 use tempfile::tempdir;
@@ -36,6 +38,18 @@ struct Cli {
     #[arg(long)]
     standalone: bool,
 
+    /// use system time instead of ntp (not reccomended)
+    #[arg(short, long)]
+    trust_system_time: bool,
+
+    /// address of the ntp server
+    #[arg(
+        long,
+        conflicts_with = "trust_system_time",
+        default_value = "pool.ntp.org"
+    )]
+    ntp_server: String,
+
     /// address:port to connect/bind to  
     #[arg(value_name = "ADDRESS")]
     address: String,
@@ -112,6 +126,20 @@ pub struct Settings {
 async fn main() {
     let args = Cli::parse();
 
+    if !args.trust_system_time {
+        let client = SntpClient::new();
+        let result = client
+            .synchronize(args.ntp_server)
+            .expect("Coudn't syncronize time with ntp server");
+        let delta: i64 = result.datetime().unix_timestamp().expect("msg").as_millis() as i64
+            - SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .expect("Couldn't get system time")
+                .as_millis() as i64;
+        println!("Clock skew: {} ms", delta);
+        TIME_DELTA.store(delta, std::sync::atomic::Ordering::SeqCst);
+    }
+
     let state = Arc::new(Mutex::new(Shared::new()));
 
     let cloned_state = Arc::clone(&state);

src/proto.rs

@@ -1,4 +1,7 @@
+use lazy_static::lazy_static;
 use std::mem::size_of;
+use std::sync::atomic::AtomicI64;
+use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 use std::vec;
 use std::{error::Error, fmt};
@@ -14,9 +17,13 @@ const PROTOCOL_VERSION: u16 = 1;
 // ^               ^          ^             ^                 ^
 // |    8 bytes    |  1 byte  |   2 bytes   |  $lenght bytes  |
 
-type TsSize = u64;
-type CmdSize = u8;
-type LenSize = u16;
+pub type TsSize = u64;
+pub type CmdSize = u8;
+pub type LenSize = u16;
+
+lazy_static! {
+    pub static ref TIME_DELTA: Arc<AtomicI64> = Arc::new(AtomicI64::new(0));
+}
 
 pub struct PacketReader {
     pub inner: OwnedReadHalf,
@@ -191,12 +198,14 @@ impl VoyeursCommand {
     }
 
     pub fn craft_packet(self) -> Packet {
-        let timestamp: TsSize = SystemTime::now()
+        let timestamp = SystemTime::now()
             .duration_since(UNIX_EPOCH)
             .expect("Couldn't get system time")
-            .as_millis() as TsSize;
+            .as_millis() as i64
+            + TIME_DELTA.load(std::sync::atomic::Ordering::SeqCst);
+
         Packet {
-            timestamp,
+            timestamp: timestamp.try_into().unwrap(),
             command: self,
         }
     }