allows restricting the streaming speed for the frontend

frontend/src/components/SvgGraph.vue

@@ -7,6 +7,7 @@ import TimeText from '@/components/TimeText.vue';
 const props = defineProps<{
     width: number;
     height: number;
+    duration?: number;
     border_left_right?: number;
     border_top_bottom?: number;
     utc?: boolean;
@@ -21,7 +22,12 @@ const height = computed(() => {
 });
 
 const now = useNow(33);
-const window_duration = 10 * 1000; // 10 seconds
+const window_duration = computed(() => {
+    if (props.duration) {
+        return props.duration;
+    }
+    return 10 * 1000; // 10 seconds
+});
 
 const time_lines = [
     1, // 1ms
@@ -58,7 +64,7 @@ const border_left_right = computed(() => props.border_left_right || 0);
 const border_top_bottom = computed(() => props.border_top_bottom || 0);
 
 const max_x = now;
-const min_x = computed(() => max_x.value - window_duration);
+const min_x = computed(() => max_x.value - window_duration.value);
 
 const x_map = (x: number) => {
     const diff_x = max_x.value - min_x.value;
@@ -79,10 +85,10 @@ provide<GraphData>(GRAPH_DATA, {
     height: () => height.value - 2 * border_top_bottom.value,
     x_map: x_map,
     lines: telemetry_lines,
-    max_update_rate: 1000 / 10
+    max_update_rate: 1000 / 10,
 });
 
-const duration = computed(() => {
+const line_duration = computed(() => {
     const diff_x = max_x.value - min_x.value;
     return time_lines.find((duration) => diff_x / duration >= 2)!;
 });
@@ -90,11 +96,11 @@ const duration = computed(() => {
 const lines = computed(() => {
     const result = [];
     for (
-        let i = Math.ceil(max_x.value / duration.value);
+        let i = Math.ceil(max_x.value / line_duration.value);
-        i >= Math.ceil(min_x.value / duration.value) - 5;
+        i >= Math.ceil(min_x.value / line_duration.value) - 5;
         i--
     ) {
-        const x = i * duration.value;
+        const x = i * line_duration.value;
         result.push(x);
     }
     return result;
@@ -140,7 +146,7 @@ const lines = computed(() => {
                     :y="height - border_top_bottom + text_offset"
                     :timestamp="tick"
                     :utc="props.utc"
-                    :show_millis="duration < 1000"
+                    :show_millis="line_duration < 1000"
                 ></TimeText>
             </template>
         </g>

frontend/src/components/TelemetryLine.vue

@@ -12,7 +12,6 @@ import {
     watch,
 } from 'vue';
 import {
-    type TelemetryDataItem,
     WEBSOCKET_SYMBOL,
     type WebsocketHandle,
 } from '@/composables/websocket';
@@ -23,16 +22,20 @@ import type { Point } from '@/graph/line';
 
 const props = defineProps<{
     data: string;
+    minimum_separation?: number;
     class?: string;
 }>();
 
-const smoothing_distance = 0.15 * 1000;
+const smoothing_distance_x = 5;
 
 const text_offset = computed(() => 10);
 
 const { data } = useTelemetry(() => props.data);
 const websocket = inject<ShallowRef<WebsocketHandle>>(WEBSOCKET_SYMBOL)!;
-const value = websocket.value.listen_to_telemetry(data);
+const value = websocket.value.listen_to_telemetry(
+    data,
+    props.minimum_separation,
+);
 
 const graph_data = inject<GraphData>(GRAPH_DATA)!;
 const axis_data = inject<AxisData>(AXIS_DATA)!;
@@ -46,7 +49,10 @@ function trigger_recompute() {
     recompute_points.value = Date.now();
 }
 function debounced_recompute() {
-    if (recompute_points.value + toValue(graph_data.max_update_rate) < Date.now()) {
+    if (
+        recompute_points.value + toValue(graph_data.max_update_rate) <
+        Date.now()
+    ) {
         trigger_recompute();
     }
 }
@@ -72,10 +78,12 @@ watch([value], ([val]) => {
         if (val_t >= min_x) {
             // TODO: Insert this in the right spot in memo
             const item_val = val.value[data.value!.data_type] as number;
-            const new_memo = [{
+            const new_memo = [
+                {
                     x: val_t,
                     y: item_val,
-            }].concat(memo.value);
+                },
+            ].concat(memo.value);
             if (item_val < min.value) {
                 min.value = item_val;
             }
@@ -100,10 +108,7 @@ watch([graph_data.min_x, graph_data.max_x], ([min_x, max_x]) => {
         }
     }
     if (max_x) {
-        while (
+        while (new_memo.length > 2 && new_memo[1].x > toValue(max_x)) {
-            new_memo.length > 2 &&
-            new_memo[1].x > toValue(max_x)
-        ) {
             new_memo.shift();
             memo_changed = true;
         }
@@ -137,7 +142,7 @@ watch(
 watch(
     [max, axis_data.axis_update_watch],
     ([max_val]) => {
-        trigger_recompute()
+        trigger_recompute();
         axis_data.max_y_callback(max_val);
     },
     {
@@ -145,7 +150,7 @@ watch(
     },
 );
 
-const points = ref("");
+const points = ref('');
 
 const old_max = ref(0);
 
@@ -153,21 +158,19 @@ const group_transform = computed(() => {
     const new_max = toValue(graph_data.max_x);
     const offset = graph_data.x_map(old_max.value) - graph_data.x_map(new_max);
     return `translate(${offset} 0)`;
-})
+});
 
-watch(
+watch([recompute_points], () => {
-    [recompute_points],
-    () => {
     let new_points = '';
     if (memo.value.length == 0 || data.value == null) {
         return '';
     }
 
-        const future_number = toValue(graph_data.max_x) + 99999999;
+    const future_number =
+        toValue(graph_data.max_x) + toValue(graph_data.max_update_rate);
 
-        let last_x = graph_data.x_map(future_number);
+    let last_x = graph_data.x_map(future_number) + smoothing_distance_x;
     old_max.value = toValue(graph_data.max_x);
-        let last_t = future_number + smoothing_distance;
 
     for (const data_item of memo.value) {
         const t = data_item.x;
@@ -175,20 +178,18 @@ watch(
         const x = graph_data.x_map(t);
         const y = axis_data.y_map(v);
 
-            if (last_t - t < smoothing_distance) {
+        if (last_x - x < smoothing_distance_x) {
             new_points += ` ${x},${y}`;
         } else {
             new_points += ` ${last_x},${y} ${x},${y}`;
         }
         last_x = x;
-            last_t = t;
         if (last_x <= 0.0) {
             break;
         }
     }
     points.value = new_points;
-    }
+});
-)
 
 const current_value = computed(() => {
     const val = value.value;
@@ -201,9 +202,7 @@ const current_value = computed(() => {
 
 <template>
     <g :class="`indexed-color color-${index}`">
-        <g
+        <g clip-path="url(#content)">
-            clip-path="url(#content)"
-        >
             <polyline
                 fill="none"
                 :transform="group_transform"

frontend/src/composables/websocket.ts

@@ -94,6 +94,7 @@ export class WebsocketHandle {
 
     listen_to_telemetry(
         telemetry: MaybeRefOrGetter<TelemetryDefinition | null>,
+        minimum_separation_ms: MaybeRefOrGetter<number> | undefined,
     ) {
         const value_result = ref<TelemetryDataItem | null>(null);
 
@@ -105,12 +106,23 @@ export class WebsocketHandle {
             return null;
         });
 
-        watch([uuid, this.connected], ([uuid_value, connected]) => {
+        const minimum_separation = computed(() => {
+            const min_sep = toValue(minimum_separation_ms);
+            if (min_sep) {
+                return min_sep;
+            }
+            return 0;
+        });
+
+        watch(
+            [uuid, this.connected, minimum_separation],
+            ([uuid_value, connected, min_sep]) => {
                 if (connected && uuid_value) {
                     this.websocket?.send(
                         JSON.stringify({
                             RegisterTlmListener: {
                                 uuid: uuid_value,
+                                minimum_separation_ms: min_sep,
                             },
                         }),
                     );
@@ -121,7 +133,8 @@ export class WebsocketHandle {
                         value_result.value = value;
                     });
                 }
-        });
+            },
+        );
 
         return value_result;
     }

frontend/src/graph/graph.ts

@@ -10,5 +10,5 @@ export interface GraphData {
     height: MaybeRefOrGetter<number>;
     x_map: (x: number) => number;
     lines: Ref<symbol[]>;
-    max_update_rate: MaybeRefOrGetter<number>,
+    max_update_rate: MaybeRefOrGetter<number>;
 }

frontend/src/graph/line.ts

@@ -1,5 +1,4 @@
-
 export interface Point {
-    x: number,
+    x: number;
-    y: number,
+    y: number;
 }

frontend/src/views/HomeView.vue

@@ -4,7 +4,6 @@ import { provide } from 'vue';
 import Graph from '@/components/SvgGraph.vue';
 import Axis from '@/components/GraphAxis.vue';
 import Line from '@/components/TelemetryLine.vue';
-import { AxisSide } from '@/graph/axis';
 
 const websocket = useWebsocket();
 provide(WEBSOCKET_SYMBOL, websocket);
@@ -29,16 +28,41 @@ provide(WEBSOCKET_SYMBOL, websocket);
             :height="400"
             :border_top_bottom="24"
             :border_left_right="128"
+            :duration="60 * 1000"
         >
             <Axis>
-                <Line data="simple_producer/sin"></Line>
+                <Line
-                <Line data="simple_producer/cos4"></Line>
+                    data="simple_producer/sin"
-                <Line data="simple_producer/sin2"></Line>
+                    :minimum_separation="100"
-                <Line data="simple_producer/cos"></Line>
+                ></Line>
-                <Line data="simple_producer/sin3"></Line>
+                <Line
-                <Line data="simple_producer/cos2"></Line>
+                    data="simple_producer/cos4"
-                <Line data="simple_producer/sin4"></Line>
+                    :minimum_separation="100"
-                <Line data="simple_producer/cos3"></Line>
+                ></Line>
+                <Line
+                    data="simple_producer/sin2"
+                    :minimum_separation="100"
+                ></Line>
+                <Line
+                    data="simple_producer/cos"
+                    :minimum_separation="100"
+                ></Line>
+                <Line
+                    data="simple_producer/sin3"
+                    :minimum_separation="100"
+                ></Line>
+                <Line
+                    data="simple_producer/cos2"
+                    :minimum_separation="100"
+                ></Line>
+                <Line
+                    data="simple_producer/sin4"
+                    :minimum_separation="100"
+                ></Line>
+                <Line
+                    data="simple_producer/cos3"
+                    :minimum_separation="100"
+                ></Line>
             </Axis>
         </Graph>
         <Graph
@@ -46,6 +70,7 @@ provide(WEBSOCKET_SYMBOL, websocket);
             :height="400"
             :border_top_bottom="24"
             :border_left_right="128"
+            :duration="5 * 1000"
         >
         </Graph>
         <Graph
@@ -53,6 +78,7 @@ provide(WEBSOCKET_SYMBOL, websocket);
             :height="400"
             :border_top_bottom="24"
             :border_left_right="128"
+            :duration="2 * 1000"
         >
         </Graph>
     </main>

server/src/grpc.rs

@@ -33,7 +33,6 @@ impl TelemetryService for CoreTelemetryService {
         trace!("CoreTelemetryService::new_telemetry");
         self.tlm_management
             .register(request.into_inner())
-            .await
             .map(|uuid| {
                 Response::new(TelemetryDefinitionResponse {
                     uuid: Some(Uuid { value: uuid }),
@@ -66,7 +65,7 @@ impl TelemetryService for CoreTelemetryService {
                         match message {
                             Ok(tlm_item) => {
                                 tx
-                                    .send(Self::handle_new_tlm_item(&tlm_management, &tlm_item).await)
+                                    .send(Self::handle_new_tlm_item(&tlm_management, &tlm_item))
                                     .await
                                     .expect("working rx");
                             }
@@ -85,7 +84,7 @@ impl TelemetryService for CoreTelemetryService {
 }
 
 impl CoreTelemetryService {
-    async fn handle_new_tlm_item(
+    fn handle_new_tlm_item(
         tlm_management: &Arc<TelemetryManagementService>,
         tlm_item: &TelemetryItem,
     ) -> Result<TelemetryInsertResponse, Status> {
@@ -93,7 +92,7 @@ impl CoreTelemetryService {
         let Some(ref uuid) = tlm_item.uuid else {
             return Err(Status::failed_precondition("UUID Missing"));
         };
-        let Some(tlm_data) = tlm_management.get_by_uuid(&uuid.value).await else {
+        let Some(tlm_data) = tlm_management.get_by_uuid(&uuid.value) else {
             return Err(Status::not_found("Telemetry Item Not Found"));
         };
 

server/src/http.rs

@@ -7,7 +7,10 @@ use derive_more::{Display, Error};
 use log::{error, trace};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
-use tokio::select;
+use std::time::Duration;
+use tokio::{pin, select};
+use tokio::sync::mpsc::Sender;
+use tokio::time::{sleep, Instant};
 use tokio_util::sync::CancellationToken;
 use tonic::codegen::tokio_stream::StreamExt;
 
@@ -30,9 +33,15 @@ impl error::ResponseError for UserError {
     }
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct RegisterTlmListenerRequest {
+    uuid: String,
+    minimum_separation_ms: u32,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 enum WebsocketRequest {
-    RegisterTlmListener { uuid: String },
+    RegisterTlmListener(RegisterTlmListenerRequest),
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -50,13 +59,70 @@ async fn get_tlm_definition(
 ) -> Result<impl Responder, UserError> {
     let string = name.to_string();
     trace!("get_tlm_definition {}", string);
-    let Some(data) = data.get_by_name(&string).await else {
+    let Some(data) = data.get_by_name(&string) else {
         return Err(UserError::TlmNotFound { tlm: string });
     };
 
     Ok(web::Json(data.definition.clone()))
 }
 
+fn handle_register_tlm_listener(data: &Arc<TelemetryManagementService>, request: RegisterTlmListenerRequest, tx: &Sender<WebsocketResponse>) {
+    if let Some(tlm_data) = data.get_by_uuid(&request.uuid) {
+        let minimum_separation = Duration::from_millis(request.minimum_separation_ms as u64);
+        let mut rx = tlm_data.data.subscribe();
+        let tx = tx.clone();
+        rt::spawn(async move {
+            let mut last_sent_at = Instant::now() - minimum_separation;
+            let mut last_value = None;
+            let sleep = sleep(Duration::from_millis(0));
+            pin!(sleep);
+            loop {
+                select! {
+                    _ = tx.closed() => {
+                        break;
+                    }
+                    Ok(_) = rx.changed() => {
+                        let now = Instant::now();
+                        let value = {
+                            let ref_val = rx.borrow_and_update();
+                            ref_val.clone()
+                        };
+                        if last_sent_at + minimum_separation > now {
+                            last_value = value;
+                            sleep.as_mut().reset(last_sent_at + minimum_separation);
+                            continue;
+                        } else {
+                            last_value = None;
+                            last_sent_at = now;
+                        }
+                        let _ = tx.send(WebsocketResponse::TlmValue {
+                            uuid: request.uuid.clone(),
+                            value,
+                        }).await;
+                    }
+                    () = &mut sleep => {
+                        if let Some(value) = last_value {
+                            let _ = tx.send(WebsocketResponse::TlmValue {
+                                uuid: request.uuid.clone(),
+                                value: Some(value),
+                            }).await;
+                        }
+                        last_value = None;
+                        let now = Instant::now();
+                        last_sent_at = now;
+                    }
+                }
+            }
+        });
+    }
+}
+
+async fn handle_websocket_message(data: &Arc<TelemetryManagementService>, request: WebsocketRequest, tx: &Sender<WebsocketResponse>) {
+    match request {
+        WebsocketRequest::RegisterTlmListener(request) => handle_register_tlm_listener(data, request, tx),
+    };
+}
+
 async fn websocket_connect(
     req: HttpRequest,
     stream: web::Payload,
@@ -101,33 +167,7 @@ async fn websocket_connect(
                         Ok(AggregatedMessage::Text(msg)) => {
                             match serde_json::from_str::<WebsocketRequest>(&msg) {
                                 Ok(request) => {
-                                    match request {
+                                    handle_websocket_message(data.get_ref(), request, &tx).await;
-                                        WebsocketRequest::RegisterTlmListener{ uuid } => {
-                                            if let Some(tlm_data) = data.get_by_uuid(&uuid).await {
-                                                let mut rx = tlm_data.data.subscribe();
-                                                let tx = tx.clone();
-                                                rt::spawn(async move {
-                                                    loop {
-                                                        select! {
-                                                            _ = tx.closed() => {
-                                                                break;
-                                                            }
-                                                            Ok(_) = rx.changed() => {
-                                                                let value = {
-                                                                    let ref_val = rx.borrow_and_update();
-                                                                    ref_val.clone()
-                                                                };
-                                                                let _ = tx.send(WebsocketResponse::TlmValue {
-                                                                    uuid: uuid.clone(),
-                                                                    value,
-                                                                }).await;
-                                                            }
-                                                        }
-                                                    }
-                                                });
-                                            }
-                                        }
-                                    }
                                 },
                                 Err(err) => {
                                     error!("JSON Deserialization Error Encountered {err}");

server/src/telemetry.rs

@@ -1,12 +1,9 @@
 use crate::core::{TelemetryDataType, TelemetryDefinitionRequest, Uuid};
-use log::trace;
 use serde::de::Visitor;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
-use std::collections::HashMap;
 use std::error::Error;
 use std::fmt::Formatter;
-use std::sync::Arc;
+use papaya::HashMap;
-use tokio::sync::RwLock;
 
 fn tlm_data_type_serialzier<S>(
     tlm_data_type: &TelemetryDataType,
@@ -70,67 +67,47 @@ pub struct TelemetryData {
 }
 
 pub struct TelemetryManagementService {
-    uuid_mapping: Arc<RwLock<HashMap<String, String>>>,
+    uuid_index: HashMap<String, String>,
-    tlm_mapping: Arc<RwLock<HashMap<String, TelemetryData>>>,
+    tlm_data: HashMap<String, TelemetryData>,
 }
 
 impl TelemetryManagementService {
     pub fn new() -> Self {
         Self {
-            uuid_mapping: Arc::new(RwLock::new(HashMap::new())),
+            uuid_index: HashMap::new(),
-            tlm_mapping: Arc::new(RwLock::new(HashMap::new())),
+            tlm_data: HashMap::new(),
         }
     }
 
-    pub async fn register(
+    pub fn register(
         &self,
         telemetry_definition_request: TelemetryDefinitionRequest,
     ) -> Result<String, Box<dyn Error>> {
-        let lock = self.uuid_mapping.read().await;
+        let uuid_index = self.uuid_index.pin();
-        if let Some(uuid) = lock.get(&telemetry_definition_request.name) {
+        let tlm_data = self.tlm_data.pin();
-            trace!("Telemetry Definition Found {:?}", uuid);
+
-            let tlm_lock = self.tlm_mapping.read().await;
+        let uuid = uuid_index.get_or_insert_with(telemetry_definition_request.name.clone(), || Uuid::random().value).clone();
-            if let Some(TelemetryData { definition, .. }) = tlm_lock.get(uuid) {
+
-                if definition.data_type != telemetry_definition_request.data_type() {
+        let _ = tlm_data.try_insert(uuid.clone(), TelemetryData {
-                    return Err("A telemetry item of the same name already exists".into());
-                }
-                Ok(uuid.clone())
-            } else {
-                Err("Could not find Telemetry Data".into())
-            }
-        } else {
-            trace!(
-                "Adding New Telemetry Definition {:?}",
-                telemetry_definition_request
-            );
-            drop(lock);
-            let mut lock = self.uuid_mapping.write().await;
-            let mut tlm_lock = self.tlm_mapping.write().await;
-            let uuid = Uuid::random().value;
-            lock.insert(telemetry_definition_request.name.clone(), uuid.clone());
-            tlm_lock.insert(
-                uuid.clone(),
-                TelemetryData {
             definition: TelemetryDefinition {
                 uuid: uuid.clone(),
                 name: telemetry_definition_request.name.clone(),
                 data_type: telemetry_definition_request.data_type(),
             },
             data: tokio::sync::watch::channel(None).0,
-                },
+        });
-            );
+
         Ok(uuid)
     }
+
+    pub fn get_by_name(&self, name: &String) -> Option<TelemetryData> {
+        let uuid_index = self.uuid_index.pin();
+        let uuid = uuid_index.get(name)?;
+        self.get_by_uuid(uuid)
     }
 
-    pub async fn get_by_name(&self, name: &String) -> Option<TelemetryData> {
+    pub fn get_by_uuid(&self, uuid: &String) -> Option<TelemetryData> {
-        let uuid_lock = self.uuid_mapping.read().await;
+        let tlm_data = self.tlm_data.pin();
-        let uuid = uuid_lock.get(name)?;
+        tlm_data.get(uuid).cloned()
-        self.get_by_uuid(uuid).await
-    }
-
-    pub async fn get_by_uuid(&self, uuid: &String) -> Option<TelemetryData> {
-        let tlm_lock = self.tlm_mapping.read().await;
-        tlm_lock.get(uuid).cloned()
     }
 }