conduitframework / conduit Goto Github PK
View Code? Open in Web Editor NEWA message queue framework, with support for middleware and multiple adapters.
Home Page: https://hexdocs.pm/conduit
License: MIT License
A message queue framework, with support for middleware and multiple adapters.
Home Page: https://hexdocs.pm/conduit
License: MIT License
We have some situations in our stack where we want to publish a plethora of messages and have the queue slowly drain. Currently, if we publish let's say 100k messages, our Broker will pull off as many as it possibly can in parallel and things grind to a halt. (Imagine if each message is extremely heavy to process, we are ok if it takes an hour to drain, but it's not ok if all 500 are attempted at the same time, for CPU reasons). In fact, when this happens, we notice a separate bug that we are trying to get steps to reproduce on where messages are "stuck" in processing - we have to kill the connections holding those messages to "un stick them" so they can be handled.
If this behavior is surprising to you, we'd love to hear there's a way to stop this from happening. If not, would you be open to a PR where we write some Middleware to essentially throttle the Broker (and perhaps you could even point us in the right starting direction?).
Thanks for all your help!
When you publish to something not declared in broker.ex
, like:
Broker.publish(:foo, ...)
you get a stack trace like:
** (MatchError) no match of right hand side value: nil
code: MyService.call(payload)
stacktrace:
(my_app) web/broker.ex:1: MiddleOut.Broker.publish/3
It'd be great to instead get an error saying you need to add publish :foo, ...
to broker.ex
Hello! We just updated from 0.12.5 to 0.12.8 and started seeing this:
03:36:45.524 [warn] Calling ApiQueue.Broker.publish/3 with message as second argument is deprecated to enable pipeline usage.
Replace:
ApiQueue.Broker.publish(:error, message, opts)
With:
ApiQueue.Broker.publish(message, :error, opts)
We don't actually publish, so I'm thinking this might have to do w/ Conduit.Plug.DeadLetter
?
Any pointers to get on top of this?
Can you start keeping a changelog? (https://keepachangelog.com/)
I'm on 0.10.4
and I see 0.11.0
was released recently, but there's no (easy) way for me to see what changed and if it's safe for me to upgrade.
Thanks!
Forgive the basic question, but I've poured over the docs for Conduit and ConduitAMQP and I can't seem to find a usage example. The generator was able to get me far enough that I know how to consume a message (I've tested this by manually publishing a message via the rabbitmq dashboard). But how do I publish a message? I see the outgoing
macro in the DSL, but I can't execute Broker.outgoing()
or Broker.publish()
.
Am I correct in understanding that Conduit does not support multiple subscribers to the same message?
Hello,
Supervisor.Spec was deprecated in Elixir v1.5.
Would be cool to update the docs to suggest the new behavior.
There is two ways to add the broker into our application supervision tree:
The following code shows the difference:
Current way (deprecated)
Also, add your broker to the supervision hierarchy in your example.ex:
def start(_type, _args) do
children = [
# ...
supervisor(Example.Broker, [])
]
supervise(children, strategy: :one_for_one)
end
New way (Map based specification)
Also, add your broker to the supervision hierarchy in your example.ex:
def start(_type, _args) do
children = [
# ...
%{
id: Example.Broker,
start: {Example.Broker, :start_link, []}
}
]
supervise(children, strategy: :one_for_one)
end
Hi Allen.
Thanks for the Conduit framework. I think it'd be worthwhile to cut a new release, one that would include Conduit's runtime configurability as added in 3e9ca0f.
Hello,
I noticed that running mix conduit.gen.subscriber user_created
isn't generating the expected output.
The example should teach how to configure a subscriber on broker, instead it's showing how to create a publisher. The generated subscriber code is looking good though.
Basically the generator needs two changes:
Current state:
In an outgoing block in your ExampleQueue.Broker add:
subscribe :user_created, UserCreatedSubscriber, to: "example.user_created"
Expected state:
In an incoming block in your ExampleQueue.Broker add:
subscribe :user_created, UserCreatedSubscriber, from: "example.user_created"
Can't do plug :put_destination, &(&1.source <> ".error")
in Broker for example.
Hello, I'm new to elixir but learning quickly. I followed the installation instructions to the letter, but I can't start my application once I append {DataDigestQueue.Broker, []}
to the children
list in lib/data_digest/application.ex
.
The error message when I run mix phx.server
is:
** (Mix) Could not start application data_digest: DataDigest.Application.start(:normal, []) returned an error: shutdown: failed to start child: DataDigestQueue.Broker
** (EXIT) shutdown: failed to start child: ConduitAMQP
** (EXIT) an exception was raised:
** (UndefinedFunctionError) function ConduitAMQP.start_link/1 is undefined or private
(conduit_amqp) ConduitAMQP.start_link([DataDigestQueue.Broker, [{:queue, "data_digest.jobs", []}], %{jobs: [from: "data_digest.jobs"]}, [adapter: ConduitAMQP, url: "amqp://guest:guest@localhost:5672"]])
(stdlib) supervisor.erl:379: :supervisor.do_start_child_i/3
(stdlib) supervisor.erl:365: :supervisor.do_start_child/2
(stdlib) supervisor.erl:349: anonymous fn/3 in :supervisor.start_children/2
(stdlib) supervisor.erl:1157: :supervisor.children_map/4
(stdlib) supervisor.erl:315: :supervisor.init_children/2
(stdlib) gen_server.erl:374: :gen_server.init_it/2
(stdlib) gen_server.erl:342: :gen_server.init_it/6
(stdlib) proc_lib.erl:249: :proc_lib.init_p_do_apply/3
That suggests to me that Conduit.Broker
is trying to start ConduitAMQP
with the wrong number of arguments (1 keyword list instead of it spread out among 4 arguments). I can see where this is likely happening in the library, but am surprised that I don't see any other reports of this issue so I assume it's user error after all.
Conduit.Broker
tries to start the adapter with one argConduitAMQP
expects four args (though so does ConduitSQS
)In the meantime, I'm able to get it work by modifying deps/conduit_amqp/lib/conduit_amqp.ex
to add:
def start_link([broker, topology, subscribers, opts]) do
start_link(broker, topology, subscribers, opts)
end
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.