refactor: making pusher and scheduler singletons

integrationos-emit/src/server.rs

@@ -193,9 +193,10 @@ impl Server {
             pusher.start(&config, s).await
         }));
 
+        let config = server.state.config.clone();
         subsys.start(SubsystemBuilder::new(
             "SchedulerSubsystem",
-            |s| async move { scheduler.start(s).await },
+            |s| async move { scheduler.start(&config, s).await },
         ));
 
         subsys.start(SubsystemBuilder::new("ServerSubsystem", |s| async move {

integrationos-emit/src/stream/mod.rs

@@ -9,6 +9,8 @@ use integrationos_domain::{Id, IntegrationOSError, Unit};
 use strum::{AsRefStr, Display, EnumIter, EnumString};
 use tokio_graceful_shutdown::SubsystemHandle;
 
+pub const SINGLETON_ID: u32 = 0;
+
 #[async_trait]
 pub trait EventStreamExt<T = EventEntity> {
     async fn publish(

integrationos-emit/src/stream/pusher.rs

@@ -1,4 +1,4 @@
-use super::EventStreamExt;
+use super::{EventStreamExt, SINGLETON_ID};
 use crate::{
     domain::{config::EmitterConfig, deduplication::Deduplication, event::EventEntity},
     stream::EventStreamTopic,
@@ -26,6 +26,11 @@ impl EventPusher {
         config: &EmitterConfig,
         subsys: SubsystemHandle,
     ) -> Result<Unit, IntegrationOSError> {
+        if config.partition()? != SINGLETON_ID {
+            tracing::info!("Limiting events to singleton id {}. Publisher proccessing finished.", SINGLETON_ID);
+            return Ok(());
+        }
+
         match self.process(config).cancel_on_shutdown(&subsys).await {
             Ok(result) => {
                 tracing::info!("Scheduled event publisher finished");

integrationos-emit/src/stream/scheduler.rs

@@ -1,5 +1,8 @@
-use super::EventStreamExt;
-use crate::{domain::event::ScheduledEvent, stream::EventStreamTopic};
+use super::{EventStreamExt, SINGLETON_ID};
+use crate::{
+    domain::{config::EmitterConfig, event::ScheduledEvent},
+    stream::EventStreamTopic,
+};
 use chrono::Utc;
 use futures::{StreamExt, TryStreamExt};
 use integrationos_domain::{IntegrationOSError, InternalError, MongoStore, Unit};
@@ -18,7 +21,19 @@ pub struct PublishScheduler {
 }
 
 impl PublishScheduler {
-    pub async fn start(&self, subsys: SubsystemHandle) -> Result<Unit, IntegrationOSError> {
+    pub async fn start(
+        &self,
+        config: &EmitterConfig,
+        subsys: SubsystemHandle,
+    ) -> Result<Unit, IntegrationOSError> {
+        if config.partition()? != SINGLETON_ID {
+            tracing::info!(
+                "Limiting events to singleton id {}. Scheduling proccessing finished.",
+                SINGLETON_ID
+            );
+            return Ok(());
+        }
+
         match self.process().cancel_on_shutdown(&subsys).await {
             Ok(result) => {
                 tracing::info!("Scheduled event publisher finished");