kafka: support YAML config (#197)

CHANGELOG.md

@@ -16,6 +16,7 @@ The minor version will be incremented upon a breaking change and the patch versi
 
 - client: add `GeyserGrpcClient::subscribe_once2` ([#195](https://github.com/rpcpool/yellowstone-grpc/pull/195)).
 - kafka: add metrics (stats, sent, recv) ([#196](https://github.com/rpcpool/yellowstone-grpc/pull/196)).
+- kafka: support YAML config ([#197](https://github.com/rpcpool/yellowstone-grpc/pull/197)).
 
 ### Fixes
 

yellowstone-grpc-kafka/Cargo.toml

@@ -14,11 +14,13 @@ clap = { version = "4.3.0", features = ["cargo", "derive"] }
 const-hex = "1.6.2"
 futures = "0.3.24"
 hyper = { version = "0.14.27", features = ["server"] }
+json5 = "0.4.1"
 lazy_static = "1.4.0"
 prometheus = "0.13.2"
 rdkafka = { version = "0.33.2", features = ["ssl", "sasl"] }
 serde = { version = "1.0.145", features = ["derive"] }
 serde_json = "1.0.86"
+serde_yaml = "0.9.25"
 sha2 = "0.10.7"
 tokio = { version = "1.21.2", features = ["rt-multi-thread", "macros", "time", "fs", "signal"] }
 tokio-stream = "0.1.11"

yellowstone-grpc-kafka/src/config.rs

@@ -35,12 +35,18 @@ pub struct Config {
 }
 
 impl Config {
-    pub async fn load(path: impl AsRef<Path>) -> anyhow::Result<Self> {
+    pub async fn load(path: impl AsRef<Path> + Copy) -> anyhow::Result<Self> {
         let text = fs::read_to_string(path)
             .await
             .context("failed to read config from file")?;
 
-        serde_json::from_str(&text).context("failed to parse config from file")
+        match path.as_ref().extension().and_then(|e| e.to_str()) {
+            Some("yaml") | Some("yml") => {
+                serde_yaml::from_str(&text).context("failed to parse config from file")
+            }
+            Some("json") => serde_yaml::from_str(&text).context("failed to parse config from file"),
+            value => anyhow::bail!("unknown config extension: {value:?}"),
+        }
     }
 }