Allow for multiple rabbit queues about broadway_rabbitmq HOT 9 CLOSED

FWIW, this is very useful for "sharding" queues as well, since RabbitMQ advises to have smaller queues. Broadway either lets you have many consumers connected to a single queue, or many consumers connected to a queue each (if you autodeclare the queue with queue: "").

from broadway_rabbitmq.

@whatyouhide i wonder if we should generalize your url/connection patch and allow it to returns a different queue too. This way you can have sharing or fault tolerance in the same API. We can complement the docs with more examples. WDYT?

from broadway_rabbitmq.

@josevalim or, we can generalize to make the whole options a function? I'm good with both things, but yeah this is easy enough to do if we make queue: fn index -> ... end

from broadway_rabbitmq.

We cannot make the whole options a function because some options may be handled by GenStage. One option is to have (merge_options: fun -> ... end), which we would call and merge into. WDYT?

from broadway_rabbitmq.

@josevalim hrm, merge_options could solve the :connection option as well as long as we do the merging as the first thing and then validate the options. Do you wanna go ahead with that route? I'm good both ways.

from broadway_rabbitmq.

Yup, let’s go this route then!

from broadway_rabbitmq.

Just wanted to thank you guys for turning this around so quickly!

It took me a little time to fully grok the solution, but for anyone reading this ticket here is an example;

    queues = ~w(messages messages2)
    queue_count = length(queues)

    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
          module: {
            BroadwayRabbitMQ.Producer, 
            queue: "",
            qos: [
              prefetch_count: 50
            ],
            merge_options: fn (index) -> 
              queue_name = Enum.at(queues, index)
              Logger.info(" connecting to queue #{queue_name}")
              [queue: queue_name] 
            end,
          },
          stages: queue_count
      ],
      processors: [
        default: [stages: 10]
      ],
      batchers: [
        elasticsearch: [stages: 1, batch_size: 400, batch_timeout: 500]
      ]
    )

from broadway_rabbitmq.

Awesome. I would only use Enum.fetch! instead of Enum.at so it raises if you give it less queues than stages.

from broadway_rabbitmq.

Great tip, thanks

from broadway_rabbitmq.