Remove SourceId from ObservableSourceState.

Author: davidv1992

ntp-proto/src/algorithm/mod.rs

@@ -35,7 +35,7 @@ pub struct StateUpdate<ControllerMessage> {
 }
 
 // Note: this default implementation is necessary since the
-// derive only works if SourceId is Default (which it isn't
+// derive only works if ControllerMessage is Default (which it isn't
 // necessarily)
 impl<ControllerMessage> Default for StateUpdate<ControllerMessage> {
     fn default() -> Self {

ntp-proto/src/source.rs

@@ -1,5 +1,5 @@
 use crate::{
-    NtpVersion,
+    ClockId, NtpVersion,
     algorithm::Measurement,
     packet::{
         ExtensionField, NtpHeader,
@@ -117,12 +117,7 @@ impl<Controller: SourceController<MeasurementDelay = ()>> OneWaySource<Controlle
         self.controller.handle_message(message);
     }
 
-    pub fn observe<SourceId>(
-        &self,
-        name: String,
-        address: String,
-        id: SourceId,
-    ) -> ObservableSourceState<SourceId> {
+    pub fn observe(&self, name: String, address: String, id: ClockId) -> ObservableSourceState {
         ObservableSourceState {
             timedata: self.controller.observe(),
             unanswered_polls: 0,
@@ -432,15 +427,15 @@ macro_rules! actions {
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
-pub struct ObservableSourceState<SourceId> {
+pub struct ObservableSourceState {
     #[serde(flatten)]
     pub timedata: ObservableSourceTimedata,
     pub unanswered_polls: u32,
     pub poll_interval: PollInterval,
     pub nts_cookies: Option<usize>,
     pub name: String,
     pub address: String,
-    pub id: SourceId,
+    pub id: ClockId,
 }
 
 impl<Controller: SourceController<MeasurementDelay = NtpDuration>> NtpSource<Controller> {
@@ -482,7 +477,7 @@ impl<Controller: SourceController<MeasurementDelay = NtpDuration>> NtpSource<Con
         )
     }
 
-    pub fn observe<SourceId>(&self, name: String, id: SourceId) -> ObservableSourceState<SourceId> {
+    pub fn observe(&self, name: String, id: ClockId) -> ObservableSourceState {
         ObservableSourceState {
             timedata: self.controller.observe(),
             unanswered_polls: self.reach.unanswered_polls(),

ntpd/src/daemon/ntp_source.rs

@@ -49,7 +49,7 @@ pub struct SourceChannels<ControllerMessage, SourceMessage> {
     pub msg_for_system_sender: tokio::sync::mpsc::Sender<MsgForSystem<SourceMessage>>,
     pub system_update_receiver:
         tokio::sync::broadcast::Receiver<SystemSourceUpdate<ControllerMessage>>,
-    pub source_snapshots: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState<ClockId>>>>,
+    pub source_snapshots: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState>>>,
 }
 
 pub(crate) struct SourceTask<

ntpd/src/daemon/observer.rs

@@ -17,7 +17,7 @@ use serde::{Deserialize, Serialize};
 pub struct ObservableState {
     pub program: ProgramData,
     pub system: SystemSnapshot,
-    pub sources: Vec<ObservableSourceState<ClockId>>,
+    pub sources: Vec<ObservableSourceState>,
     pub servers: Vec<ObservableServerState>,
 }
 
@@ -70,7 +70,7 @@ impl From<&ServerData> for ObservableServerState {
 #[instrument(level = tracing::Level::ERROR, skip_all, name = "Observer", fields(path = debug(config.observation_path.clone())))]
 pub fn spawn<C: 'static + NtpClock + Send>(
     config: &super::config::ObservabilityConfig,
-    sources_reader: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState<ClockId>>>>,
+    sources_reader: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState>>>,
     server_reader: tokio::sync::watch::Receiver<Vec<ServerData>>,
     system_reader: tokio::sync::watch::Receiver<SystemSnapshot>,
     clock: C,
@@ -92,7 +92,7 @@ pub fn spawn<C: 'static + NtpClock + Send>(
 
 async fn observer<C: 'static + NtpClock + Send>(
     config: super::config::ObservabilityConfig,
-    sources_reader: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState<ClockId>>>>,
+    sources_reader: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState>>>,
     server_reader: tokio::sync::watch::Receiver<Vec<ServerData>>,
     system_reader: tokio::sync::watch::Receiver<SystemSnapshot>,
     clock: C,
@@ -174,7 +174,7 @@ async fn observer<C: 'static + NtpClock + Send>(
 async fn handle_connection(
     stream: &mut (impl tokio::io::AsyncWrite + Unpin),
     start_time: Instant,
-    sources_reader: &std::sync::RwLock<HashMap<ClockId, ObservableSourceState<ClockId>>>,
+    sources_reader: &std::sync::RwLock<HashMap<ClockId, ObservableSourceState>>,
     server_reader: tokio::sync::watch::Receiver<Vec<ServerData>>,
     system_reader: tokio::sync::watch::Receiver<SystemSnapshot>,
     now: NtpTimestamp,

ntpd/src/daemon/system.rs

@@ -85,7 +85,7 @@ impl<T: Wait> Wait for SingleshotSleep<T> {
 }
 
 pub struct DaemonChannels {
-    pub source_snapshots: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState<ClockId>>>>,
+    pub source_snapshots: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState>>>,
     pub server_data_receiver: tokio::sync::watch::Receiver<Vec<ServerData>>,
     pub system_snapshot_receiver: tokio::sync::watch::Receiver<SystemSnapshot>,
 }
@@ -185,7 +185,7 @@ struct SystemTask<C: NtpClock, Controller: TimeSyncController<Clock = C>, T: Wai
     system_snapshot_sender: tokio::sync::watch::Sender<SystemSnapshot>,
     system_update_sender:
         tokio::sync::broadcast::Sender<SystemSourceUpdate<Controller::ControllerMessage>>,
-    source_snapshots: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState<ClockId>>>>,
+    source_snapshots: Arc<std::sync::RwLock<HashMap<ClockId, ObservableSourceState>>>,
     server_data_sender: tokio::sync::watch::Sender<Vec<ServerData>>,
     keyset: tokio::sync::watch::Receiver<Arc<KeySet>>,
     ip_list: tokio::sync::watch::Receiver<Arc<[IpAddr]>>,