Search code examples
spring-bootspring-amqpspring-rabbit

i want send message in rabbitMQ with delay of 5seconds using @Scheduler annotation

What I am Doing: : I am actually fetching values 5 at a time(in a chunk) from the postgres table and storing these 5 values and sending it to rabbitMQ. {I am able to do that}
What i want is after sending one chunk(chunk means 5 rows of table at a time) of data I need to create a delay of 5seconds then send the next chunk.
Error: Error is because i am using @Scheduler annotation in AccessDataJpaApplication.java. I need to where should i placed these two annotation @EnableSchedule and @Scheduled(fixedDelay=5000L) .
Note: I also know why i am getting error but i want a solution. Like I know @Scheduled annotation can be used if there is no written argument, then tell me where should i placed in this program. I actually want to delay sendMessage function inside AccessingDataJpaApplaication.java . Or if not possible then tell me how to use delay_message_exchange plugin in spring boot application in here. I am using the latest spring version.

Code: OutBox

package com.example.demo;


import com.fasterxml.jackson.annotation.JsonProperty;

import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;
import java.io.Serializable;

@Entity
public class OutBox implements Serializable {
    @Id
    private String id;
    private String aggregrate;
    private String operation;
    private String message;

    public OutBox() {
    }
    public OutBox(@JsonProperty("id") String id, @JsonProperty("aggregrate") String aggregrate, @JsonProperty("operation") String operation, @JsonProperty("message") String message) {
        this.id=id;
        this.aggregrate=aggregrate;
        this.operation=operation;
        this.message=message;
    }

    @Override
    public String toString() {
        return String.format(
                "OutBox{ id='%s', aggregrate='%s', operations='%s', message='%s' }",
                id, aggregrate, operation, message);
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getAggregrate() {
        return aggregrate;
    }

    public void setAggregrate(String aggregrate) {
        this.aggregrate = aggregrate;
    }

    public String getOperation() {
        return operation;
    }

    public void setOperation(String operation) {
        this.operation = operation;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }
}


Code: OutBoxRepository 

package com.example.demo;

import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.repository.JpaRepository;

public interface OutBoxRepository extends JpaRepository<OutBox, String> {
    Page<OutBox> findAll(Pageable pageable);
}


Code: RabbitMQConfig 

package com.example.demo;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitMQConfig {
    @Value("${javainuse.rabbitmq.queue}")
    String queueName;

    @Value("${javainuse.rabbitmq.exchange}")
    String exchange;

    @Value("${javainuse.rabbitmq.routingkey}")
    private String routingkey;

    @Bean
    Queue queue() {
        return new Queue(queueName, false);
    }

//    @Bean
//    CustomExchange delayExchange() {
//        Map<String, Object> args = new HashMap<String, Object>();
//        args.put("x-delayed-type", "direct");
//        return new CustomExchange("my-exchange", "x-delayed-message", true, false, args);
//    }

    @Bean
    DirectExchange exchange() {
        return new DirectExchange(exchange);
    }

    @Bean
    Binding binding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(routingkey);
    }

    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }


    public AmqpTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }
//    @Bean
//    public AmqpTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
//        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//        rabbitTemplate.setMessageConverter(jsonMessageConverter());
//        return rabbitTemplate;
//    }
}

Code: AccessingDataJpaApplication 

package com.example.demo;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.utils.SerializationUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import java.util.List;
import java.util.Optional;

import org.springframework.context.annotation.Configuration;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

@SpringBootApplication
public class AccessingDataJpaApplication {

    private static final Logger log = LoggerFactory.getLogger(AccessingDataJpaApplication.class);

    @Autowired
    private AmqpTemplate rabbitTemplate;

    @Autowired
    private OutBoxRepository repository;

    @Value("${javainuse.rabbitmq.exchange}")
    private String exchange;

    @Value("${javainuse.rabbitmq.routingkey}")
    private String routingkey;

    public static void main(String[] args) {
        SpringApplication.run(AccessingDataJpaApplication.class);
    }

    @Scheduled(fixedRate=5000)
    public void sendMessage(List<OutBox> message) {

        log.info("Sending message...");
//        rabbitTemplate.convertAndSend(exchange, routingkey,SerializationUtils.deserialize(message));
        for(int i=0;i<message.size();i++)
            rabbitTemplate.convertAndSend(exchange,routingkey, message.get(i));
    }

    @Bean
    public CommandLineRunner demo(OutBoxRepository repository) {

        return (args) -> {
            repository.save(new OutBox("fsks-ghty-eryr-jghd","OO_FLOW_SCHEDULES","UPDATE","{ \"brand\" : \"Mercedes\", \"doors\" : 5 }"));
            repository.save(new OutBox("fsks-bnmb-eryr-jghd","OO_FLOW_ENTITY","UPDATE","{ \"brand\" : \"BMW\", \"doors\" : 7 }"));


            log.info("Customers found with findAll():");
            log.info("--------------PAGE: 0-----------------");


            int count = 0;
            List<OutBox> lst=null;
            for (OutBox outbox : repository.findAll()) {
                Page<OutBox> u = repository.findAll(PageRequest.of(count, 5));
                lst=u.getContent();
                sendMessage(lst);
                log.info(outbox.toString());
                count ++;
            }



            log.info("");//log is to used for printing in console

            Optional<OutBox> outbox = repository.findById("fsks-ghty-eryr-jite"); // L means of type long
            log.info("Customer found with findById(1L):");
            log.info("--------------------------------");
            log.info(outbox.toString());
        };

    }
}


Output in Queue: [5:41 PM] Akash Anand

{"id":"fsks-ghty-eryr-jghd","aggregrate":"OO_FLOW_SCHEDULES","operation":"UPDATE","message":"{ \"brand\" : \"Mercedes\", \"doors\" : 5 }"}


Error:
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'accessingDataJpaApplication' defined in file [C:\Users\AkAnand\Downloads\rabbitMQTable\demo\demo\target\classes\com\example\demo\AccessingDataJpaApplication.class]: Initialization of bean failed; nested exception is java.lang.IllegalStateException: Encountered invalid @Scheduled method 'sendMessage': Only no-arg methods may be annotated with @Scheduled at
org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:603) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at
org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:517) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at
org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:323) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at
org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:226) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at
org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:321) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at
org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:202) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at
org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:895) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at
org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:878) ~[spring-context-5.2.6.RELEASE.jar:5.2.6.RELEASE] at
org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:550) ~[spring-context-5.2.6.RELEASE.jar:5.2.6.RELEASE] at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:143) ~[spring-boot-2.3.0.RELEASE.jar:2.3.0.RELEASE] at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:758) [spring-boot-2.3.0.RELEASE.jar:2.3.0.RELEASE] at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:750) [spring-boot-2.3.0.RELEASE.jar:2.3.0.RELEASE] at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) [spring-boot-2.3.0.RELEASE.jar:2.3.0.RELEASE] at org.springframework.boot.SpringApplication.run(SpringApplication.java:315) [spring-boot-2.3.0.RELEASE.jar:2.3.0.RELEASE] at org.springframework.boot.SpringApplication.run(SpringApplication.java:1237) [spring-boot-2.3.0.RELEASE.jar:2.3.0.RELEASE] at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226) [spring-boot-2.3.0.RELEASE.jar:2.3.0.RELEASE] at com.example.demo.DemoApplication.main(DemoApplication.java:12) [classes/:na] Caused by: java.lang.IllegalStateException: Encountered invalid @Scheduled method 'sendMessage': Only no-arg methods may be annotated with @Scheduled at org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor.processScheduled(ScheduledAnnotationBeanPostProcessor.java:499) ~[spring-context-5.2.6.RELEASE.jar:5.2.6.RELEASE] at org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor.lambda$null$1(ScheduledAnnotationBeanPostProcessor.java:362) ~[spring-context-5.2.6.RELEASE.jar:5.2.6.RELEASE] at java.lang.Iterable.forEach(Iterable.java:75) ~[na:1.8.0_241] at org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor.lambda$postProcessAfterInitialization$2(ScheduledAnnotationBeanPostProcessor.java:362) ~[spring-context-5.2.6.RELEASE.jar:5.2.6.RELEASE] at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684) ~[na:1.8.0_241] at org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor.postProcessAfterInitialization(ScheduledAnnotationBeanPostProcessor.java:361) ~[spring-context-5.2.6.RELEASE.jar:5.2.6.RELEASE] at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.applyBeanPostProcessorsAfterInitialization(AbstractAutowireCapableBeanFactory.java:431) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1800) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:595) ~[spring-beans-5.2.6.RELEASE.jar:5.2.6.RELEASE] ... 16 common frames omitted

Solution

  • The error is quite clear.

    Caused by: java.lang.IllegalStateException: Encountered invalid @Scheduled method 'sendMessage': Only no-arg methods may be annotated with @Scheduled

    You can only schedule methods with no parameters.