Merge pull request #4 from ryankurte/rebuild-client

rework client + config, add tls support

Author: gwbres

.github/workflows/daily.yml

@@ -22,7 +22,7 @@ jobs:
     - uses: actions-rs/toolchain@v1
       name: Install Rust
       with:
-        toolchain: 1.78.0
+        toolchain: 1.82.0
         override: true
 
     - uses: actions-rs/cargo@v1

.github/workflows/rust.yml

@@ -33,7 +33,7 @@ jobs:
     - uses: actions-rs/toolchain@v1
       name: Install Rust
       with:
-        toolchain: 1.78.0
+        toolchain: 1.82.0
         override: true
 
     - uses: actions-rs/cargo@v1

Cargo.toml

@@ -2,7 +2,10 @@
 name = "ntrip-client"
 version = "0.0.1"
 license = "MPL-2.0"
-authors = ["Guillaume W. Bres <guillaume.bressaix@gmail.com>"]
+authors = [
+    "Guillaume W. Bres <guillaume.bressaix@gmail.com>",
+    "Ryan Kurte <ryan@kurte.nz>"
+]
 description = "NTRIP client"
 homepage = "https://github.com/rtk-rs"
 repository = "https://github.com/rtk-rs/ntrip-client"
@@ -12,21 +15,42 @@ edition = "2021"
 readme = "README.md"
 
 [package.metadata]
-msrv = "1.78"
+msrv = "1.82"
 
 [package.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docrs", "--generate-link-to-definition"]
 
 [features]
-default = []
+default = [ "log", "clap", "serde", "anyhow" ]
 
 # Unlock client logs
-log = ["dep:log"]
+log = ["dep:tracing", "dep:tracing-subscriber"]
+clap = ["dep:clap"]
+serde = ["dep:serde", "rtcm-rs/serde", "geoutils/serde"]
 
 [dependencies]
 thiserror = "2"
 base64 = "0.22"
 rtcm-rs = "0.11"
-log = { version = "0.4", optional = true }
+futures = "0.3"
 tokio = { version = "1.45.0", features = ["full"] }
+strum = { version = "0.27.2", features = ["derive"] }
+reqwest = { version = "0.12", features = ["rustls-tls"] }
+http = "1.3.1"
+tokio-rustls = "0.26.2"
+rustls = "0.23.31"
+webpki-roots = "1.0.2"
+geoutils = "0.5.1"
+isocountry = "0.3.2"
+
+tracing = { version = "0.1.41", optional = true, features = ["log"] }
+tracing-subscriber = { version = "0.3.17", optional = true, features = ["fmt", "env-filter"] }
+clap = { version = "4.5", optional = true, features = ["derive", "env"] }
+serde = { version = "1.0", optional = true, features = ["derive"] }
+anyhow = { version = "1.0.0", optional = true }
+
+[[examples]]
+name = "simple-cli"
+path = "examples/simple-cli.rs"
+required-features = ["clap", "log", "anyhow"]

README.md

@@ -23,16 +23,46 @@ Getting started
 ntrip-client = "0.0.1"
 ```
 
+Refer to the [provided example](examples/) for a complete demo.
+
 ```rust
-let mut client = NTRIPClient::new("caster.centipede.fr", 2101, "ENSMM")
-    .with_credentials("centipede", "centipede");
-
-// deploy using 'tokio' framework
-client.run()
-    .await
-    .unwrap_or_else(|e| {
-        panic!("Failed to deploy NTRIP client: {}", e);
-    });
+// Configure server
+let ntrip_config = "centipede".parse::<NtripConfig>();
+let ntrip_creds = NtripCredentials{
+    user: "centipede".to_string(),
+    pass: "centipede".to_string(),
+}
+
+// Setup client
+let mut client = NtripClient::new(ntrip_config, ntrip_creds).await.unwrap();
+
+// List mounts
+let server_info = client.list_mounts().await.unwrap();
+for m in server_info.mounts {
+    println!("{} - {}", m.name, m.details);
+}
+
+// Subscribe to a mount
+let (exit_tx, exit_rx) = tokio::sync::broadcast(1);
+let handle = client.mount("VALDM", exit_tx.clone());
+
+loop {
+    select!{
+        m = client.next() => match m {
+            Some(m) => {
+                info!("Received RTCM message: {:?}", m);
+            },
+            None => {
+                error!("NTRIP client stream ended");
+                break;
+            }
+        },
+        _ = exit_rx.recv() => {
+            info!("Exiting on signal");
+            break;
+        }
+    }
+}
 ```
 
 Licensing

examples/simple-cli.rs

@@ -0,0 +1,149 @@
+use clap::Parser;
+use futures::StreamExt;
+use geoutils::Location;
+use ntrip_client::{
+    config::{NtripConfig, NtripCredentials},
+    NtripClient,
+};
+use tokio::select;
+use tracing::{debug, error, info, level_filters::LevelFilter};
+use tracing_subscriber::{fmt::Subscriber as FmtSubscriber, EnvFilter};
+
+/// NTRIP command line tool
+#[derive(Clone, PartialEq, Debug, Parser)]
+struct Args {
+    #[clap()]
+    /// NTRIP server identifier or URI ("rtk2go", "linz" etc., or "[ntrip|http|https]://host:port")
+    pub ntrip_host: NtripConfig,
+
+    #[clap(flatten)]
+    pub ntrip_creds: NtripCredentials,
+
+    #[clap(subcommand)]
+    pub command: Commands,
+
+    #[clap(long, default_value = "info")]
+    /// Set log level
+    pub log_level: LevelFilter,
+}
+
+#[derive(Clone, PartialEq, Debug, Parser)]
+pub enum Commands {
+    /// List mount points on an NTRIP server
+    List,
+    /// Find the nearest mount point to a specified location
+    FindNearest {
+        #[clap()]
+        lat: f64,
+        #[clap()]
+        lon: f64,
+    },
+    /// Subscribe to a specified mount point and print received RTCM messages
+    Subscribe {
+        #[clap()]
+        mount: String,
+    },
+}
+
+#[tokio::main]
+async fn main() -> Result<(), anyhow::Error> {
+    // Parse command line arguments
+    let args = Args::parse();
+
+    // Setup logging
+    let filter = EnvFilter::from_default_env().add_directive(args.log_level.into());
+    let _ = FmtSubscriber::builder()
+        .compact()
+        .without_time()
+        .with_max_level(args.log_level)
+        .with_env_filter(filter)
+        .try_init();
+
+    info!("Start NTRIP/RTMP tool");
+
+    debug!("Args {args:?}");
+
+    // Setup interrupt / exit handler
+    let (exit_tx, mut exit_rx) = tokio::sync::broadcast::channel(1);
+    let e = exit_tx.clone();
+    tokio::task::spawn(async move {
+        tokio::signal::ctrl_c().await.unwrap();
+        debug!("Received Ctrl-C, shutting down...");
+        e.send(()).unwrap();
+    });
+
+    let mut client = NtripClient::new(args.ntrip_host.clone(), args.ntrip_creds.clone()).await?;
+
+    match args.command {
+        Commands::List => {
+            // List available NTRIP mounts using SNIP
+            info!("Listing NTRIP mounts");
+
+            let info = client.list_mounts().await.unwrap();
+
+            for s in info.services {
+                info!(
+                    "{} - {} ({:.3}, {:.3})",
+                    s.name,
+                    s.details,
+                    s.location.latitude(),
+                    s.location.longitude()
+                );
+            }
+        },
+        Commands::FindNearest { lat, lon } => {
+            // Find the nearest NTRIP mount to the specified location
+            info!("Finding nearest NTRIP mount to ({}, {})", lat, lon);
+
+            let info = client.list_mounts().await.unwrap();
+
+            let target_location = Location::new(lat, lon);
+
+            match info.find_nearest(&target_location) {
+                Some((s, d)) => {
+                    info!(
+                        "Nearest mount: {} - {} ({:.3}, {:.3}), {:.3} km away",
+                        s.name,
+                        s.details,
+                        s.location.latitude(),
+                        s.location.longitude(),
+                        d / 1000.0
+                    );
+                },
+                None => {
+                    info!("No mounts found");
+                },
+            }
+        },
+        Commands::Subscribe { mount } => {
+            // Subscribe to the specified NTRIP mount
+            debug!("Connecting to NTRIP server");
+
+            // Setup the NTRIP client
+            let mut client = client.mount(mount, exit_tx.clone()).await?;
+
+            // Process incoming RTCM messages
+            loop {
+                select! {
+                    m = client.next() => match m {
+                        Some(m) => {
+                            info!("Received RTCM message: {:?}", m);
+                        },
+                        None => {
+                            error!("NTRIP client stream ended");
+                            break;
+                        }
+                    },
+                    _ = exit_rx.recv() => {
+                        info!("Exiting on signal");
+                        break;
+                    }
+                }
+            }
+        },
+    }
+
+    debug!("Exiting");
+
+    Ok(())
+}