Skip to content

Removed Watcher From Push Command #247

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
raphael-goetz merged 1 commit into from
Feb 23, 2026
Merged

Removed Watcher From Push Command #247

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions crates/cli/src/command/push/data_type_client_impl.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use crate::command::push::auth::get_authorization_metadata;
use crate::formatter::{error_without_trace, info};
use tonic::{Extensions, Request, transport::Channel};
use tucana::sagittarius::{
DataTypeUpdateRequest as SagittariusDataTypeUpdateRequest,
Expand All @@ -15,7 +16,9 @@ impl SagittariusDataTypeServiceClient {
pub async fn new(sagittarius_url: String, token: String) -> Self {
let client = match DataTypeServiceClient::connect(sagittarius_url).await {
Ok(client) => {
log::info!("Successfully connected to Sagittarius DataType Endpoint!");
info(String::from(
"Successfully connected to Sagittarius DataType Endpoint!",
));
client
}
Err(err) => panic!(
Expand All @@ -36,13 +39,13 @@ impl SagittariusDataTypeServiceClient {

match self.client.update(request).await {
Ok(response) => {
log::info!(
info(format!(
"Successfully transferred data types. Did Sagittarius updated them? {:?}",
&response
);
&response.into_inner().success
));
}
Err(err) => {
log::error!("Failed to update DataTypes: {:?}", err);
error_without_trace(format!("Failed to update DataTypes: {:?}", err));
}
};
}
Expand Down
14 changes: 9 additions & 5 deletions crates/cli/src/command/push/flow_type_client_impl.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
use crate::command::push::auth::get_authorization_metadata;
use crate::formatter::error_without_trace;
use crate::formatter::info;
use tonic::Extensions;
use tonic::Request;
use tonic::transport::Channel;
Expand All @@ -15,7 +17,9 @@ impl SagittariusFlowTypeServiceClient {
pub async fn new(sagittarius_url: String, token: String) -> Self {
let client = match FlowTypeServiceClient::connect(sagittarius_url).await {
Ok(client) => {
log::info!("Successfully connected to Sagittarius FlowType Endpoint!");
info(String::from(
"Successfully connected to Sagittarius FlowType Endpoint!",
));
client
}
Err(err) => panic!(
Expand All @@ -36,13 +40,13 @@ impl SagittariusFlowTypeServiceClient {

match self.client.update(request).await {
Ok(response) => {
log::info!(
info(format!(
"Successfully transferred FlowTypes. Did Sagittarius updated them? {:?}",
&response
);
&response.into_inner().success
));
}
Err(err) => {
log::error!("Failed to update FlowTypes: {:?}", err);
error_without_trace(format!("Failed to update FlowTypes: {:?}", err));
}
};
}
Expand Down
14 changes: 9 additions & 5 deletions crates/cli/src/command/push/function_client_impl.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
use crate::command::push::auth::get_authorization_metadata;
use crate::formatter::error_without_trace;
use crate::formatter::info;
use tonic::Extensions;
use tonic::Request;
use tonic::transport::Channel;
Expand All @@ -15,7 +17,9 @@ impl SagittariusRuntimeFunctionServiceClient {
pub async fn new(sagittarius_url: String, token: String) -> Self {
let client = match RuntimeFunctionDefinitionServiceClient::connect(sagittarius_url).await {
Ok(client) => {
log::info!("Successfully connected to Sagittarius RuntimeFunction Endpoint!");
info(String::from(
"Successfully connected to Sagittarius RuntimeFunction Endpoint!",
));
client
}
Err(err) => panic!(
Expand All @@ -39,13 +43,13 @@ impl SagittariusRuntimeFunctionServiceClient {

match self.client.update(request).await {
Ok(response) => {
log::info!(
info(format!(
"Successfully transferred RuntimeFunctions. Did Sagittarius updated them? {:?}",
&response
);
&response.into_inner().success
));
}
Err(err) => {
log::error!("Failed to update RuntimeFunctions: {:?}", err);
error_without_trace(format!("Failed to update RuntimeFunctions: {:?}", err));
}
};
}
Expand Down
147 changes: 30 additions & 117 deletions crates/cli/src/command/push/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,6 @@ use crate::analyser::core::Analyser;
use crate::command::push::data_type_client_impl::SagittariusDataTypeServiceClient;
use crate::command::push::flow_type_client_impl::SagittariusFlowTypeServiceClient;
use crate::command::push::function_client_impl::SagittariusRuntimeFunctionServiceClient;
use crate::formatter::{default, info};
use notify::event::ModifyKind;
use notify::{EventKind, RecursiveMode, Watcher, recommended_watcher};
use std::sync::mpsc::channel;
use std::time::{Duration, Instant};

mod auth;
mod data_type_client_impl;
Expand All @@ -16,122 +11,40 @@ mod function_client_impl;
pub async fn push(token: String, url: String, path: Option<String>) {
let dir_path = path.unwrap_or_else(|| "./definitions".to_string());

info(format!("Watching directory: {dir_path}"));
info(String::from("Press Ctrl+C to stop watching..."));

{
Analyser::new(dir_path.as_str()).report(false, true);
}

// Set up file watcher
let (tx, rx) = channel();
let mut watcher = recommended_watcher(tx).unwrap();
watcher
.watch(std::path::Path::new(&dir_path), RecursiveMode::Recursive)
.unwrap();

let mut last_run = Instant::now();

let mut analyzer = Analyser::new(dir_path.as_str());
let mut data_type_client =
SagittariusDataTypeServiceClient::new(url.clone(), token.clone()).await;
let mut flow_type_client =
SagittariusFlowTypeServiceClient::new(url.clone(), token.clone()).await;
let mut function_client = SagittariusRuntimeFunctionServiceClient::new(url, token).await;

loop {
if let Ok(Ok(event)) = rx.recv() {
match event.kind {
EventKind::Modify(modify) => {
if let ModifyKind::Data(_) = modify
&& last_run.elapsed() > Duration::from_millis(500)
{
default(String::from(
"\n\n\n--------------------------------------------------------------------------\n\n",
));
info(String::from("Change detected! Regenerating report..."));
let mut analyzer = Analyser::new(dir_path.as_str());

// No errors when reporter is empty!
if analyzer.reporter.is_empty() {
data_type_client
.update_data_types(
analyzer
.data_types
.iter()
.map(|d| d.definition_data_type.clone())
.collect(),
)
.await;
flow_type_client
.update_flow_types(
analyzer
.flow_types
.iter()
.map(|d| d.flow_type.clone())
.collect(),
)
.await;
function_client
.update_runtime_function_definitions(
analyzer
.functions
.iter()
.map(|d| d.function.clone())
.collect(),
)
.await;
}

analyzer.report(false, true);

last_run = Instant::now();
}
}
EventKind::Remove(_) => {
if last_run.elapsed() > Duration::from_millis(500) {
default(String::from(
"\n\n\n--------------------------------------------------------------------------\n\n",
));
info(String::from("Change detected! Regenerating report..."));
let mut analyzer = Analyser::new(dir_path.as_str());

// No errors when reporter is empty!
if analyzer.reporter.is_empty() {
data_type_client
.update_data_types(
analyzer
.data_types
.iter()
.map(|d| d.definition_data_type.clone())
.collect(),
)
.await;
flow_type_client
.update_flow_types(
analyzer
.flow_types
.iter()
.map(|d| d.flow_type.clone())
.collect(),
)
.await;
function_client
.update_runtime_function_definitions(
analyzer
.functions
.iter()
.map(|d| d.function.clone())
.collect(),
)
.await;
}

analyzer.report(false, true);
last_run = Instant::now();
}
}
_ => {}
}
}
}
analyzer.report(false, true);

data_type_client
.update_data_types(
analyzer
.data_types
.iter()
.map(|d| d.definition_data_type.clone())
.collect(),
)
.await;
flow_type_client
.update_flow_types(
analyzer
.flow_types
.iter()
.map(|d| d.flow_type.clone())
.collect(),
)
.await;
function_client
.update_runtime_function_definitions(
analyzer
.functions
.iter()
.map(|d| d.function.clone())
.collect(),
)
.await;
}