Support auth in frost connector

met_binary/src/main.rs

@@ -1,6 +1,6 @@
 use clap::Parser;
-use met_connectors::Frost;
 use met_connectors::LustreNetatmo;
+use met_connectors::{frost, Frost};
 use rove::{
     data_switch::{DataConnector, DataSwitch},
     load_pipelines, start_server,
@@ -15,8 +15,12 @@ struct Args {
     address: String,
     #[arg(short = 'l', long, default_value_t = Level::INFO)]
     max_trace_level: Level,
-    #[arg(short, long, default_value_t = String::from("sample_pipeline/fresh"))]
+    #[arg(short, long, default_value_t = String::from("sample_pipelines/fresh"))]
     pipeline_dir: String,
+    #[arg(long, default_value_t = String::from(""))]
+    frost_username: String,
+    #[arg(long, default_value_t = String::from(""))]
+    frost_password: String,
 }
 
 // TODO: use anyhow for error handling?
@@ -28,9 +32,22 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .with_max_level(args.max_trace_level)
         .init();
 
+    let frost_connector = Frost {
+        credentials: frost::Credentials {
+            username: args.frost_username,
+            password: args.frost_password,
+        },
+    };
+
     let data_switch = DataSwitch::new(HashMap::from([
-        ("frost", &Frost as &dyn DataConnector),
-        ("lustre_netatmo", &LustreNetatmo as &dyn DataConnector),
+        (
+            "frost",
+            Box::new(frost_connector) as Box<dyn DataConnector + Send>,
+        ),
+        (
+            "lustre_netatmo",
+            Box::new(LustreNetatmo) as Box<dyn DataConnector + Send>,
+        ),
     ]));
 
     start_server(

met_connectors/src/frost/fetch.rs

@@ -3,6 +3,8 @@ use chrono::{prelude::*, Duration};
 use chronoutil::RelativeDuration;
 use rove::data_switch::{self, DataCache, Polygon, SpaceSpec, TimeSpec, Timeseries, Timestamp};
 
+use super::Credentials;
+
 #[allow(clippy::type_complexity)]
 fn extract_data(
     mut resp: serde_json::Value,
@@ -168,6 +170,7 @@ pub async fn fetch_data_inner(
     num_leading_points: u8,
     num_trailing_points: u8,
     extra_spec: Option<&str>,
+    credentials: &Credentials,
 ) -> Result<DataCache, data_switch::Error> {
     // TODO: figure out how to share the client between rove reqs
     let client = reqwest::Client::new();
@@ -194,6 +197,7 @@ pub async fn fetch_data_inner(
 
     let resp: serde_json::Value = client
         .get("https://frost-beta.met.no/api/v1/obs/met.no/filter/get")
+        .basic_auth(&credentials.username, Some(&credentials.password))
         .query(&[
             extra_query_param,
             ("elementids", element_id.to_string()),

met_connectors/src/frost/mod.rs

@@ -39,8 +39,16 @@ pub enum Error {
     Misalignment(String),
 }
 
+#[derive(Debug, Clone)]
+pub struct Credentials {
+    pub username: String,
+    pub password: String,
+}
+
 #[derive(Debug)]
-pub struct Frost;
+pub struct Frost {
+    pub credentials: Credentials,
+}
 
 #[derive(Deserialize, Debug)]
 struct FrostObsBody {
@@ -114,6 +122,7 @@ impl DataConnector for Frost {
             num_leading_points,
             num_trailing_points,
             extra_spec,
+            &self.credentials,
         )
         .await
     }

met_connectors/src/lib.rs

@@ -1,4 +1,4 @@
-mod frost;
+pub mod frost;
 mod lustre_netatmo;
 
 pub use frost::Frost;

src/data_switch.rs

@@ -216,23 +216,23 @@ pub trait DataConnector: Sync + std::fmt::Debug {
 /// use std::collections::HashMap;
 ///
 /// let data_switch = DataSwitch::new(HashMap::from([
-///     ("test", &TestDataSource{
+///     ("test", Box::new(TestDataSource{
 ///         data_len_single: 3,
 ///         data_len_series: 1000,
 ///         data_len_spatial: 1000,
-///     } as &dyn DataConnector),
+///     }) as Box<dyn DataConnector + Send>),
 /// ]));
 /// ```
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub struct DataSwitch<'ds> {
-    sources: HashMap<&'ds str, &'ds dyn DataConnector>,
+    sources: HashMap<&'ds str, Box<dyn DataConnector + Send>>,
 }
 
 impl<'ds> DataSwitch<'ds> {
     /// Instantiate a new DataSwitch
     ///
     /// See the DataSwitch struct documentation for more info
-    pub fn new(sources: HashMap<&'ds str, &'ds dyn DataConnector>) -> Self {
+    pub fn new(sources: HashMap<&'ds str, Box<dyn DataConnector + Send>>) -> Self {
         Self { sources }
     }
 

src/lib.rs

@@ -16,11 +16,11 @@
 //! #[tokio::main]
 //! async fn main() -> Result<(), Box<dyn std::error::Error>> {
 //!     let data_switch = DataSwitch::new(HashMap::from([
-//!         ("test", &TestDataSource{
+//!         ("test", Box::new(TestDataSource{
 //!             data_len_single: 3,
 //!             data_len_series: 1000,
 //!             data_len_spatial: 1000,
-//!         } as &dyn DataConnector),
+//!         }) as Box<dyn DataConnector + Send>),
 //!     ]));
 //!
 //!     start_server(
@@ -46,11 +46,11 @@
 //! #[tokio::main]
 //! async fn main() -> Result<(), Box<dyn std::error::Error>> {
 //!     let data_switch = DataSwitch::new(HashMap::from([
-//!         ("test", &TestDataSource{
+//!         ("test", Box::new(TestDataSource{
 //!             data_len_single: 3,
 //!             data_len_series: 1000,
 //!             data_len_spatial: 1000,
-//!         } as &dyn DataConnector),
+//!         }) as Box<dyn DataConnector + Send>),
 //!     ]));
 //!
 //!     let rove_scheduler = Scheduler::new(construct_hardcoded_pipeline(), data_switch);

src/scheduler.rs

@@ -20,7 +20,7 @@ pub enum Error {
 /// Receiver type for QC runs
 ///
 /// Holds information about test pipelines and data sources
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub struct Scheduler<'a> {
     // this is pub so that the server can determine the number of checks in a pipeline to size
     // its channel with. can be made private if the server functionality is deprecated

tests/integration_test.rs

@@ -53,11 +53,11 @@ pub async fn set_up_rove(
 async fn integration_test_hardcoded_pipeline() {
     let data_switch = DataSwitch::new(HashMap::from([(
         "test",
-        &TestDataSource {
+        Box::new(TestDataSource {
             data_len_single: DATA_LEN_SINGLE,
             data_len_series: 1,
             data_len_spatial: DATA_LEN_SPATIAL,
-        } as &dyn DataConnector,
+        }) as Box<dyn DataConnector + Send>,
     )]));
 
     let (coordinator_future, mut client) =