Handling Mono<> object in a JobWorker in a Reactive Way

Camunda 8 Topics
,
1

I have a design question regarding reactive programming. I have a service that returns a Mono<> object (in the background, it calls an endpoint of another service), and I’d like to handle it in a reactive way to avoid ‘killing’ the idea with a block() function. Is this solution an acceptable way to handle it (I’d like to put the created object into variables for further processing), or will I run into various problems like concurrency and so on?

    @JobWorker(type = "persistOneEntityWorker", autoComplete = false)
    @SuppressWarnings("unused")
    public void persistOneEntity(final JobClient client, final ActivatedJob job) {

        Map<String, Object> variables = job.getVariablesAsMap();

        OneEntity oneEntity = new OneEntity(Long.toString(job.getProcessInstanceKey()), "body one camunda");
        oOneService.createOneEntity(oneEntity)
                .subscribe(created -> variables.put("oneEntity", created),
                        error -> {
                            throw new RuntimeException("Could not complete job " + job, error);
                        },
                        () -> {
                            client.newCompleteCommand(job.getKey())
                                    .variables(variables)
                                    .send()
                                    .exceptionally(throwable -> {
                                        throw new RuntimeException("Could not complete job " + job, throwable);
                                    });

                            logger.debug("PersistOneEntity system task {} completed!", job.getKey());
                        });
    }
2

So, I found this great video https://www.youtube.com/watch?v=ZHKz9l5yG3Q where @BerndRuecker shows how to do it, which is pretty similar to what I’ve done. Therefore, I assume that the approach is correct and does not produce any negative side effects. In the updated version, I’ve incorporated proper error handling, so the final code looks like this:

    @JobWorker(type = "persistOneEntityReactiveWorker", autoComplete = false)
    @SuppressWarnings("unused")
    public void persistOneEntity(final JobClient client, final ActivatedJob job) {

        Map<String, Object> variables = job.getVariablesAsMap();

        OneEntity oneEntity = new OneEntity(Long.toString(job.getProcessInstanceKey()), "body one reactive");
        oOneService.createOneEntity(oneEntity)
                .subscribe(created -> variables.put("oneEntityReactive", created),
                        error -> {
                            client.newFailCommand(job.getKey())
                                    .retries(1)
                                    .errorMessage("Could not complete job " + job + " due to: " + error.getMessage())
                                    .send()
                                    .exceptionally(throwable -> {
                                        throw new RuntimeException("Could not fail job " + job, throwable);
                                    });
                        },
                        () -> {
                            client.newCompleteCommand(job.getKey())
                                    .variables(variables)
                                    .send()
                                    .thenApply(jobResponse -> jobResponse)
                                    .exceptionally(throwable -> {
                                        throw new RuntimeException("Could not complete job " + job, throwable);
                                    });
                            logger.debug("PersistOneEntity reactive way system task {} completed!", job.getKey());
                        });
    }