From 1869e731316a8590f5415747d4d50db37a7d6421 Mon Sep 17 00:00:00 2001
From: Bryan Stitt <bryan@llamanodes.com>
Date: Thu, 2 Mar 2023 22:50:14 -0800
Subject: [PATCH] actually add the script

---
 .../src/bin/web3_proxy_cli/search_kafka.rs    | 60 +++++++++++++++++++
 1 file changed, 60 insertions(+)
 create mode 100644 web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs

diff --git a/web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs b/web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs
new file mode 100644
index 00000000..c4bc3bca
--- /dev/null
+++ b/web3_proxy/src/bin/web3_proxy_cli/search_kafka.rs
@@ -0,0 +1,60 @@
+use anyhow::Context;
+use argh::FromArgs;
+use futures::TryStreamExt;
+use log::info;
+use rdkafka::{
+    consumer::{Consumer, StreamConsumer},
+    ClientConfig, Message,
+};
+use std::num::NonZeroU64;
+use web3_proxy::{config::TopConfig, frontend::authorization::RpcSecretKey};
+
+/// Second subcommand.
+#[derive(FromArgs, PartialEq, Debug, Eq)]
+#[argh(subcommand, name = "search_kafka")]
+pub struct SearchKafkaSubCommand {
+    #[argh(positional)]
+    group_id: String,
+    #[argh(positional)]
+    input_topic: String,
+    #[argh(option)]
+    /// rpc_key to search. Be careful when handling keys!
+    rpc_key: Option<RpcSecretKey>,
+    #[argh(option)]
+    /// rpc_key_id to search
+    rpc_key_id: Option<NonZeroU64>,
+}
+
+impl SearchKafkaSubCommand {
+    pub async fn main(self, top_config: TopConfig) -> anyhow::Result<()> {
+        let brokers = top_config
+            .app
+            .kafka_urls
+            .context("top_config.app.kafka_urls is required")?;
+
+        // TODO: headers
+        // TODO: headers
+        let consumer: StreamConsumer = ClientConfig::new()
+            .set("group.id", &self.group_id)
+            .set("bootstrap.servers", &brokers)
+            .set("enable.partition.eof", "false")
+            .set("session.timeout.ms", "6000")
+            .set("enable.auto.commit", "false")
+            .create()
+            .context("Consumer creation failed")?;
+
+        consumer
+            .subscribe(&[&self.input_topic])
+            .expect("Can't subscribe to specified topic");
+
+        let stream_processor = consumer.stream().try_for_each(|msg| async move {
+            info!("Message received: {}", msg.offset());
+
+            Ok(())
+        });
+
+        stream_processor.await?;
+
+        Ok(())
+    }
+}