RealtimeTrigger
Consume a message in real-time from a Pub/Sub topic and create one execution per message.
If you would like to consume multiple messages processed within a given time frame and process them in batch, you can use the io.kestra.plugin.gcp.pubsub.Trigger instead.
type: "io.kestra.plugin.gcp.pubsub.RealtimeTrigger"Consume a message from a Pub/Sub topic in real-time.
id: realtime_pubsub
namespace: company.team
tasks:
- id: log
type: io.kestra.plugin.core.log.Log
message: "Received: {{ trigger.data }}"
triggers:
- id: trigger
type: io.kestra.plugin.gcp.pubsub.RealtimeTrigger
projectId: test-project-id
topic: test-topic
subscription: test-subscription
Use GCP Pub/Sub Realtime Trigger to push events into Firestore
id: pubsub_realtime_trigger
namespace: company.team
tasks:
- id: insert_into_firestore
type: io.kestra.plugin.gcp.firestore.Set
projectId: test-project-id
collection: orders
document:
order_id: "{{ trigger.data | jq('.order_id') | first }}"
customer_name: "{{ trigger.data | jq('.customer_name') | first }}"
customer_email: "{{ trigger.data | jq('.customer_email') | first }}"
product_id: "{{ trigger.data | jq('.product_id') | first }}"
price: "{{ trigger.data | jq('.price') | first }}"
quantity: "{{ trigger.data | jq('.quantity') | first }}"
total: "{{ trigger.data | jq('.total') | first }}"
triggers:
- id: realtime_trigger
type: io.kestra.plugin.gcp.pubsub.RealtimeTrigger
projectId: test-project-id
topic: orders
subscription: kestra-subscription
serdeType: JSON
The Pub/Sub topic
The Pub/Sub topic. It must be created before executing the task.
Whether the Pub/Sub subscription should be created if not exist
The GCP service account to impersonate.
Max duration in the Duration ISO format, after that the task will end.
Max number of records, when reached the task will end.
The GCP project ID.
The GCP scopes to be used.
The serializer/deserializer to use.
The GCP service account.
List of execution states after which a trigger should be stopped (a.k.a. disabled).
The Pub/Sub subscription
The Pub/Sub subscription. It will be created automatically if it didn't exist and 'autoCreateSubscription' is enabled.
The message attributes map
The message data, must be a string if serde type is 'STRING', otherwise a JSON object
If it's a string, it can be a dynamic property otherwise not.
The message identifier
The message ordering key