WIP

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 {

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,

src/network/indexer_blocklist.rs

@@ -1,5 +1,6 @@
 use std::collections::{HashMap, HashSet};
 
+use futures::StreamExt;
 use thegraph_core::{alloy::primitives::Address, DeploymentId, ProofOfIndexing};
 use tokio::sync::watch;
 
@@ -13,11 +14,13 @@ 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());
+        let consumer = kafka_config.create().unwrap();
         let mut actor = Actor {
+            consumer,
             blocklist: blocklist_tx,
             poi: poi_tx,
             indexer: indexer_tx,
@@ -37,14 +40,18 @@ 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>>>,
+    consumer: rdkafka::consumer::StreamConsumer,
+    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!();
+        let mut stream = self.consumer.stream();
+        while let Some(msg) = stream.next().await {
+            todo!();
+        }
     }
 
     fn add_entry(&mut self, entry: BlocklistEntry) {