Consume

Consume messages from an AMQP queue.

Requires maxDuration or maxRecords.

yaml
type: "io.kestra.plugin.amqp.Consume"

Examples

yaml
id: amqp_consume
namespace: company.team

tasks:
  - id: consume
    type: io.kestra.plugin.amqp.Consume
    url: amqp://guest:guest@localhost:5672/my_vhost
    queue: kestramqp.queue
    maxRecords: 1000

Properties

host *string

The broker host.

queue *string

The queue to pull messages from.

consumerTag string

Default Kestra

A client-generated consumer tag to establish context.

maxDuration string

Format duration

The maximum duration to wait for new rows.

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

maxRecords integerstring

The maximum number of rows to fetch before stopping.

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

password string

The broker password.

port string

Default 5672

The broker port.

serdeType string

Default STRING

Possible Values

STRINGJSON

Serializer / Deserializer used for the message.

username string

The broker username.

virtualHost string

Default /

The broker virtual host.

Outputs

count integer

Number of rows consumed.

uri string

Format uri

File URI containing consumed messages.

​Consume

Consume