PSA: AMQP Declarations Changes about connect HOT 5 CLOSED

So I've got something very similar right now, all declarations are optional, you can declare a queue + N bindings on the input, and an exchange on the output. I haven't got fields for passive declarations, but I can add those if anyone requests it. Similarly I haven't got args support but I can add it in if someone requests it.

from connect.

Work is currently on https://github.com/Jeffail/benthos/tree/feature/amqp-declarations, I'll merge later today if I'm still happy with it.

from connect.

Didn't look at the changes yet, but since you asked for my thoughts, here we go.

AMQP input

Minimal configuration
Inputs are consumers, and consumers consume from queues exclusively, not exchanges.

Therefore, by default I would expect to be required to configure just a queue name, and be ready to start consuming. There is no reason to automatically create any exchanges or bindings, that's more relevant to the output type. I would expect to be able configure a few things related to the consumer, like the consumer tag (I would expect by default this to be set to something like benthos-[n]), the exclusive flag and qos (prefetch) config. In this scenario, we expect that some application is responsible for preparing the queue Benthos can consume from.

This should be enough, and should not attempt to make any structural modifications to the broker (keep in mind that the user might not even have permissions to declare queues or exchanges).

type: amqp
amqp:
  queue: benthos-queue
  url: amqp://guest:guest@localhost:5672/

Declaring the queue
As an extra option, Benthos could declare the queue for us. This would result in any producer application to just wait until the queue exists. Alternatively, Benthos could passively declare the queue, which acts more of an assertion that the queue exists and has specific settings. This would introduce some more settings like this:

type: amqp
amqp:
  queue: benthos-queue
  queue_declare: true
  queue_declare_passive: true
  queue_durable: true
  queue_exclusive: false
  url: amqp://guest:guest@localhost:5672/

Or maybe more structured, like this:

type: amqp
amqp:
  queue: 
    name: benthos-queue
    declare:
      passive: true
      durable: true
      exclusive: false
  url: amqp://guest:guest@localhost:5672/

Binding the queue to an exchange
Another separate feature would be creating the binding between the queue and an existing exchange (or the exchange that Benthos is configured to create, see below). The use case here would be that there already exists an (probably fan-out or topic) exchange that emits messages that Benthos is interested in and wants to tap into. This is where creating the binding with a specific routing key and properties comes in. Depending on the type of exchange, it would also be nice to be able to configure the binding arguments. For example for a headers exchange, this can be used to specify some filters. It would also be nice to be able to create multiple bindings while we're at it.

type: amqp
amqp:
  queue: 
    name: benthos-queue
    declare: true
  bindings:
    - exchange: fruity-events
      routing_key: ''
      args:
        x-match: all
        color: orange
        shape: round
    - exchange: fruity-events
      routing_key: ''
      args:
        x-match: all
        color: yellow
        shape: curved
  url: amqp://guest:guest@localhost:5672/

Declaring an exchange
Although I can't actually really think of a use case where we would need to declare a dedicated exchange for a Benthos input, this should be definitely be a separate option as well if you want to support it. I think, if there is a central object dedicated to publishing to Benthos, it might as well be a plain queue.

AMQP output

Minimal configuration
In AMQP land, you always publish to an exchange. You might think you're publishing directly to a queue, but in fact you're publishing to the default exchange, which routes the message to the queue specified in the routing key. This should be the default minimal configuration, so again, no creating exchanges, queues or bindings by default. This should be enough.

type: amqp
amqp:
  exchange: '' # Empty string resolves to the default exchange
  routing_key: benthos-messages  # This is the queue name we directly publish to
  url: amqp://guest:guest@localhost:5672/

Also declaring the exchange
Okay, this is the only scenario where declaring an exchange actually makes sense in my opinion. It's a good approach to let Benthos declare it's own exchange (or passively check that it exists with the correct settings), and not worry about which queues should be bound to it. Binding to the Benthos exchange should be the responsibility of whichever application is interested in actually receiving the messages.

type: amqp
amqp:
  exchange:
    name: benthos
    declare:
      passive: false
      durable: true
      type: fanout
      args: {}  # For some exchange types it would be nice to be able to specify the arguments field
  routing_key: ''
  url: amqp://guest:guest@localhost:5672/

Okay I think that's about it. Good luck :p

from connect.

Awesome, thanks @EdeMeijer. The new set up looks similar to what I have mocked except I'm missing the list of bindings, which is nice.

from connect.

Great job. I think I don't need either passive or args right now, so that's fine.

from connect.