Replace zmq with kafka

api/src/comms.rs

@@ -8,6 +8,7 @@ pub use zmq::Socket as ZmqSocket;
 use tokio::sync::{broadcast, mpsc, Mutex};
 
 use msgs::*;
+use utils::kafka::{Consumer, Producer};
 
 use utils::time;
 
@@ -32,8 +33,8 @@ impl CommsActor {
     pub async fn start(
         _tx: mpsc::Sender<Envelope>,
         mut rx: mpsc::Receiver<Envelope>,
-        subscriber: ZmqSocket,
-        sender: ZmqSocket,
+        mut kafka_consumer: Consumer,
+        kafka_producer: Producer,
         _api_settings: ApiSettings,
     ) {
         // users of the node actor leave their "contact details" behind so the response can be transfered back later.
@@ -54,8 +55,8 @@ impl CommsActor {
             let a_tx = a_tx.clone();
 
             thread::spawn(move || {
-                while let Ok(frames) = subscriber.recv_multipart(0x00) {
-                    if let Ok(message) = bincode::deserialize::<Message>(&frames[2]) {
+                while let Some(maybe_message) = kafka_consumer.consume() {
+                    if let Some(message) = maybe_message {
                         let _ = a_tx.send(message);
                     };
                 }
@@ -108,7 +109,7 @@ impl CommsActor {
                 waiting.lock().await.push((tx, func, time::time_now()));
             }
 
-            utils::xzmq::send_as_bincode(&sender, &message);
+            kafka_producer.produce("bank", &message);
         }
     }
 }

api/src/lib.rs

@@ -10,19 +10,18 @@ use tokio::sync::mpsc;
 
 use actix_ratelimit::{MemoryStore, MemoryStoreActor, RateLimiter};
 use core_types::DbPool;
-use utils::xzmq::SocketContext;
 
 pub mod comms;
 pub mod jwt;
 pub mod routes;
 
 use comms::*;
+use utils::kafka::{Consumer, Producer};
 
 #[derive(Serialize, Deserialize, Clone)]
 pub struct ApiSettings {
     psql_url: String,
-    api_zmq_push_address: String,
-    api_zmq_subscribe_address: String,
+    kafka_broker_addresses: String,
     quota_replenishment_interval_millis: u64,
     quota_size: u64,
 }
@@ -42,11 +41,16 @@ pub async fn start(settings: ApiSettings) -> std::io::Result<()> {
 
     let (tx, rx) = mpsc::channel(1024);
 
-    let context = SocketContext::new();
-    let subscriber = context.create_subscriber(&settings.api_zmq_subscribe_address);
-    let pusher = context.create_push(&settings.api_zmq_push_address);
+    let kafka_consumer = Consumer::new("api", "api", &settings.kafka_broker_addresses);
+    let kafka_producer = Producer::new(&settings.kafka_broker_addresses);
 
-    tokio::task::spawn(CommsActor::start(tx.clone(), rx, subscriber, pusher, settings.clone()));
+    tokio::task::spawn(CommsActor::start(
+        tx.clone(),
+        rx,
+        kafka_consumer,
+        kafka_producer,
+        settings.clone(),
+    ));
 
     let ratelimiter_store = MemoryStore::new();