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,
Popular in Questions
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
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
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
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
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
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
Currently suffering from paralysis by [HTTP client] analysis. This is rather unusual in Elixirland as there tends to be consensus on the ...
New
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
Original source of discussion: This topic on the Pragmatic Programmers' Functional Web Development with Elixir, OTP, and Phoenix forum.
...
New
Why is it that the mnesia database isn’t the most preferred database for use in Elixir/Phoenix?
New
Other popular topics
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
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
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
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
I have an umbrella app.
Some of the apps inside depend on other apps in the umbrella, unsurprisingly.
I'm writing a test for one of t...
New
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
As promised, the first release candidate of Phoenix 1.3.0 is out! This release focuses on code generators with improved project structure...
New
The Elixir Typespec docs show the following syntax for keyword lists in typespecs:
# ...
| [key: type] # keyword lis...
New
After calling mix ecto.create I get this error:
17:00:32.162 [error] GenServer #PID<0.412.0> terminating
** (Postgrex.Error) FATAL...
New
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







