conduitframework / conduit Goto Github PK

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:

• Map based specification
• Module based supervisors

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:

• outgoing => incoming
• to: "application_name.queue_name" => from: "application_name.queue_name"

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 arg
• ConduitAMQP 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