Message Buffering Camunda rabbitmq spring-boot

Question

i have a setup of 3 spring-boot projects each acting as a separate participant, each project deploys one bpmn model communicating with each other in a loop based pattern.

Although, i am using a message broker as a queue i do not find a way to correlate messages to camunda where the receive task is not ready yet. So, e.g process 1 sends a message to process 2 but process 2 is not yet on receive task so i encounter a MismatchingCorrelationError which tells me that the sended message was too fast and the corresponding receive task was not ready yet.

I thought of a message queue so i decided to use rabbitmq broker but the java setup simply takes in a message to the queue and it sends is right away to camunda but i do not want that it gets sended to camunda, instead i would like to only send the queued messages when i am on a receive task.

So i could define a start listener on each receive task which will call java delegate class somehow calling rabbitmq queue that i am now ready and then queue will send the next message right away but not before. Does anybody have a clou to solve this?

In the configuration class i needed to set up a Bean with MessageListenerAdapter: `

@Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }

`

The Receiver class:

@Component
public class Receiver {

    private CountDownLatch latch = new CountDownLatch(1);

    @Autowired
    CamundaMessageProcessor messageProcessor;

    public void receiveMessage(String message) {
        try {
            Response response = messageProcessor.processMessage(message);
        }catch(Exception e){
            // @TODO: Fix up error catching with the service
            
        }finally{
            latch.countDown();
        }

    }

    public CountDownLatch getLatch() {
        return latch;
    }
}

And the MessageProcessor class:

@Service
public class CamundaMessageProcessor{

    @Autowired
    private ProcessEngine engine;

    @Autowired
    private ObjectMapper objectMapper;

    public Response processMessage(String message){
        CorrelationMessageDto messageDto = null;
        try {
            System.out.println("MESSAGE: " + message);
            messageDto = objectMapper.readValue(message, CorrelationMessageDto.class);
            
        } catch (JsonMappingException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (JsonProcessingException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        MessageRestServiceImpl service = new MessageRestServiceImpl(engine.getName(), objectMapper);
        Response response = service.deliverMessage(messageDto);
        System.out.println("Response deliverMessage " + response.toString());
        return response;
    }
}

These are my sample github project: https://github.com/SebastianAT/camunda_springboot_rabbitmq

Somehow i expect that camunda will buffer my message as long as i am not on a receive task but it seems that this won't work with camunda 7.

Thanks in advance. Best Regards Sebastian

rob2universe · Accepted Answer · 2022-11-21 15:46:23Z

0

Camunda 8 can buffer your message, please see: https://docs.camunda.io/docs/components/concepts/messages/

If you want to call out when ready, then you don't need messaging. Telling a queue when a receiver is ready is not how such architectures normally work. If you can call a service when the stage is reached, why not let the one process write to a DB and the other process fetch the information when it needs it?

With a message broker, you would normally attempt a delivery and if needed perform a retry after a certain delay, and potentially another retries after a potentially higher delay. Found this example for RabbitMQ https://github.com/fefferoni/rabbitmq-delayed-retry-experiment/ I cannot comment on the quality of the example, but the design/diagram looks reasonable. RabbitMq also seems to support delayed queues.

answered Nov 21, 2022 at 15:46

rob2universe

7,64844 silver badges59 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

sebi Over a year ago

Thanks. Yes C8 usines message buffering but for now the migration from c7 to c8 is not straight forward. Yes you are right this architecture with db is not what i want.

sebi Over a year ago

Thanks also for the posted repo. I looked into and tried to build it with my architecture but i am not able to receive messages in camunda, if i send it with postman everything works. I am not sure if this is because i use a MessagePostProcessor from Spring and not the offical MessageServiceImpl from camunda with Response class.

sebi Over a year ago

You could also try with my repo: github.com/SebastianAT/camunda_springboot_rabbitmq/tree/main/… I deployed "my-project5" and "my-project6" to different ports (8095, 8096) and started my-project5 manually on camunda tasklist. (mvn clean install spring-boot:run)

Collectives™ on Stack Overflow

Message Buffering Camunda rabbitmq spring-boot

1 Answer 1

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related