DayDreamer

DayDreamer

Broadway's ProducerConsumer stage unable to hook up with Broadway's Producer.

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,

Where Next?

Popular in Questions Top

srinivasu
How to handle excepions in elixir? Suppose i have A, B, C ,D, E modules. and each module has get() function. A.get() method will call th...
New
dotdotdotPaul
Okay, I'm having a heck of a time trying to figure out how to best handle the validation of belongs_to associations in Ecto. I'm sure I'...
New
sergio_101
I am VERY much an elixir newbie. I have taken one elixir course and one phoenix course on Udemy. During that course, I saw the instructor...
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
gonzofish
I’m currently trying to understand how to join three tables using Ecto. All the examples I’ve seen use 2, so maybe I’m just missing somet...
New
ovidiubadita
Hey all, I discovered Elixir and I love it. I always wanted to learn a functional programming and I intended to go for Haskell, but afte...
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
wernerlaude
In AR this is so simple @articles = current_user.articles How to do in Ecto? def index(conn, _params) do current_user = conn.assig...
New
Qqwy
Original source of discussion: This topic on the Pragmatic Programmers' Functional Web Development with Elixir, OTP, and Phoenix forum. ...
New
jay1
Why is it that the mnesia database isn’t the most preferred database for use in Elixir/Phoenix?
New

Other popular topics Top

yurko
Here are few pieces of (common) Linux knowledge that we use for reasonably small one server apps. We use Ubuntu but this should work for ...
New
joaquinalcerro
Hi there, I am working with Ecto-Postgresql and I need to call all of the records from a specific table but the table has 40,000 record...
New
aalberti333
As the title describes, I’m trying to run Enum.map() over a list of key/value pairs, where the value is a map. My data looks like this: ...
New
sergio_101
I am VERY much an elixir newbie. I have taken one elixir course and one phoenix course on Udemy. During that course, I saw the instructor...
New
New
vertexbuffer
Hello, can anybody help here..? I have a list of players and I what to delete an element, but every for loop the list is reverting to ori...
New
chrismccord
As promised, the first release candidate of Phoenix 1.3.0 is out! This release focuses on code generators with improved project structure...
New
myronmarston
The Elixir Typespec docs show the following syntax for keyword lists in typespecs: # ... | [key: type] # keyword lis...
New
Patoshizzle
After calling mix ecto.create I get this error: 17:00:32.162 [error] GenServer #PID<0.412.0> terminating ** (Postgrex.Error) FATAL...
New
belgoros
I’m not a pro in using Regex and can’t figure out why the following behaviour happens, especially if we take into account the difference ...
New

We're in Beta

About us Mission Statement