Merge pull request #378 from systemaccounting/377-graphql-measure-sink

377 graphql measure sink

Cargo.toml

@@ -8,6 +8,7 @@ members = [
 	"crates/service",
 	"crates/shutdown",
 	"crates/types",
+	"crates/wsclient",
 	"services/auto-confirm",
 	"services/balance-by-account",
 	"services/event",

crates/wsclient/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "wsclient"
+version = "0.1.0"
+edition = "2021"
+rust-version.workspace = true
+
+[dependencies]
+tungstenite = "0.24.0"

crates/wsclient/src/lib.rs

@@ -0,0 +1,128 @@
+use std::net::TcpStream;
+use tungstenite::{connect, stream::MaybeTlsStream, WebSocket};
+
+pub struct WsClient {
+    uri: WsUri,
+}
+
+impl WsClient {
+    pub fn new(
+        base_uri: String,
+        measure: String,
+        date: String,
+        country: Option<String>,
+        region: Option<String>,
+        municipality: Option<String>,
+    ) -> Self {
+        let uri = WsUri::new(base_uri, measure, date, country, region, municipality);
+        WsClient { uri }
+    }
+
+    pub fn connect(&self) -> WebSocket<MaybeTlsStream<TcpStream>> {
+        let (socket, _) = connect(self.uri.query_string()).expect("failed to connect");
+        socket
+    }
+}
+
+// todo: merge with Params in services/measure/src/main.rs as separate crate
+struct WsUri {
+    base_uri: String,
+    measure: String,
+    date: String,
+    country: Option<String>,
+    region: Option<String>,
+    municipality: Option<String>,
+}
+
+impl WsUri {
+    pub fn new(
+        base_uri: String,
+        measure: String,
+        date: String,
+        country: Option<String>,
+        region: Option<String>,
+        municipality: Option<String>,
+    ) -> Self {
+        WsUri {
+            base_uri,
+            measure,
+            date,
+            country,
+            region,
+            municipality,
+        }
+    }
+
+    pub fn query_string(&self) -> String {
+        let mut uri = format!("{}?", self.base_uri);
+        uri.push_str(&format!("measure={}", self.measure));
+        uri.push_str(&format!("&date={}", self.date));
+        if let Some(country) = &self.country {
+            uri.push_str(&format!("&country={}", country));
+        }
+        if let Some(region) = &self.region {
+            uri.push_str(&format!("&region={}", region));
+        }
+        if let Some(municipality) = &self.municipality {
+            uri.push_str(&format!("&municipality={}", municipality));
+        }
+        uri.replace(" ", "%20")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_ws_uri() {
+        let ws_uri = WsUri::new(
+            "ws://localhost:10010/ws".to_string(),
+            "gdp".to_string(),
+            "2024-08-21".to_string(),
+            Some("United States of America".to_string()),
+            Some("California".to_string()),
+            Some("Sacramento".to_string()),
+        );
+
+        assert_eq!(ws_uri.measure, "gdp");
+        assert_eq!(ws_uri.date, "2024-08-21");
+        assert_eq!(ws_uri.country, Some("United States of America".to_string()));
+        assert_eq!(ws_uri.region, Some("California".to_string()));
+        assert_eq!(ws_uri.municipality, Some("Sacramento".to_string()));
+    }
+
+    #[test]
+    fn test_ws_uri_full_query_string() {
+        let ws_uri = WsUri::new(
+            "ws://localhost:10010/ws".to_string(),
+            "gdp".to_string(),
+            "2024-08-21".to_string(),
+            Some("United States of America".to_string()),
+            Some("California".to_string()),
+            Some("Sacramento".to_string()),
+        );
+
+        assert_eq!(
+            ws_uri.query_string(),
+            "ws://localhost:10010/ws?measure=gdp&date=2024-08-21&country=United%20States%20of%20America&region=California&municipality=Sacramento"
+        );
+    }
+
+    #[test]
+    fn test_ws_uri_cal_query_string() {
+        let ws_uri = WsUri::new(
+            "ws://localhost:10010/ws".to_string(),
+            "gdp".to_string(),
+            "2024-08-21".to_string(),
+            Some("United States of America".to_string()),
+            Some("California".to_string()),
+            None,
+        );
+
+        assert_eq!(
+            ws_uri.query_string(),
+            "ws://localhost:10010/ws?measure=gdp&date=2024-08-21&country=United%20States%20of%20America&region=California"
+        );
+    }
+}

project.yaml

@@ -75,6 +75,12 @@ crates:
     type: lib
     env_var: null
     params: []
+  wsclient:
+    runtime: rust1.x
+    min_code_cov: 0
+    type: lib
+    env_var: null
+    params: []
 docker:
   env_var:
     set:
@@ -355,6 +361,8 @@ services:
         - BALANCE_BY_ACCOUNT_URL
         - READINESS_CHECK_PATH
         - GRAPHQL_PORT
+        - MEASURE_URL
+        - MEASURE_RESOURCE
   request-create:
     runtime: rust1.x
     min_code_cov: null
@@ -613,6 +621,12 @@ services:
         MEASURE_PORT:
           ssm: null
           default: 10010
+        MEASURE_URL:
+          ssm: null
+          default: null # script sets as $LOCAL_ADDRESS:$PORT
+        MEASURE_RESOURCE:
+          ssm: null
+          default: ws
       get:
         - PGHOST
         - PGPORT

scripts/create-env-file.sh

@@ -63,7 +63,14 @@ source ./scripts/set-uri-vars.sh
 
 function set_default_values() {
 	for s in "${SECRETS[@]}"; do
-		if [[ "$s" == *'_URL' ]]; then
+		# todo: hardcode protocol prefixes in apps. this ws:// exception for MEASURE_URL is a hack
+		if [[ "$s" == 'MEASURE_URL' ]]; then
+			SVC_NAME=$(printf '%s' "$s" | sed 's/_URL//')
+			PORT_ENV_VAR="$SVC_NAME"_PORT
+			PORT_VAL=$(yq "... | select(has(\"$PORT_ENV_VAR\")).$PORT_ENV_VAR.default" $PROJECT_CONF)
+			echo "$s=ws://$LOCAL_ADDRESS:$PORT_VAL" >> $ENV_FILE
+			continue
+		elif [[ "$s" == *'_URL' ]]; then
 			SVC_NAME=$(printf '%s' "$s" | sed 's/_URL//')
 			PORT_ENV_VAR="$SVC_NAME"_PORT
 			PORT_VAL=$(yq "... | select(has(\"$PORT_ENV_VAR\")).$PORT_ENV_VAR.default" $PROJECT_CONF)