WorkflowQueue FIFO with put() and take()

Community Support
1

With debugging messages, it appears to me if I use WorkflowQueue.put() to produce items from a list and then use WorkflowQueue.take() to consume(), it processes the item in a reverse order from the original order of the input list? Could someone confirm if that is the case and how can I get FIFO from a WorkflowQueue? Thanks

2024-12-20 00:29:29,579 INFO i.t.i.l.ReplayAwareLogger.info 234 - Adding: JiraStateChange(id=1, jiraKey=DSCE-917, newState=New Test Request, timestamp=2024-12-18 03:57:17.0)
2024-12-20 00:29:29,581 INFO i.t.i.l.ReplayAwareLogger.info 234 - Adding: JiraStateChange(id=2, jiraKey=DSCE-917, newState=Aborted, timestamp=2024-12-18 03:59:17.0)
2024-12-20 00:29:29,582 INFO i.t.i.l.ReplayAwareLogger.info 234 - Adding: JiraStateChange(id=3, jiraKey=DSCE-917, newState=New Test Request, timestamp=2024-12-19 22:55:57.0)
2024-12-20 00:29:29,585 INFO i.t.i.l.ReplayAwareLogger.info 234 - Handling event: JiraStateChange(id=3, jiraKey=DSCE-917, newState=New Test Request, timestamp=2024-12-19 22:55:57.0) // get the last one first

        List<JiraStateChange> stateChanges = jiraActivity.getQueuedUnprocessedEvents(
                getJiraKey());
        log.info("{} queued events were loaded: {}", stateChanges.size(), stateChanges);
        stateChanges.forEach(event -> {
            log.info("Adding: " + event);
            eventQueue.put(event);
        });

        entity.setState(RequestEntity.State.Running);
        while (entity.getState() == RequestEntity.State.Running && !signalStopping) {
            // Wait for an item in the queue or for an interruption signal
            Workflow.await(() -> (eventQueue.peek() != null) || signalStopping);

            if (signalStopping) {
                log.info("Workflow interrupted. Exiting...");
                break;
            }

            // Process item from the queue
            JiraStateChange event = eventQueue.take();
            log.info("Handling event: " + event);

…

2

Could someone confirm if that is the case and how can I get FIFO from a WorkflowQueue? Thanks

WorkflowQueue is backed by Dequeue which supports both fifo and lifo. That said, WorkflowQueue put method calls Dequeue addLast, and take uses Dequeue poll (remove from head) so its fifo.

3

I should have used poll() which takes from the head, instead of take() which from the end, for FIFO. Thanks

@Override
public E poll() {
if (queue.isEmpty()) {
return null;
}
return queue.remove();
}

@Override
public E take() {
WorkflowThread.await(“WorkflowQueue.take”, () → !queue.isEmpty());
// this implementation is incorrect and has been fixed in WorkflowQueueImpl
return queue.pollLast();
}