WIP

edgeandnode · Dec 12, 2024 · 2637b18 · 2637b18
1 parent 7543ecf
commit 2637b18
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 9 deletions.
diff --git a/src/config.rs b/src/config.rs
@@ -140,7 +140,7 @@ pub enum ExchangeRateProvider {
 /// Kafka configuration.
 ///
 /// See [`Config`]'s [`kafka`](struct.Config.html#structfield.kafka).
-#[derive(Deserialize)]
+#[derive(Clone, Deserialize)]
 pub struct KafkaConfig(BTreeMap<String, String>);
 
 impl Default for KafkaConfig {

diff --git a/src/main.rs b/src/main.rs
@@ -109,7 +109,8 @@ async fn main() {
         }
         None => Default::default(),
     };
-    let indexer_blocklist = indexer_blocklist::Blocklist::spawn(conf.blocklist);
+    let indexer_blocklist =
+        indexer_blocklist::Blocklist::spawn(conf.blocklist, conf.kafka.clone().into());
     let mut network = network::service::spawn(
         http_client.clone(),
         network_subgraph_client,

diff --git a/src/network/indexer_blocklist.rs b/src/network/indexer_blocklist.rs
@@ -1,5 +1,7 @@
 use std::collections::{HashMap, HashSet};
 
+use futures::StreamExt;
+use rdkafka::Message;
 use thegraph_core::{alloy::primitives::Address, DeploymentId, ProofOfIndexing};
 use tokio::sync::watch;
 
@@ -13,7 +15,7 @@ pub struct Blocklist {
 }
 
 impl Blocklist {
-    pub fn spawn(init: Vec<BlocklistEntry>) -> Self {
+    pub fn spawn(init: Vec<BlocklistEntry>, kafka_config: rdkafka::ClientConfig) -> Self {
         let (blocklist_tx, blocklist_rx) = watch::channel(Default::default());
         let (poi_tx, poi_rx) = watch::channel(Default::default());
         let (indexer_tx, indexer_rx) = watch::channel(Default::default());
@@ -26,7 +28,7 @@ impl Blocklist {
             actor.add_entry(entry);
         }
         tokio::spawn(async move {
-            actor.run().await;
+            actor.run(kafka_config.create().unwrap()).await;
         });
         Self {
             blocklist: blocklist_rx,
@@ -37,14 +39,33 @@ impl Blocklist {
 }
 
 struct Actor {
-    pub blocklist: watch::Sender<Vec<BlocklistEntry>>,
-    pub poi: watch::Sender<HashMap<DeploymentId, Vec<(u64, ProofOfIndexing)>>>,
-    pub indexer: watch::Sender<HashMap<Address, HashSet<DeploymentId>>>,
+    blocklist: watch::Sender<Vec<BlocklistEntry>>,
+    poi: watch::Sender<HashMap<DeploymentId, Vec<(u64, ProofOfIndexing)>>>,
+    indexer: watch::Sender<HashMap<Address, HashSet<DeploymentId>>>,
 }
 
 impl Actor {
-    async fn run(&mut self) {
-        todo!();
+    async fn run(&mut self, consumer: rdkafka::consumer::StreamConsumer) {
+        let mut stream = consumer.stream();
+        while let Some(msg) = stream.next().await {
+            let msg = match msg {
+                Ok(msg) => msg,
+                Err(blocklist_recv_error) => {
+                    tracing::error!(%blocklist_recv_error);
+                    continue;
+                }
+            };
+            let entry: BlocklistEntry = match msg.payload().map(serde_json::from_slice) {
+                Some(Ok(entry)) => entry,
+                Some(Err(blocklist_deserialize_err)) => {
+                    tracing::error!(%blocklist_deserialize_err);
+                    continue;
+                }
+                None => continue,
+            };
+            self.add_entry(entry);
+        }
+        tracing::error!("blocklist consumer stopped");
     }
 
     fn add_entry(&mut self, entry: BlocklistEntry) {