Hello Everyone,

I am new in elixir and Broadway and want to setup a data processing pipeline with RabbitMQ Consumer (Filter stage) --> Format stage (Broadway ProducerConsumer) --> Tag stage(Broadway Consumer).
I got the Filter stage correct however, for the Format stage Filter stage should be the producer however, this change does not work.

Filter stage RabbitMQ consumer:

defmodule MyApp.Filter do
  use Broadway

  def start_link(_) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {
          BroadwayRabbitMQ.Producer,
          queue: "ingress",
          declare: [durable: true],
          bindings: [{"events", []}],
          connection: [host: "localhost"]
        }
      ],
      processors: [
        default: [concurrency: 1]
      ]
    )
  end

  def handle_message(_, message, _) do
    IO.puts("Received message: #{message.data}")
    message
  end
end

Format stage:

defmodule MyApp.Formatter do
  use Broadway

  alias Broadway.Message

  def start_link(_) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {MyApp.Filter}  # this requires "args" however, [] or {} or nil does not work 
# main question how to make this ProducerConsumer work with BroadwayProducer(Filter).

      ],
      processors: [
        default: [concurrency: 1]
      ]
    )
  end

  def handle_message(_, %Message{data: data} = message, _) do
    # Example processing
    transformed_data = String.upcase(data)
    IO.puts("Processing message: #{transformed_data}")
    %Message{message | data: transformed_data}
  end
end

I am not sure what args should look like so that this stage will work

application.ex

defmodule MyApp.Application do
  use Application

  impl true
  def start(_type, _args) do
    children = [
      {MyApp.Filter, []},
      {MyApp.Formatter, []}
    ]

    opts = [strategy: :one_for_one, name: MyApp.Supervisor]
    Supervisor.start_link(children, opts) end
end

mix.exs deps

{:broadway, "~> 1.0"},
{:broadway_rabbitmq, "~> 0.7"}

Could someone point out what am I missing

Cheers,