feat: add attributes field to GCP pubsub

book/src/sinks/gcp_pubsub.md

@@ -14,13 +14,19 @@ max_retries = 30
 backoff_unit =  5000
 backoff_factor = 2
 max_backoff = 100000
+
+[sink.attributes]
+network = "mainnet"
+version = "1"
 ```
 
 ### Section: `sink`
 
 - `type`: the literal value `GcpPubSub`.
 - `topic`: the short name of the topic to send message to.
 - `error_policy` (optional): either `Continue` or `Exit`. Default value is `Exit`.
+- `ordering_key` (optional): the key to use for ordering messages. If not specified, messages will be sent in an unordered manner.
+- `attributes` (optional): a map of attributes to add to each message. The key and value must be strings.
 - [retry_policy](../advanced/retry_policy.md)
 
 ### GCP Authentication
@@ -29,3 +35,4 @@ The GCP authentication process relies on the following conventions:
 
 - If the `GOOGLE_APPLICATION_CREDENTIALS` environmental variable is specified, the value will be used as the file path to retrieve the JSON file with the credentials.
 - If the server is running on GCP, the credentials will be retrieved from the metadata server.
+- If `PUBSUB_EMULATOR_HOST` environment variable is present, the sink will skip authentication and connect to the emulator instead of the production service.

src/sinks/gcp_pubsub/run.rs

@@ -15,15 +15,19 @@ use crate::{
     utils::{retry, Utils},
 };
 
+pub type GenericKV = HashMap<alloc::string::String, alloc::string::String>;
+
 async fn send_pubsub_msg(
     publisher: &Publisher,
     event: &Event,
     ordering_key: &str,
+    attributes: &GenericKV,
 ) -> Result<(), crate::Error> {
     let body = json!(event).to_string();
     let msg = PubsubMessage {
         data: body.into(),
         ordering_key: ordering_key.into(),
+        attributes: attributes.into(),
         ..Default::default()
     };
 
@@ -43,6 +47,7 @@ pub fn writer_loop(
     error_policy: &ErrorPolicy,
     retry_policy: &retry::Policy,
     ordering_key: &str,
+    attributes: &GenericKV,
     utils: Arc<Utils>,
 ) -> Result<(), crate::Error> {
     let rt = tokio::runtime::Builder::new_current_thread()
@@ -58,7 +63,7 @@ pub fn writer_loop(
 
     for event in input.iter() {
         let result = retry::retry_operation(
-            || rt.block_on(send_pubsub_msg(&publisher, &event, ordering_key)),
+            || rt.block_on(send_pubsub_msg(&publisher, &event, ordering_key, attributes)),
             retry_policy,
         );
 

src/sinks/gcp_pubsub/setup.rs

@@ -1,8 +1,11 @@
+use std::collections::HashMap;
+
 use serde::Deserialize;
 
 use crate::{
     pipelining::{BootstrapResult, SinkProvider, StageReceiver},
     sinks::common::web::ErrorPolicy,
+    sinks::gcp_pubsub::run::GenericKV,
     utils::{retry, WithUtils},
 };
 
@@ -14,6 +17,7 @@ pub struct Config {
     pub error_policy: Option<ErrorPolicy>,
     pub retry_policy: Option<retry::Policy>,
     pub ordering_key: Option<String>,
+    pub attributes: Option<GenericKV>,
 
     #[warn(deprecated)]
     pub credentials: Option<String>,
@@ -39,13 +43,16 @@ impl SinkProvider for WithUtils<Config> {
 
         let utils = self.utils.clone();
 
+        let attributes = self.inner.attributes.cloned().unwrap_or_default();
+
         let handle = std::thread::spawn(move || {
             writer_loop(
                 input,
                 &topic_name,
                 &error_policy,
                 &retry_policy,
                 &ordering_key,
+                &attributes,
                 utils,
             )
             .expect("writer loop failed");