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?
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.
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.
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.
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.
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…?
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.
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.
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.
