Spring cloud stream with solace

@akg17 said:
@marc I had gone through this document, I know how errorqueue and DMQ is different and what is used for now the challenge is reprocessing.
1. I can add a listener to my error queue and subscribe and reprocess the message but as per our project existing code they are making use of kafkaTemplate, RabbitMqTemplate to siphon message from the queue, even proving number of message
and even for solace i did something long back using jms & jndi Template but i am not sure how to do the same thing with spring cloud stream. I need to consume message whenever we want not using a listener. approach could be to siphon a number of message from the queue , save to data base, and then from the UI view the message and also reprocess which could dropping to same queue back or some other way depending on the status of message may be.
2. And same question for DMQ message reprocessing,
I need your help about reproccessing of the message how to do the same thing with spring cloud stream and solace like we do using Jms and JndiTemplate. see the below code i used a year back for jms,

    InitialContext initialContext = null;
    SolSession session = null;
    List<DeadLetterMessage> deadLetterMessages = new ArrayList<>();
    try {
        session = (SolSession) solConnection.createSession(true, 0);
        session.start();
        initialContext = (InitialContext) jndiTemplate.getContext();
        Queue dlqQueue = (Queue) initialContext.lookup(dlq);
        MessageConsumer consumer = session.createConsumer(dlqQueue);
        for (int i = 0; i < numOfMessage; i++) {
            try {
                TextMessage textMessage = (TextMessage) consumer.receive(1000);
                DeadLetterMessage deadLetterMessage = objectMapper.readValue(textMessage.getText(), DeadLetterMessage.class);
                deadLetterMessage.setRecoveryStatus(RecoveryStatus.SIPHONED);
                deadLetterMessages.add(deadLetterMessage);
            } catch (Exception e) {
                log.error("Failed Parsing Message " + e.getMessage());
            }
        }

@UshShukla said:
Hey @akg17
Thanks for your patience while we worked this out. If I understood your question, you were asking how to poll a DLQ on-demand when using Spring Cloud Stream.
I've posted a demo on the Solace Github that details how to go about doing this.

The general approach is as follows:
1. Define a pollable source (the queue you want to poll) in application.properties, along with the number of messages to get on each poll.
2. Instantiate PollableMessageSource to get a reference to the queue in your code
3. Invoke the poll method on PollableMessageSource as needed to get data.

@UshShukla said:
Hey @akg17
Thanks for your patience while we worked this out. If I understood your question, you were asking how to poll a DLQ on-demand when using Spring Cloud Stream.
I've posted a demo on the Solace Github that details how to go about doing this.

The general approach is as follows:
1. Define a pollable source (the queue you want to poll) in application.properties, along with the number of messages to get on each poll.
2. Instantiate PollableMessageSource to get a reference to the queue in your code
3. Invoke the poll method on PollableMessageSource as needed to get data.

Hi @UshShukla , Thanks for this demo code. This helped me in a great deal. I am working on a solution where: I need to queue up all the messages in error queue where one of the sync system is down (to achieve zero RPO). I have provisioned error queue size for 30 mins downtime of database. And my poller runs every 30 mins and needs to process all the messages in the error queue. I have modified the demo code as below:

public void poll() {
        while (true) {
            Boolean result = this.source.poll(receive -> {
                String msg = new String((byte[]) receive.getPayload());
                System.out.println("Polling received message with Message " + msg);
                // Do process and if some conditions arise, throw exception
                Boolean condition = true;
                if (condition) {
                    throw new RuntimeException("Permanent failure");
                }
            });

            if (!result) {
                System.out.println("No more messages to process");
                break;
            }
        }
    }

This works great if RuntimeException is not thrown or only one message in the queue. If there are more than one message in the queue. e.g. "Hello World 1", "Hello World 2", "Hello World 3", the above code prints "Hello World 1" twice and all the 3 messages disappear from the queue. Please note max-delivery is set to 1 and queue itself has redelivery as 1.
Please note the following setting:

spring.cloud.stream.bindings.poller-in-0.destination = <somevalue>
spring.cloud.stream.bindings.poller-in-0.group = <somegroup>
spring.cloud.stream.pollable-source = poller
spring.cloud.stream.poller.max-messages-per-poll  = 1

Would you have any suggestion? Thanks in advance.