rept

rept

How to avoid code duplication (newbie question)

Hi,

I currently have a RoR application that uses the Sneakers gem to monitor RabbitMQ queues and process messages. I’m looking into migrating this to Elixir and have started experimenting. We have about 20 queues, all with different logic that needs to be processed.

I was able to get Broadway up and running and want to create a worker per queue (or per 2 - 3 queues that share business logic). So I started creating workers like this:

defmodule MessageWorker do
  require Logger
  use Broadway

  @queue_names ~w(messages)

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwayRabbitMQ.Producer,
          queue: "",
          connection: [
            username: "guest",
            password: "guest",
            host: "localhost"
          ],
          merge_options: fn (index) ->
            queue_name = Enum.fetch!(@queue_names, index)
            Logger.info(" connecting to queue #{queue_name}")
            [queue: queue_name]
          end
        },
        concurrency: 1
      ],
      processors: [
        default: [
          concurrency: 10
        ]
      ]
    )
  end

  @spec handle_message(any, any, any) :: none
  def handle_message(_, message, _context) do
    Broadway.Message.ack_immediately(message)
    IO.inspect(message.data, label: "MESSAGE Got message")
  end

end

This all seems to work, which is great. However since the start_link will be the same for all workers (except for the queue name and maybe the number of processors I would like to put this ‘somewhere’ so I could avoid the code being duplicated in each worker. I just want each worker to specify the queue(s) to monitor and the concurrency. So the complete start_link method shouldn’t be duplicated.

How can I do this?

Marked As Solved

axelson

axelson

Scenic Core Team

I would recommend creating a helper module for this. Here’s how one might look:

defmodule MessageWorkerUtils do
 def broadway_options(module_name, queue_names) do
    [
      name: module_name,
      producer: [
        module: {BroadwayRabbitMQ.Producer,
          queue: "",
          connection: [
            username: "guest",
            password: "guest",
            host: "localhost"
          ],
          merge_options: fn (index) ->
            queue_name = Enum.fetch!(queue_names, index)
            Logger.info(" connecting to queue #{queue_name}")
            [queue: queue_name]
          end
        },
        concurrency: 1
      ],
      processors: [
        default: [
          concurrency: 10
        ]
      ]
    )
  ]
end

Then you could call it in your start_link like:

def start_link(_opts) do
  options = MessageWorkerUtils.broadway_options(__MODULE__, @queue_names)
  Broadway.start_link(__MODULE__, options)
end

And welcome to the forum! :wave:

Also Liked

rept

rept

Hi @axelson,

Thanks a lot! Tried it out and this works great.

Aetherus

Aetherus

I don’t know if this is a good idea, but it sounds like we could use a little bit metaprogramming.

defmodule MQWorker do
  @callback handle_message(any(), any(), any()) :: none()

  defmacro __using__(queue_names) do
    quote do
      @behaviour unquote(__MODULE__)

      def start_link(_opts) do
        Broadway.start_link(__MODULE__,
          name: __MODULE__,
          producer: [
            module: {BroadwayRabbitMQ.Producer,
              queue: "",
              connection: [
                username: "guest",
                password: "guest",
                host: "localhost"
              ],
              merge_options: fn (index) ->
                queue_name = Enum.fetch!(unquote(queue_names), index)  #<---- Replace `@queue_names` with `unquote(queue_names)`
                Logger.info(" connecting to queue #{queue_name}")
                [queue: queue_name]
              end
            },
            concurrency: 1
          ],
          processors: [
            default: [
              concurrency: 10
            ]
          ]
        )
      end
    end
  end
end

Then in each of your actual worker module, just use MQWorker, ["queue1", "queue2", ...] and implement handle_message/3

Last Post!

rept

rept

I tried the metaprogramming suggestion, works too and looks even cleaner!

Thanks.

I’ve added workers like this:

defmacro __using__({queue_names, workers}) do

....

      processors: [
        default: [
          concurrency: unquote(workers)
        ]
      ]

And now call it like this:

use MQWorker, {["messages2"], 10}

Seems to work.

Where Next?

Popular in Questions Top

rms.mrcs
Hi, I need to transform a list of numbers into a map where the keys are the indexes and the values are the original values of the list. ...
New
electic
Hi, I am new to Elixir. I am trying to use the DateTime component to insert a date into MySQL however the there seems to be no way to fo...
New
minhajuddin
I have seen a lot of code which picks the first element from a list using Enum.at(0) instead of List.first. Is there a reason why people ...
New
Brian
What is the proper way to load a module from a file in to IEX? In the python world, doing something like this pretty standard: from ....
New
openscript
Hello! Sorry for this astonishing simple question, but I’m really stuck. I try to set up the intellij-elixir plugin, but I don’t know ho...
New
Fl4m3Ph03n1x
About me? ( if you have nothing better to do than reading about some random guy in the internet :stuck_out_tongue: ) Hello all, this is ...
New
WestKeys
Currently suffering from paralysis by [HTTP client] analysis. This is rather unusual in Elixirland as there tends to be consensus on the ...
New

Other popular topics Top

ashish173
I am using Ecto timestamps with postgres, I can see the timestamps() use the :naive_dateime but for my use case I wanted to store the ti...
New
sen
Hi All, I set a environment variables in dev.exs , like below code. when i start server, how can i set the ${enable} value? thanks. d...
New
shijith.k
I am trying to start a new phoenix project with elixir 1.9, but mix phx.new does not work. It says that ** (Mix) The task "phx.new" could...
New
msaraiva
Surface is an experimental library built on top of Phoenix LiveView and its new LiveComponent API that aims to provide a more declarative...
564 44139 214
New
Patoshizzle
After calling mix ecto.create I get this error: 17:00:32.162 [error] GenServer #PID&lt;0.412.0&gt; terminating ** (Postgrex.Error) FATAL...
New
Harrisonl
We have an ECS cluster with 4 services, where each task joins a single cluster, via discovery ECS discovery service. Currently when I de...
New

We're in Beta

About us Mission Statement