I need an help for my following topic.
I want to know when an instance of flows is finished.
I explain : I have a flow (I call it here sendMessages). The flow send 100 messages in UM with a loop statement.
We have something like this :
- pub.flow.debugLog(message = init)
- myPackages.createAnArrayOf100UUID (output = messages (array))
- LOOP on messages
– pub.jms.send(Connection = ConnectionJMS, Topic = Topic, JMSMessage contaie messages) - MAP (Drop)
I have a Trigger on Topic which execute a flow (I call it here receiveMessages)
The fact is the Trigger send 30 messages in concurrent threads.
I want send a mail (for instance) when all messages are executed in receiveMessages.
In previous vversion I saved messages in database and read database each time in receiveMessages. But it required to read database (and write for each messages).
So I wonder if I could use cache.
My idea is the following
In sendMessages
…
- pub.cache.put(key = UUID, value = MessagesNumber)
…
– pub.jms.send (UUID in JMS Message)
…
In receiveMessages
- MAP (JMSMessages → UUID)
- pub.cache.get(key = UUID, value → MessagesNumber)
- pub.math.addInt(MessagesNumber = MessagesNumber - 1)
- pub.cache.put(key = UUID, MessagesNumber → value)
… - BRANCH on MessagesNumber
– SEQUENCE Label = 0
---- MAP (What I have to do when the flow is finished)
---- pub.cache.remove(key = UUID)
That’s work when my trigger process in serial and not in concurrent.
In conccurent process, that doesn’t work because the time to decrement MessagesNumber and put it in cache, another thread changes the value and 0 is never reached.
So I use pub.cache.acquiredLock and pub.cache.releaseLock flow. But it doesn’t work to because I need to lock get, decrement and put in the receiveMessages flow. Therefore, we need to specify the lock type (read or write) or I need the lock for the two types.