fix: pubsub emulator

src/sinks/gcp_pubsub/run.rs

@@ -49,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()

src/sinks/gcp_pubsub/setup.rs

@@ -27,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
@@ -50,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");