msw10100

msw10100

Can Conduit handle dynamically-named queues?

We have a deployment pattern that launches multiple processors in parallel. There’s usually a single work queue to which all instances subscribe, so they can load-balance request handling. Each instance also creates and subscribes to a couple of unique “instance” queues of their own, with special subscriptions for processor management layer, and to allow clients to interact with a specific instance as needed. These instance queues are named at create time with a unique name issued by our fabric service manager.

We use topic exchanges with some dynamic bindings on the dynamically-created queues, as well. Can subscribers be dynamically created and destroyed with custom bindings, or is it all declaratively described at compile time in the Broker?

Perhaps instead of defining the Broker in source and starting it as an application child, I could instead define it as a string at application startup, interpolating in all the dynamic info that I’ve received from my microservice manager, and use Code.eval in some fashion to compile and load it into memory and then start it?

Cheers … @msw10100

1
Post #11
blatyo

blatyo OP

Conduit Core Team

Yea, Conduit works fine with stateful things. I’ve primarily used it to interact with a database using ecto or in some cases ETS. There’s nothing that would stop you from communicating with other processes though, if that’s where you keep your state.

Yep, that’s how you would do it. RabbitMQ has RPC built in. Direct Reply-To | RabbitMQ

Someone else was asking about RPC recently and there wasn’t a way to set a setting necessary to use RabbitMQ’s RPC. However, I just released a new version of ConduitAMQP (0.6.3) that does support that a couple minutes ago. If you use it and discover something that could be more ergonomic, let me know.

Yes, you can pass functions for the queue names. It’s not well documented, but there are tests for it. The function doesn’t have to be defined inline, you can use a module function by passing &MyModule.queue_name/0.

https://github.com/conduitframework/conduit/blob/master/test/conduit/broker_test.exs#L50
https://github.com/conduitframework/conduit/blob/master/test/conduit/broker_test.exs#L76
https://github.com/conduitframework/conduit/blob/master/test/conduit/broker_test.exs#L89

That should work fine as long as you have some function that can get that name.

Everything is created at application start, but depending on how you specified your config (inline vs. a function vs. application config), some of it may be set at compile time. You cannot currently create a subscriber or queue whenever you want. That will always happen when the broker starts with whatever you’ve specified in your broker. I don’t understand the situation where you would want to do something like that. It’s possible that could be supported, but it’s quite a bit of work, so I’d want to understand that better. Maybe you could explain it to me here or in a DM?

That might be possible, but I would hope that isn’t necessary.

2
Post #12
blatyo

blatyo OP

Conduit Core Team

ConduitAMQP v0.6.3 was just released!

This release adds the ability to set consume options for subscribers consuming from a queue. You can see the options here. This enabled a couple use cases, but the one that prompted it was RabbitMQ RPC.

2
Post #13
msw10100

msw10100

What’s the concurrency model for Conduit Framework? I see that each subscriber invocation is a separate process for each message, much like Phoenix spawns a process per request. How many subscriber instances run at one time for a given connection/channel/etc…?

0
Post #14
blatyo

blatyo OP

Conduit Core Team

For ConduitAMQP:

  • There’s a connection pool, where you configure the size.
  • There’s a channel pool, where you configure the size. These are used for publishing messages. They are created using the connection pool.
  • There’s a channel per consuming queue. They are created using the connection pool.
  • Every received message in the system will have a separate subscriber process. You can use the AMQP settings to limit the number of messages in flight per queue. Either prefetch_size or prefetch_count.
4
Post #15
blatyo

blatyo OP

Conduit Core Team

ConduitSQS v0.2.7 was just released!

This release handles an annoying issue with hackney, where it’s leaking a message for the connection being closed. It was happening occasionally and causing some of the processes in ConduitSQS to crash. Note that it didn’t stop messages from getting processed.

4
Post #16
dimitarvp

dimitarvp

Semi-random question: have you considered using gun instead of hackney? I’ve been watching a few tutorials on it lately and I kind of like it but not sure yet if it’s a better interface towards the Erlang HTTP clients.

1
Post #17
blatyo

blatyo OP

Conduit Core Team

conduit_sqs wraps ex_aws_sqs, which uses hackney by default. Because you can configure that with ex_aws_sqs, you can technically use anything.

I haven’t really had more than a cursory glance at gun to be honest. Maybe ask in this thread: HTTP client libraries and wrappers - #12 by chulkilee

1
Post #18
blatyo

blatyo OP

Conduit Core Team

Conduit v0.12.10 was just released!

This fixes a bug in the retry plug, when nacking a message. Also fixes a deprecation warning when running on Elixir 1.8.

3
Post #19
blatyo

blatyo OP

Conduit Core Team

ConduitSQS v0.3.0 was released recently!

This release upgrades gen_stage used internally.

5
Post #20