add support for pubsub emulator

Author: joacohoyos

Cargo.toml

@@ -80,5 +80,5 @@ elasticsink = ["elasticsearch", "tokio"]
 fingerprint = ["murmur3"]
 aws = ["aws-config", "aws-sdk-sqs", "aws-sdk-lambda", "aws-sdk-s3", "tokio"]
 redissink = ["redis", "tokio"]
-gcp = ["google-cloud-pubsub", "google-cloud-googleapis", "tokio", "web"]
+gcp = ["google-cloud-pubsub", "google-cloud-googleapis", "tokio", "web" ,"google-cloud-gax"]
 rabbitmqsink = ["lapin", "tokio"]

daemon.toml

@@ -0,0 +1,39 @@
+[source]
+type = "N2N"
+address = ["Tcp","node-mainnet-multiaz.jpgs-2uc.demeter.run:3000"]
+magic = "mainnet"
+# Because we are using blockfrost we had to delay the incoming blocks because
+# blockfrost would not have the txs indexed yet.
+# min_depth = 2
+
+# include_block_cbor = false
+# include_block_end_events = false
+# include_block_details = true
+
+[[filters]]
+type = "Selection"
+
+[filters.check]
+predicate = "all_of"
+
+[[filters.check.argument]]
+predicate = "variant_in"
+argument = ["Transaction"]
+
+[[filters.check.argument]]
+predicate = "policy_equals"
+argument = "8eb03958e1c5e0fbf2e8d4473050a92d178e66f9ff0c9089f07d7440"
+
+[sink]
+type = "Webhook"
+url = "http://localhost:9091"
+authorization = "user:pass"
+timeout = 30000
+error_policy = "Continue"
+
+[source.mapper]
+include_block_end_events = false
+include_transaction_details = true
+include_transaction_end_events = false
+include_block_cbor = false
+include_byron_ebb = false

src/sinks/gcp_pubsub/run.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashMap, sync::Arc};
 
+use google_cloud_gax::conn::Environment;
 use google_cloud_googleapis::pubsub::v1::PubsubMessage;
 use google_cloud_pubsub::{
     client::{Client, ClientConfig},
@@ -48,6 +49,9 @@ pub fn writer_loop(
     retry_policy: &retry::Policy,
     ordering_key: &str,
     attributes: &GenericKV,
+    emulator: bool,
+    emulator_endpoint: &Option<String>,
+    emulator_project_id: &Option<String>,
     utils: Arc<Utils>,
 ) -> Result<(), crate::Error> {
     let rt = tokio::runtime::Builder::new_current_thread()
@@ -56,7 +60,16 @@ pub fn writer_loop(
         .build()?;
 
     let publisher: Publisher = rt.block_on(async {
-        let client = Client::new(ClientConfig::default().with_auth().await?).await?;
+        let client_config = if emulator {
+            ClientConfig {
+                project_id: Some(emulator_project_id.clone().unwrap_or_default()),
+                environment: Environment::Emulator(emulator_endpoint.clone().unwrap_or_default()),
+                ..Default::default()
+            }
+        } else {
+            ClientConfig::default().with_auth().await?
+        };
+        let client = Client::new(client_config).await?;
         let topic = client.topic(topic_name);
         Result::<_, crate::Error>::Ok(topic.new_publisher(None))
     })?;

src/sinks/gcp_pubsub/setup.rs

@@ -16,6 +16,9 @@ pub struct Config {
     pub retry_policy: Option<retry::Policy>,
     pub ordering_key: Option<String>,
     pub attributes: Option<GenericKV>,
+    pub emulator: Option<bool>,
+    pub emulator_endpoint: Option<String>,
+    pub emulator_project_id: Option<String>,
 
     #[warn(deprecated)]
     pub credentials: Option<String>,
@@ -24,6 +27,12 @@ pub struct Config {
 impl SinkProvider for WithUtils<Config> {
     fn bootstrap(&self, input: StageReceiver) -> BootstrapResult {
         let topic_name = self.inner.topic.to_owned();
+        let mut use_emulator = self.inner.emulator.unwrap_or(false);
+        let emulator_endpoint = self.inner.emulator_endpoint.to_owned();
+        let emulator_project_id = self.inner.emulator_project_id.to_owned();
+        if use_emulator && (emulator_endpoint.is_none() || emulator_project_id.is_none()) {
+            use_emulator = false;
+        }
 
         let error_policy = self
             .inner
@@ -47,6 +56,9 @@ impl SinkProvider for WithUtils<Config> {
                 &retry_policy,
                 &ordering_key,
                 &attributes,
+                use_emulator,
+                &emulator_endpoint,
+                &emulator_project_id,
                 utils,
             )
             .expect("writer loop failed");