Control enabled of AMQP-in by msg.enabled or at specified intervals like a cronjob

UNOPARATOR commented 4 years ago

What I would like to achieve is this:

If an error occurs during the processing of a message from an AMQP in Manual ACK node, I return the flow with the original msg to ACK the message and using an "AMQP out" to move the message to an error queue.
- Retry the message in the error queue by moving them to the original queue as a cron job (every X hours, at specific times etc) by enabling an "AMQP in" consuming the error queue but only enabled by the cron job's trigger.
If no error is occured just ACK the message and be done with it.

What do you think @amodelbello ?

UNOPARATOR commented 3 years ago

bump?

amodelbello commented 3 years ago

Hi @UNOPARATOR, Apologies for the super late response. This sounds like a cool idea. Did it work for you?

UNOPARATOR commented 3 years ago

No problem.

Since you haven't released a new version, the current version doesn't allow me to do that (enable/disable the AMQP-in node). I'm not a node.js guy, so providing a PR for this would be difficult for me which was the reason for asking it in the first place (I hoped it might get picked if it would be beneficial for you too in any way).

UNOPARATOR commented 3 years ago

@pauldeng, would you like to have a go at this issue too? ;)

pauldeng commented 3 years ago

I am not quite understand this requirement.

My understanding is you try to limit the rate of retry a failed message. With the nack feature, you can simply delay the nack to the interval you prefer. After the interval, failed message is nack and requeued.

Please give my nack demo a try. In this demo, message re-try is set to every 20 seconds (the delay node): nack-demo

UNOPARATOR commented 3 years ago

That wouldn't be covering my need to use a queue as a batch job queue: (every X hours, at specific times etc) You might think it as scheduled processing at specific times only which would make the messages in the queue consumed only for a brief period. Whatever the usage is, what I need is to be able to change the enabled property (or a new property that would disconnect the consumer if set to false or connect the consumer if set to true) of the consumer node, so that until the consumer is enabled again, the new messages would keep on waiting on the queue.

amodelbello commented 3 years ago

@UNOPARATOR I may be misunderstanding your use case here but couldn't what you describe be accomplished with a simple Delay node? If an error occurs during message processing just send it back to the original flow via an amqp out node but with a delay in front of it.

pauldeng commented 3 years ago

I think what you want is manually control the node to connect or disconnect from RabbitMQ server.

The amqp-in-manual-ack node would be the best candidate to have this feature. Perhaps, when it receives a msg.manualConn, it will connect/disconnect accordingly. You need to setup your own cron outside of this node to trigger this message.

Is it make sense to you? @UNOPARATOR

UNOPARATOR commented 3 years ago

@pauldeng, that is exactly what I tried to explain. But the amqp-in-manual-ack node needs the original message to ACK the message as input. According to your example if msg.manualConn is added and used, then this should be set in the last message to disconnect the connection, right? What if the last message fails? Too bad, Node-RED doesn't allow multiple inputs for nodes.

UNOPARATOR commented 3 years ago

@UNOPARATOR I may be misunderstanding your use case here but couldn't what you describe be accomplished with a simple Delay node? If an error occurs during message processing just send it back to the original flow via an amqp out node but with a delay in front of it.

Come to think of it that might work, but wouldn't that mean message loss in case of a Node-RED restart?

pauldeng commented 3 years ago

Well, not exactly has to be the original message. At the moment, the manual-ack node is trying to stay connected. We can set a flag there to stop reconnecting if a msg.manulConn is injected. Perhaps it could be a solution. Unfortunately, I am not that familiar with that part of code and @amodelbello would be the best person.

If Node-Red restart, you will not lose any message. As not acked message will be requeue when Node-Red disconnect from RabbitMQ.

amodelbello commented 3 years ago

@UNOPARATOR I don't think messing with the node's connection is the right way to go. I would suggest trying to filter or limit the amqp-in node's input instead.

UNOPARATOR commented 3 years ago

If what you are saying is that it would be problematic in general (the underlaying modules do not support such or etc.) or it would cause erratic behaviour when messing with the node's connection then I'll try to think of some other approach.

In my case, the amqp-in node is not present, the queue is populated from another API (not by Node-RED), that's the reason limiting would not be appropriate/possible.

amodelbello commented 3 years ago

Ah okay. Then maybe think of some other way to programmatically limit or filter messages before they get to the queue? Using the node's connection itself as the condition for whether a message goes through just strikes me as a hack, basically using the wrong tool for the job.

UNOPARATOR commented 3 years ago

Thanks for your time & valuable inputs. I'm closing this issue now.

MeowWolf / node-red-contrib-amqp

Control enabled of AMQP-in by msg.enabled or at specified intervals like a cronjob #26