Edit

Waiting for Transactions

A commonly encountered task in ODF 2 is waiting for the completion of all transactions in a given Business Process. There is no out-of-the-box solution that fits all use cases. For example, a Monitor Task often serves as the source of transactions, making it impossible to know when new ones appear. However, once all requirements are known, such a feature is quite easy to implement.

The proposed solution relies on the following logic:

• The solution is a task placed at the end of the Business Process.
• A Monitor Task, if present, can be checked for completion.
• PoolObjectRunner can be used to synchronize a code block to process incoming transactions sequentially.
• A Transaction has a status field used to track its completion.

For details, see the example below:

@BotTask
class WaitForTransactionsTask implements AdHocTask /* just for example - in practice use any convenient abstraction */ {

    private static final String COMPLETED_STATUS = "COMPLETED"; // for example

    private static final String MONITOR_TASK_ID = "MyMonitorTask"; //  must be the same as monitor task`s getMonitorId() method returns
    private static final MonitorIdentity MONITOR_IDENTITY = new MonitorPerTask(); // must be the same as monitor task uses

    private static final String POOL_KEY = WaitForTransactionsTask.class.getName();

    private final PoolObjectRunner poolObjectRunner;
    private final MonitorStateRepository monitorStateRepository;
    private final CurrentTransaction currentTransaction;
    private final TransactionRepository transactionRepository;
    private final UUID bpRunId;

    private final AtomicBoolean allTransactionsCompleted = new AtomicBoolean();

    public WaitForTransactionsTask(PoolObjectFactory poolObjectFactory, MonitorStateRepository monitorStateRepository,
            CurrentTransaction currentTransaction, TransactionRepository transactionRepository,
            @BpRunId UUID bpRunId) {

        this.poolObjectRunner = poolObjectFactory.runner(POOL_KEY);
        this.monitorStateRepository = monitorStateRepository;
        this.currentTransaction = currentTransaction;
        this.transactionRepository = transactionRepository;
        this.bpRunId = bpRunId;
    }

    @Override
    public TaskRunnerOutput run(TaskInput taskInput) {

        if (currentTransaction.isPresent()) {
            poolObjectRunner.execute(Collections.singletonList(POOL_KEY), k -> checkIfAllTransactionsAreCompleted());
        }

        if (allTransactionsCompleted.get()) {
            // do something
        } else {
            // do something
        }
    }

    private void checkIfAllTransactionsAreCompleted() {
        final Transaction transaction = currentTransaction.get();
        transaction.setStatus(COMPLETED_STATUS); 
        transactionRepository.update(transaction);
        
        if (isMonitorCompleted()) {
            allTransactionsCompleted.set(isNoUncompletedTransactionsPresent());
        }
    }

    private boolean isMonitorCompleted() {
        try {

            final long count = monitorStateRepository.getDao().queryBuilder()
                    .where()
                    .eq(OdfVariationEntity.VARIATION_ID_COLUMN, monitorStateRepository.getVariation().getId()) // important to maintain isolation of variations 
                    .and()
                    .eq(MonitorStateEntity.MONITOR_ID_COLUMN, MONITOR_IDENTITY.getMonitorStateId(MONITOR_TASK_ID, bpRunId.toString()))
                    .and()
                    .eq(MonitorStateEntity.STOPPED_COLUMN, Boolean.TRUE.toString())
                    .countOf();

            return count < 1;

        } catch (SQLException e) {
            throw new IllegalStateException(e);
        }
    }

    private boolean isNoUncompletedTransactionsPresent() {
        try {
            final long count = transactionRepository.getDao().queryBuilder()
                    .where()
                    .eq(OdfVariationEntity.VARIATION_ID_COLUMN, transactionRepository.getVariation().getId()) // important to maintain isolation of variations 
                    .and()
                    .eq(Transaction.START_BP_UUID_COLUMN, bpRunId)
                    .and()
                    .ne(Transaction.STATUS_COLUMN, COMPLETED_STATUS)
                    .countOf();

            return count < 1;

        } catch (SQLException e) {
            throw new IllegalStateException(e);
        }

    }

}