@@ -17,7 +17,7 @@ mod subscription;
1717mod topic;
1818mod utils;
1919
20- use std:: sync:: Arc ;
20+ use std:: { fs :: read_to_string , path :: Path , sync:: Arc } ;
2121
2222use crate :: {
2323 broker_metrics:: init_metrics,
@@ -45,68 +45,58 @@ pub(crate) mod admin_proto {
4545#[ derive( Parser , Debug ) ]
4646#[ command( version, about, long_about = None ) ]
4747struct Args {
48- /// Cluster Name
49- #[ arg( short = 'x' , long) ]
50- cluster_name : String ,
51-
52- /// Path to config file
48+ /// Path to config file (required)
5349#[ arg( short = 'c' , long) ]
54- config_file : Option < String > ,
50+ config_file : String ,
5551
56- /// Danube Broker advertised address
57- #[ arg( short = 'b' , long, default_value = "0.0.0.0:6650" ) ]
58- broker_addr : String ,
52+ /// Danube Broker advertised address (optional, overrides config file)
53+ #[ arg( short = 'b' , long) ]
54+ broker_addr : Option < String > ,
5955
60- /// Danube Admin address
61- #[ arg( short = 'a' , long, default_value = "0.0.0.0:50051" ) ]
62- admin_addr : String ,
56+ /// Danube Broker Admin address (optional, overrides config file)
57+ #[ arg( short = 'a' , long) ]
58+ admin_addr : Option < String > ,
6359
64- /// Prometheus exporter address
60+ /// Prometheus Exporter http address (optional, overrides config file)
6561#[ arg( short = 'p' , long) ]
6662 prom_exporter : Option < String > ,
67-
68- /// Metadata store address
69- #[ arg( short = 'm' , long) ]
70- meta_store_addr : Option < String > ,
71-
72- /// List of namespaces (comma-separated)
73- #[ arg( short = 'n' , long, value_parser = parse_namespaces_list) ]
74- namespaces : Vec < String > ,
7563}
7664
7765#[ tokio:: main]
7866async fn main ( ) -> Result < ( ) > {
79- // install global collector configured based on RUST_LOG env var.
80- //tracing_subscriber::fmt()
81- // .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
82- // .init();
83-
67+ // Initialize logging
8468 tracing_subscriber:: fmt:: init ( ) ;
8569
70+ // Parse command line arguments
8671 let args = Args :: parse ( ) ;
8772
88- let broker_addr: std:: net:: SocketAddr = args. broker_addr . parse ( ) ?;
89- let admin_addr: std:: net:: SocketAddr = args. admin_addr . parse ( ) ?;
73+ // Load the configuration from the specified YAML file
74+ let config_content = read_to_string ( Path :: new ( & args. config_file ) ) ?;
75+ let mut service_config: ServiceConfiguration = serde_yaml:: from_str ( & config_content) ?;
9076
91- // init metrics with or without prometheus exporter
92- if let Some ( prometheus_exporter) = args. prom_exporter {
77+ // If `broker_addr` is provided via command-line args, override the value from the config file
78+ if let Some ( broker_addr) = args. broker_addr {
79+ service_config. broker_addr = broker_addr;
80+ }
81+
82+ // If `admin_addr` is provided via command-line args, override the value from the config file
83+ if let Some ( admin_addr) = args. admin_addr {
84+ service_config. admin_addr = admin_addr;
85+ }
86+
87+ // If `prom_exporter` is provided via command-line args, override the value from the config file
88+ if let Some ( prom_exporter) = args. prom_exporter {
89+ service_config. prom_exporter = Some ( prom_exporter) ;
90+ }
91+
92+ // Init metrics with or without prometheus exporter
93+ if let Some ( prometheus_exporter) = service_config. prom_exporter . clone ( ) {
9394 let prom_addr: std:: net:: SocketAddr = prometheus_exporter. parse ( ) ?;
9495 init_metrics ( Some ( prom_addr) ) ;
9596 } else {
9697 init_metrics ( None )
9798 }
9899
99- // configuration settings for a Danube broker service
100- // includes various parameters that control the behavior and performance of the broker
101- // TODO! read from a config file, like danube.conf
102- let service_config = ServiceConfiguration {
103- cluster_name : args. cluster_name ,
104- broker_addr : broker_addr,
105- admin_addr : admin_addr,
106- meta_store_addr : args. meta_store_addr ,
107- bootstrap_namespaces : args. namespaces ,
108- } ;
109-
110100 // initialize the storage layer for Danube Metadata
111101 let store_config = MetadataStoreConfig :: new ( ) ;
112102 let metadata_store: MetadataStorage =
@@ -166,9 +156,3 @@ async fn main() -> Result<()> {
166156
167157 Ok ( ( ) )
168158}
169-
170- fn parse_namespaces_list ( s : & str ) -> Result < Vec < String > > {
171- Ok ( s. split ( ',' )
172- . map ( |nam| nam. trim ( ) . to_string ( ) )
173- . collect :: < Vec < String > > ( ) )
174- }
0 commit comments