refactor(agent): move notifier into services

Author: Reisen

src/agent.rs

@@ -66,7 +66,6 @@ use {
         config::Config,
         pyth::rpc,
         solana::network,
-        state::notifier,
     },
     anyhow::Result,
     futures_util::future::join_all,
@@ -79,6 +78,7 @@ pub mod legacy_schedule;
 pub mod market_schedule;
 pub mod metrics;
 pub mod pyth;
+pub mod services;
 pub mod solana;
 pub mod state;
 
@@ -144,7 +144,7 @@ impl Agent {
         }
 
         // Create the Notifier task for the Pythd RPC.
-        jhs.push(tokio::spawn(notifier(state.clone())));
+        jhs.push(tokio::spawn(services::notifier(state.clone())));
 
         // Spawn the Pythd API Server
         jhs.push(tokio::spawn(rpc::run(

src/agent/services.rs

@@ -0,0 +1,3 @@
+mod notifier;
+
+pub use notifier::notifier;

src/agent/services/notifier.rs

@@ -0,0 +1,31 @@
+//! Notifier
+//!
+//! The notifier is responsible for notifying subscribers who have registered
+//! for price sched updates.
+
+use {
+    crate::agent::state::Prices,
+    std::sync::Arc,
+};
+
+pub async fn notifier<S>(state: Arc<S>)
+where
+    S: Prices,
+{
+    let mut interval = tokio::time::interval(state.notify_interval_duration());
+    let mut exit = crate::agent::EXIT.subscribe();
+    loop {
+        Prices::drop_closed_subscriptions(&*state).await;
+        tokio::select! {
+            _ = exit.changed() => {
+                tracing::info!("Shutdown signal received.");
+                return;
+            }
+            _ = interval.tick() => {
+                if let Err(err) = state.send_notify_price_sched().await {
+                    tracing::error!(err = ?err, "Notifier: failed to send notify price sched.");
+                }
+            }
+        }
+    }
+}

src/agent/state.rs

@@ -19,10 +19,7 @@ pub mod api;
 pub mod global;
 pub mod keypairs;
 pub mod local;
-pub use api::{
-    notifier,
-    Prices,
-};
+pub use api::Prices;
 
 #[derive(Clone, Serialize, Deserialize, Debug)]
 #[serde(default)]
@@ -92,7 +89,6 @@ mod tests {
                 self,
                 AllAccountsData,
             },
-            notifier,
             Config,
             Prices,
             State,
@@ -108,6 +104,7 @@ mod tests {
                 ProductAccountMetadata,
                 PublisherAccount,
             },
+            services::notifier,
             solana::{
                 self,
                 network::Network,

src/agent/state/api.rs

@@ -46,10 +46,7 @@ use {
     },
     std::{
         collections::HashMap,
-        sync::{
-            atomic::AtomicI64,
-            Arc,
-        },
+        sync::atomic::AtomicI64,
         time::Duration,
     },
     tokio::sync::{
@@ -171,7 +168,6 @@ pub trait Prices {
         account_pubkey: &solana_sdk::pubkey::Pubkey,
         notify_price_sched_tx: mpsc::Sender<NotifyPriceSched>,
     ) -> SubscriptionID;
-    fn next_subscription_id(&self) -> SubscriptionID;
     async fn subscribe_price(
         &self,
         account: &solana_sdk::pubkey::Pubkey,
@@ -187,7 +183,8 @@ pub trait Prices {
         status: String,
     ) -> Result<()>;
     async fn update_global_price(&self, network: Network, update: &Update) -> Result<()>;
-    // TODO: implement FromStr method on PriceStatus
+    fn notify_interval_duration(&self) -> Duration;
+    fn next_subscription_id(&self) -> SubscriptionID;
     fn map_status(status: &str) -> Result<PriceStatus>;
 }
 
@@ -433,6 +430,10 @@ where
         }
     }
 
+    fn notify_interval_duration(&self) -> Duration {
+        self.into().notify_price_sched_interval_duration
+    }
+
     // TODO: implement FromStr method on PriceStatus
     fn map_status(status: &str) -> Result<PriceStatus> {
         match status {
@@ -445,27 +446,3 @@ where
         }
     }
 }
-
-pub async fn notifier<S>(state: Arc<S>)
-where
-    for<'a> &'a S: Into<&'a PricesState>,
-    S: Prices,
-{
-    let prices: &PricesState = (&*state).into();
-    let mut interval = tokio::time::interval(prices.notify_price_sched_interval_duration);
-    let mut exit = crate::agent::EXIT.subscribe();
-    loop {
-        Prices::drop_closed_subscriptions(&*state).await;
-        tokio::select! {
-            _ = exit.changed() => {
-                tracing::info!("Shutdown signal received.");
-                return;
-            }
-            _ = interval.tick() => {
-                if let Err(err) = state.send_notify_price_sched().await {
-                    tracing::error!(err = ?err, "Notifier: failed to send notify price sched.");
-                }
-            }
-        }
-    }
-}