Feat/publish events to rmq

Cargo.toml

@@ -42,6 +42,7 @@ http = "0.2.6"
 humantime = "2.1.0"
 hyper = { version = "0.14.24", features = ["client", "http2"] }
 indicatif = "0.17.1"
+lapin = "2.3.3"
 lazy_static = "1.4.0"
 log = "0.4.14"
 mime = "0.3.16"

rmqconfig.json

@@ -0,0 +1,32 @@
+{
+  "queues": [
+    {
+      "name": "ord-q",
+      "vhost": "/",
+      "durable": true,
+      "auto_delete": false,
+      "arguments": {}
+    }
+  ],
+  "exchanges": [
+    {
+      "name": "ord-tx",
+      "vhost": "/",
+      "type": "topic",
+      "durable": true,
+      "auto_delete": false,
+      "internal": false,
+      "arguments": {}
+    }
+  ],
+  "bindings": [
+    {
+      "source": "ord-tx",
+      "vhost": "/",
+      "destination": "ord-q",
+      "destination_type": "queue",
+      "routing_key": "",
+      "arguments": {}
+    }
+  ]
+}

src/event_publisher.rs

@@ -0,0 +1,70 @@
+use anyhow::{Context, Result};
+use lapin::options::ConfirmSelectOptions;
+use lapin::{options::BasicPublishOptions, BasicProperties, Connection, ConnectionProperties};
+use tokio::runtime::Runtime;
+use tokio::sync::mpsc;
+
+use crate::index::event::Event;
+use crate::settings::Settings;
+
+pub struct EventPublisher {
+  pub(crate) sender: mpsc::Sender<Event>,
+}
+
+impl EventPublisher {
+  pub fn run(settings: &Settings) -> Result<Self, anyhow::Error> {
+    let addr = settings
+      .rabbitmq_addr()
+      .context("rabbitmq amqp credentials and url must be defined")?;
+
+    let exchange = settings
+      .rabbitmq_exchange()
+      .context("rabbitmq exchange path must be defined")?
+      .to_owned();
+
+    let (tx, mut rx) = mpsc::channel::<Event>(128);
+
+    let receiver = std::thread::spawn(move || {
+      Runtime::new().expect("runtime is setup").block_on(async {
+        let conn = Connection::connect(&addr, ConnectionProperties::default())
+          .await
+          .expect("connects to rabbitmq ok");
+
+        let channel = conn
+          .create_channel()
+          .await
+          .expect("creates rmq connection channel");
+
+        channel
+          .confirm_select(ConfirmSelectOptions::default())
+          .await
+          .expect("enable msg confirms");
+
+        while let Some(event) = rx.recv().await {
+          let message = serde_json::to_vec(&event).expect("failed to serialize event");
+
+          log::info!("publishing event: {:#?}", event);
+
+          let publish = channel
+            .basic_publish(
+              &exchange,
+              "",
+              BasicPublishOptions::default(),
+              &message,
+              BasicProperties::default(),
+            )
+            .await
+            .expect("published rmq msg")
+            .await
+            .expect("confirms rmq msg received");
+
+          assert!(publish.is_ack());
+        }
+      })
+    });
+
+    receiver.join().expect("spawn blocking event rx thread");
+
+    Ok(EventPublisher { sender: tx })
+  }
+}

src/index/event.rs

@@ -1,6 +1,6 @@
 use super::*;
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
 pub enum Event {
   InscriptionCreated {
     block_height: u32,

src/lib.rs

@@ -104,6 +104,7 @@ mod blocktime;
 pub mod chain;
 pub mod decimal;
 mod deserialize_from_str;
+mod event_publisher;
 mod fee_rate;
 pub mod index;
 mod inscriptions;

src/options.rs

@@ -82,6 +82,14 @@ pub struct Options {
     help = "Require basic HTTP authentication with <SERVER_USERNAME>. Credentials are sent in cleartext. Consider using authentication in conjunction with HTTPS."
   )]
   pub(crate) server_username: Option<String>,
+  #[arg(long, help = "RMQ url.")]
+  pub(crate) rabbitmq_url: Option<String>,
+  #[arg(long, help = "RMQ username.")]
+  pub(crate) rabbitmq_username: Option<String>,
+  #[arg(long, help = "RMQ password.")]
+  pub(crate) rabbitmq_password: Option<String>,
+  #[arg(long, help = "RMQ exchange to publish index events.")]
+  pub(crate) rabbitmq_exchange: Option<String>,
   #[arg(long, short, help = "Use regtest. Equivalent to `--chain regtest`.")]
   pub(crate) regtest: bool,
   #[arg(long, short, help = "Use signet. Equivalent to `--chain signet`.")]

src/settings.rs

@@ -27,6 +27,10 @@ pub struct Settings {
   server_password: Option<String>,
   server_url: Option<String>,
   server_username: Option<String>,
+  rabbitmq_url: Option<String>,
+  rabbitmq_username: Option<String>,
+  rabbitmq_password: Option<String>,
+  rabbitmq_exchange: Option<String>,
 }
 
 impl Settings {
@@ -141,6 +145,10 @@ impl Settings {
       server_password: self.server_password.or(source.server_password),
       server_url: self.server_url.or(source.server_url),
       server_username: self.server_username.or(source.server_username),
+      rabbitmq_url: self.rabbitmq_url.or(source.rabbitmq_url),
+      rabbitmq_username: self.rabbitmq_username.or(source.rabbitmq_username),
+      rabbitmq_password: self.rabbitmq_password.or(source.rabbitmq_password),
+      rabbitmq_exchange: self.rabbitmq_exchange.or(source.rabbitmq_exchange),
     }
   }
 
@@ -175,6 +183,10 @@ impl Settings {
       server_password: options.server_password,
       server_url: None,
       server_username: options.server_username,
+      rabbitmq_url: options.rabbitmq_url,
+      rabbitmq_username: options.rabbitmq_username,
+      rabbitmq_password: options.rabbitmq_password,
+      rabbitmq_exchange: options.rabbitmq_exchange,
     }
   }
 
@@ -253,6 +265,10 @@ impl Settings {
       server_password: get_string("SERVER_PASSWORD"),
       server_url: get_string("SERVER_URL"),
       server_username: get_string("SERVER_USERNAME"),
+      rabbitmq_url: get_string("RMQ_URL"),
+      rabbitmq_username: get_string("RMQ_USERNAME"),
+      rabbitmq_password: get_string("RMQ_PASSWORD"),
+      rabbitmq_exchange: get_string("RMQ_EXCHANGE"),
     })
   }
 
@@ -282,6 +298,10 @@ impl Settings {
       server_password: None,
       server_url: Some(server_url.into()),
       server_username: None,
+      rabbitmq_url: None,
+      rabbitmq_username: None,
+      rabbitmq_password: None,
+      rabbitmq_exchange: None,
     }
   }
 
@@ -361,6 +381,10 @@ impl Settings {
       server_password: self.server_password,
       server_url: self.server_url,
       server_username: self.server_username,
+      rabbitmq_url: self.rabbitmq_url,
+      rabbitmq_username: self.rabbitmq_username,
+      rabbitmq_password: self.rabbitmq_password,
+      rabbitmq_exchange: self.rabbitmq_exchange,
     })
   }
 
@@ -553,6 +577,18 @@ impl Settings {
   pub(crate) fn server_url(&self) -> Option<&str> {
     self.server_url.as_deref()
   }
+
+  pub fn rabbitmq_exchange(&self) -> Option<&str> {
+    self.rabbitmq_exchange.as_deref()
+  }
+
+  pub fn rabbitmq_addr(&self) -> Option<String> {
+    let user = self.rabbitmq_username.as_ref()?;
+    let pass = self.rabbitmq_password.as_ref()?;
+    let url = self.rabbitmq_url.as_ref()?;
+
+    Some(format!("amqp://{}:{}@{}", user, pass, url))
+  }
 }
 
 #[cfg(test)]
@@ -1000,7 +1036,10 @@ mod tests {
       ("NO_INDEX_INSCRIPTIONS", "1"),
       ("SERVER_PASSWORD", "server password"),
       ("SERVER_URL", "server url"),
-      ("SERVER_USERNAME", "server username"),
+      ("RMQ_URL", "http://127.0.0.1"),
+      ("RMQ_USERNAME", "rmq username"),
+      ("RMQ_PASSWORD", "rmq password"),
+      ("RMQ_EXCHANGE", "rmq exchange"),
     ]
     .into_iter()
     .map(|(key, value)| (key.into(), value.into()))
@@ -1044,6 +1083,10 @@ mod tests {
         server_password: Some("server password".into()),
         server_url: Some("server url".into()),
         server_username: Some("server username".into()),
+        rabbitmq_url: Some("http://127.0.0.1".into()),
+        rabbitmq_username: Some("rmq username".into()),
+        rabbitmq_password: Some("rmq password".into()),
+        rabbitmq_exchange: Some("rmq exchange".into()),
       }
     );
   }
@@ -1076,6 +1119,10 @@ mod tests {
           "--no-index-inscriptions",
           "--server-password=server password",
           "--server-username=server username",
+          "--rabbitmq-url=http://127.0.0.1",
+          "--rabbitmq-username=rmq username",
+          "--rabbitmq-password=rmq password",
+          "--rabbitmq-exchange=rmq exchange",
         ])
         .unwrap()
       ),
@@ -1104,6 +1151,10 @@ mod tests {
         server_password: Some("server password".into()),
         server_url: None,
         server_username: Some("server username".into()),
+        rabbitmq_url: Some("http://127.0.0.1".into()),
+        rabbitmq_username: Some("rmq username".into()),
+        rabbitmq_password: Some("rmq password".into()),
+        rabbitmq_exchange: Some("rmq exchange".into()),
       }
     );
   }

src/subcommand.rs

@@ -1,3 +1,5 @@
+use crate::event_publisher::EventPublisher;
+
 use super::*;
 
 pub mod balances;
@@ -39,6 +41,8 @@ pub(crate) enum Subcommand {
   Runes,
   #[command(about = "Run the explorer server")]
   Server(server::Server),
+  #[command(about = "Run the explorer server in event emit mode")]
+  EventServer(server::Server),
   #[command(about = "Display settings")]
   Settings,
   #[command(about = "Display information about a block's subsidy")]
@@ -71,6 +75,18 @@ impl Subcommand {
         LISTENERS.lock().unwrap().push(handle.clone());
         server.run(settings, index, handle)
       }
+      Self::EventServer(server) => {
+        log::info!("Starting event server");
+        let publisher = EventPublisher::run(&settings)?;
+        let handle = axum_server::Handle::new();
+        let index = Arc::new(Index::open_with_event_sender(
+          &settings,
+          Some(publisher.sender.clone()),
+        )?);
+
+        LISTENERS.lock().unwrap().push(handle.clone());
+        server.run(settings, index, handle)
+      }
       Self::Settings => settings::run(settings),
       Self::Subsidy(subsidy) => subsidy.run(),
       Self::Supply => supply::run(),