Paulo/fix pubsub (#3)

* fix: use atomic operation to write cursor (txpipe#798)

* Release v1.8.6

* feat: add support for GCP pubsub emulator (txpipe#803)

* feat: make v1 compatible with Conway era (txpipe#807)

* Release v1.9.0

* fix: running pubsub on platform

* chore: formatting

* chore: cargo fmt

---------

Co-authored-by: Santiago Carmuega <[email protected]>
Co-authored-by: Joaquin Hoyos (Clark) <[email protected]>

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "oura"
 description = "The tail of Cardano"
-version = "1.9.2"
+version = "1.10.0"
 edition = "2021"
 repository = "https://github.com/txpipe/oura"
 homepage = "https://github.com/txpipe/oura"
@@ -86,5 +86,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" ,"google-cloud-gax"]
+gcp = ["google-cloud-pubsub", "google-cloud-googleapis", "tokio", "web", "google-cloud-gax"]
 rabbitmqsink = ["lapin", "tokio"]

src/sinks/elastic/run.rs

@@ -1,4 +1,4 @@
-use elasticsearch::{params::OpType, Elasticsearch, IndexParts, http::StatusCode};
+use elasticsearch::{http::StatusCode, params::OpType, Elasticsearch, IndexParts};
 use serde::Serialize;
 use serde_json::json;
 use std::sync::Arc;

src/sinks/gcp_pubsub/run.rs

@@ -1,6 +1,5 @@
 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},
@@ -50,9 +49,6 @@ 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()

src/sinks/gcp_pubsub/setup.rs

@@ -27,12 +27,6 @@ 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
@@ -56,9 +50,6 @@ impl SinkProvider for WithUtils<Config> {
                 &retry_policy,
                 &ordering_key,
                 &attributes,
-                use_emulator,
-                &emulator_endpoint,
-                &emulator_project_id,
                 utils,
             )
             .expect("writer loop failed");