Adding Redis support (wip) (#1755)

* wip

* Revert setup.py to a previous version

* cleanup

* cleanup to move redisclient into routines for dev and prod functions.

* updated docker-compose to use redisinsight v2 or greater

* wip2

* wip cleanup

* presence code simplification

* further cleanup

* More cleanup

* Remove async await on start_periodic_tasks

* fix for leadership locking

* remove log messages

* Cleanup and Use tokio Mutex

* Remove tokio-console support

* Updates to support Rust 2024 compiler

* remove unnecessary async

apps/framework-cli/Cargo.toml

@@ -78,6 +78,7 @@ sha2 = "0.10.8"
 hex = "0.4.2"
 constant_time_eq = "0.3.0"
 tokio-cron-scheduler = "0.11.0"
+redis = { version = "0.23.3", features = ["aio", "tokio-comp"] }
 indexmap = "2.5.0"
 
 [dev-dependencies]

apps/framework-cli/src/cli.rs

@@ -334,9 +334,10 @@ async fn top_command_handler(
  arc_metrics.start_listening_to_metrics(rx).await;
 
  check_project_name(&project_arc.name())?;
- run_local_infrastructure(&project_arc)?.show();
+ let project_arc_clone = Arc::clone(&project_arc);
+ run_local_infrastructure(&project_arc_clone)?.show();
 
- routines::start_development_mode(project_arc, &settings.features, arc_metrics)
+ routines::start_development_mode(project_arc_clone, &settings.features, arc_metrics)
  .await
  .map_err(|e| {
  RoutineFailure::error(Message {
@@ -513,9 +514,13 @@ async fn top_command_handler(
 
  check_project_name(&project_arc.name())?;
 
- routines::start_production_mode(project_arc, settings.features, arc_metrics)
- .await
- .unwrap();
+ routines::start_production_mode(
+ Arc::clone(&project_arc),
+ settings.features,
+ arc_metrics,
+ )
+ .await
+ .unwrap();
 
  wait_for_usage_capture(capture_handle).await;
 

apps/framework-cli/src/cli/routines.rs

@@ -110,6 +110,7 @@ use crate::infrastructure::processes::consumption_registry::ConsumptionProcessRe
 use crate::infrastructure::processes::functions_registry::FunctionProcessRegistry;
 use crate::infrastructure::processes::kafka_clickhouse_sync::SyncingProcessesRegistry;
 use crate::infrastructure::processes::process_registry::ProcessRegistries;
+use crate::infrastructure::redis::redis_client::RedisClient;
 use crate::infrastructure::stream::redpanda::fetch_topics;
 use crate::project::Project;
 
@@ -171,7 +172,7 @@ impl RoutineSuccess {
  }
 
  pub fn show(&self) {
- show_message!(self.message_type, self.message);
+ show_message!(self.message_type, self.message.clone());
  }
 }
 
@@ -279,6 +280,9 @@ pub async fn start_development_mode(
  }
  );
 
+ let mut redis_client = RedisClient::new(&project.name()).await?;
+ redis_client.start_periodic_tasks();
+
  let server_config = project.http_server_config.clone();
  let web_server = Webserver::new(
  server_config.host.clone(),
@@ -464,6 +468,8 @@ pub async fn start_development_mode(
  .await;
  };
 
+ let _ = redis_client.stop_periodic_tasks();
+
  Ok(())
 }
 
@@ -481,6 +487,9 @@ pub async fn start_production_mode(
  }
  );
 
+ let mut redis_client = RedisClient::new(&project.name()).await?;
+ redis_client.start_periodic_tasks();
+
  let server_config = project.http_server_config.clone();
  let web_server = Webserver::new(
  server_config.host.clone(),
@@ -607,6 +616,8 @@ pub async fn start_production_mode(
  .await;
  }
 
+ let _ = redis_client.stop_periodic_tasks();
+
  Ok(())
 }
 

apps/framework-cli/src/infrastructure.rs

@@ -19,4 +19,5 @@ pub mod ingest;
 pub mod migration;
 pub mod olap;
 pub mod processes;
+pub mod redis;
 pub mod stream;

apps/framework-cli/src/infrastructure/redis.rs

@@ -0,0 +1 @@
+pub mod redis_client;