Rebuild RabbitMQ node and trigger #3244
Conversation
* Use amqp-connection-manager to automatically reconnect * Support options on messages * Add ability to bind a queue to an exchange * Use prefetch on queues to prevent overloading n8n
michael-radency
added
node/improvement
| // TODO should we nack it? It probably won't succeed on a reattempt, but it wasn't successful either | ||
| // FIXME how do we report a failure to the ux? | ||
| if(options?.nack) { | ||
| channel.nack(message, false, false); |
There was a problem hiding this comment.
I'm not sure if this should be used. Normally you'd nack to put it back in the queue so it's reattempted, but would there be a case where that would be helpful? If it errors out on the message parsing, it's not going to succeed on a second attempt right?
If we could nack the message if the workflow failed somewhere down the line, that'd probably be ideal, but I couldn't see a way to do that.
There was a problem hiding this comment.
I do actually not putting it back in queue but rejecting, second false. There is this PR related to what you are suggesting, maybe you can take a look at
There was a problem hiding this comment.
Oh, nice. I didn't catch that. In that case, I don't see a reason that you wouldn't want to nack it, so you probably don't need the option. It should just always send that nack if it can't parse a message.
Does that deferred promise on the emit normally get the response of the last node in the workflow, and reject it if the workflow fails anywhere? I can take a look at implementing something like that for the emit later, but it's probably not needed to merge this in.
There was a problem hiding this comment.
I'm not sure if this should be used. Normally you'd
nackto put it back in the queue so it's reattempted, but would there be a case where that would be helpful? If it errors out on the message parsing, it's not going to succeed on a second attempt right?If we could nack the message if the workflow failed somewhere down the line, that'd probably be ideal, but I couldn't see a way to do that.
Just to be sure I understand: the aim would be to be able to wait until the end of the workflow, like for the webhook trigger
and based on the workflow's result (or some data returned by the last node in the workflow), decide to ack or nack the message, right?
if I understood it right, yes @Ken-Michalak that would be awesome if you could take a look and super useful for our use case
you can run |
|
Any update on this? We also need the feature to bind a queue to an exchange. |
|
Hello @Ken-Michalak I made update to this PR due to RabbitMQ node changes |
|
@michael-radency I wasn't following what happened with the As an aside, we've been using the code from my original PR in production (with a different name), and we haven't been having any issues. The original nodes were not usable because of the reconnect and prefetch issues. |
|
@Ken-Michalak I was trying to keep changes to current nodes to a minimum, avoid breaking changes and use only amqplib. if it wouldn't be possible I will add amqp-connection-manager back. |
|
@michael-radency Yeah, amqplib doesn't do anything on errors but fire events. You'd have to listen for the right error events, and then recreate the connection, consumer, queues, etc, which is what the connection manager does. There's also supposed to be only 1 connection with many consumers, but that's not how the old node was working. |
| const connections = getAllConnections(); | ||
| connections.forEach(connection => connection.close()); |
There was a problem hiding this comment.
I think this is going to close every RabbitMQ connection when one workflow (actually one node) is stopped. The connections are 1 per set of credentials and running process, and should only be closed when the entire n8n process shuts down. (You could attempt to track the number of consumers per connection and shut it down, but I thought that would be difficult to track accurately and could lead to bugs.)
https://www.cloudamqp.com/blog/part1-rabbitmq-best-practice.html#connections-and-channels
When you have multiple workflows with rabbitmq nodes and multiple workers, you don't want to open and close a connection for every node, which is why there's a global map of connections.
There was a problem hiding this comment.
Hi @Ken-Michalak , I will make it as an option
|
@Ken-Michalak @michael-radency Hi guys, any update on this PR? |
|
@mirobertod Sorry, I'm not a maintainer, so it's out of my hands at this point. I'm not really sure what's going on either. |
|
Hello @mirobertod , @Ken-Michalak this PR is currently on review process, right now I can not give any ETA, but I will let you know of any updates as soon as possible |
|
Hi guys, any update on this PR? |
|
@mirobertod The issue of this PR has been fixed by those PR #4019 & #3385. I will close this PR if there is no additional feature wanted. |
|
@agobrech There are still a few other things from that pr that are missing though. I added the ability to subscribe to an exchange or topic. What's in there now was pretty confusing at first because you can send messages into an exchange, and the trigger says, "queue / exchange" but it only actually works with a queue. The only way to get messages from a topic is to manually create the queue and binding outside of n8n, or by using an HTTP request node attached to a trigger for workflow activated. The pr also adds the ability to set separate options on messages besides headers, like priority, or content type, which can use expressions so they're different per item. (The trigger looked at the content type header and if it was json, it would parse it without needing the option set.) That might be a part of #2806 but with RabbitMQ a TTL can be set as an argument on the queue, and/or on specific messages. The current node does open and close the connection each time it publishes messages, which goes against RabbitMQ's best practices. I think I fixed in that pr, but I'm not sure how much of a performance improvement that really has. |
continuation of #2921 by Ken-Michalak