Trigger Trigger

yaml
type: "io.kestra.plugin.mqtt.Trigger"

Consume messages periodically from MQTT topics and create one execution per batch.

Note that you don't need an extra task to consume the message from the event trigger. The trigger will automatically consume messages and you can retrieve their content in your flow using the {{ trigger.uri }} variable. If you would like to consume each message from MQTT topics in real-time and create one execution per message, you can use the io.kestra.plugin.mqtt.RealtimeTrigger instead.

Examples

yaml
    id: mqtt_trigger
    namespace: company.team

    tasks:
      - id: log
        type: io.kestra.plugin.core.log.Log
        message: "{{ trigger.payload }}"

    triggers:
      - id: trigger
        type: io.kestra.plugin.mqtt.Trigger
        server: tcp://localhost:1883
        clientId: kestraProducer
        topic:
          - kestra/sensors/cpu
          - kestra/sensors/mem
        serdeType: JSON
        maxRecords: 10

Properties

qos

  • Type: integer
  • Dynamic:
  • Required: ✔️
  • Default: 1

Sets the quality of service for this message.

  • Quality of Service 0: indicates that a message should be delivered at most once (zero or one times). The message will not be persisted to disk, and will not be acknowledged across the network. This QoS is the fastest, but should only be used for messages which are not valuable - note that if the server cannot process the message (for example, there is an authorization problem). Also known as "fire and forget".
  • Quality of Service 1: indicates that a message should be delivered at least once (one or more times). The message can only be delivered safely if it can be persisted, so the application must supply a means of persistence using MqttConnectOptions. If a persistence mechanism is not specified, the message will not be delivered in the event of a client failure. The message will be acknowledged across the network.
  • Quality of Service 2: indicates that a message should be delivered once. The message will be persisted to disk, and will be subject to a two-phase acknowledgement across the network. The message can only be delivered safely if it can be persisted, so the application must supply a means of persistence using MqttConnectOptions. If a persistence mechanism is not specified, the message will not be delivered in the event of a client failure. If persistence is not configured, QoS 1 and 2 messages will still be delivered in the event of a network or server problem as the client will hold state in memory. If the MQTT client is shutdown or fails and persistence is not configured then delivery of QoS 1 and 2 messages can not be maintained as client-side state will be lost.

serdeType

  • Type: string
  • Dynamic:
  • Required: ✔️
  • Default: JSON
  • Possible Values:
    • STRING
    • JSON
    • BYTES

Serializer / Deserializer used for the payload

topic

  • Type: object
  • Dynamic: ✔️
  • Required: ✔️

Topic where to consume message

Can be a string or a List of string to consume from multiple topic

version

  • Type: string
  • Dynamic:
  • Required: ✔️
  • Default: V5
  • Possible Values:
    • V3
    • V5

authMethod

  • Type: string
  • Dynamic:
  • Required:

clientId

  • Type: string
  • Dynamic:
  • Required:

conditions

  • Type: array
  • SubType: Condition
  • Dynamic:
  • Required:

List of conditions in order to limit the flow trigger.

connectionTimeout

  • Type: string
  • Dynamic:
  • Required:
  • Format: duration

crt

  • Type: string
  • Dynamic:
  • Required:

httpsHostnameVerificationEnabled

  • Type: boolean
  • Dynamic:
  • Required:

interval

  • Type: string
  • Dynamic:
  • Required:
  • Default: 60.000000000
  • Format: duration

Interval between polling.

The interval between 2 different polls of schedule, this can avoid to overload the remote system with too many calls. For most of the triggers that depend on external systems, a minimal interval must be at least PT30S. See ISO_8601 Durations for more information of available interval values.

maxDuration

  • Type: string
  • Dynamic:
  • Required:
  • Format: duration

The max duration waiting for new rows

It's not an hard limit and is evaluated every second

maxRecords

  • Type: integer
  • Dynamic:
  • Required:

The max number of rows to fetch before stopping

It's not an hard limit and is evaluated every second

password

  • Type: string
  • Dynamic:
  • Required:

server

  • Type: string
  • Dynamic:
  • Required:

stopAfter

  • Type: array
  • SubType: string
  • Dynamic:
  • Required:

List of execution states after which a trigger should be stopped (a.k.a. disabled).

username

  • Type: string
  • Dynamic:
  • Required:

Outputs

messagesCount

  • Type: integer
  • Required:

Number of message produced

uri

  • Type: string
  • Required:
  • Format: uri

URI of a kestra internal storage file

Was this page helpful?