ExternalTaskTest.java

import org.camunda.bpm.engine.ExternalTaskService;
import org.camunda.bpm.engine.ProcessEngine;
import org.camunda.bpm.engine.RuntimeService;
import org.camunda.bpm.engine.delegate.DelegateExecution;
import org.camunda.bpm.engine.delegate.JavaDelegate;
import org.camunda.bpm.engine.externaltask.LockedExternalTask;
import org.camunda.bpm.engine.impl.cfg.StandaloneInMemProcessEngineConfiguration;
import org.camunda.bpm.engine.runtime.ProcessInstanceWithVariables;
import org.camunda.bpm.engine.variable.Variables;
import org.camunda.bpm.model.bpmn.Bpmn;
import org.camunda.bpm.model.bpmn.BpmnModelInstance;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.xml.transform.*;
import javax.xml.transform.dom.DOMSource;
import javax.xml.transform.stream.StreamResult;
import java.io.StringWriter;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static java.lang.String.format;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class ExternalTaskTest {

    private static final Logger logger = LoggerFactory.getLogger(ExternalTaskTest.class);

    @Test
    public void testEngine() throws InterruptedException, TransformerException {

        final BpmnModelInstance model = Bpmn.createExecutableProcess("test")
                .startEvent()
                .signal("start").camundaExecutionListenerClass("start", SignalListener.class)
                .serviceTask("Step1").camundaClass(TaskStep.class)
                .serviceTask("Step2").camundaExternalTask("Step2")
                .serviceTask("Step3").camundaClass(TaskStep.class)
                .endEvent()
                .signal("end").camundaExecutionListenerClass("end", SignalListener.class)
                .done();

        logger.debug("Model:\n" + prettyPrint(model) + "\n\n");

        final ProcessEngine engine = new StandaloneInMemProcessEngineConfiguration().buildProcessEngine();
        engine.getRepositoryService().createDeployment().addModelInstance("test.bpmn", model).deploy();

        final ExecutorService executorService = createExternalTaskService(engine);

        final RuntimeService runtimeService = engine.getRuntimeService();
        final ProcessInstanceWithVariables instance = (ProcessInstanceWithVariables) runtimeService.startProcessInstanceByKey("test", Variables.putValue("trace", "initial"));

        Thread.sleep(3000L);
        executorService.shutdownNow();

        logger.info("ExecutorService shut down");
        Thread.sleep(3000L);

        assertTrue(instance.isEnded());
    }

    private ExecutorService createExternalTaskService(final ProcessEngine engine) {
        final ExecutorService executorService = Executors.newFixedThreadPool(1);
        executorService.submit(() -> {
            final ExternalTaskService externalTaskService = engine.getExternalTaskService();
            try {
                while (true) {
                    logger.info("MockWorker polling ...");
                    final List<LockedExternalTask> tasks = externalTaskService.fetchAndLock(1, "MockWorker")
                            .topic("Step2", 60 * 1000L)
                            .execute();
                    if (tasks.isEmpty()) {
                        Thread.sleep(1000L);
                        continue;
                    }
                    for (final LockedExternalTask task : tasks) {
                        final String trace = (String) task.getVariables().get("trace");
                        logger.info(format("In MockWorker %s, trace=%s", task.getTopicName(), trace));
                        externalTaskService.complete(task.getId(), "MockWorker", Variables.putValue("trace", trace + " -> Step2"));
                    }
                }
            } catch (Throwable t) {
                logger.info("MockWorker caught " + t.getClass().getName());
                logger.debug(t.getMessage(), t);
            } finally {
                logger.info("Exiting MockWorker");
            }
        });
        return executorService;
    }

    public static class TaskStep implements JavaDelegate {

        @Override
        public void execute(final DelegateExecution execution) {
            final String trace = (String) execution.getVariable("trace");
            final String step = execution.getCurrentActivityId();
            logger.info(format("In %s, trace=%s", step, trace));
            execution.setVariable("trace", trace + " -> " + step);
        }

    }

    public static class SignalListener implements JavaDelegate {
        @Override
        public void execute(DelegateExecution execution) throws Exception {
            logger.info(format("In %s: received %s", SignalListener.class.getSimpleName(), execution.getEventName()));
        }
    }

    static String prettyPrint(final BpmnModelInstance model) throws TransformerException {
        final DOMSource domSource = model.getDocument().getDomSource();
        final StringWriter writer = new StringWriter();
        final Transformer transformer = TransformerFactory.newInstance().newTransformer();
        transformer.setOutputProperty(OutputKeys.INDENT, "yes");
        transformer.transform(domSource, new StreamResult(writer));
        return writer.toString();
    }

}