Implement Application event in order to subscribe kafka consumers

Author: ardizioa

src/main/java/com/aardizio/SpringReactiveApplication.java

@@ -1,17 +1,35 @@
 package com.aardizio;
 
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.ApplicationListener;
 import org.springframework.web.reactive.config.EnableWebFlux;
 
+import reactor.kafka.receiver.KafkaReceiver;
 import springfox.documentation.swagger2.annotations.EnableSwagger2;
 
 @SpringBootApplication
 @EnableSwagger2
 @EnableWebFlux
-public class SpringReactiveApplication {
+public class SpringReactiveApplication implements ApplicationListener<ApplicationReadyEvent>{
+
+@Qualifier("simpleConsumer")
+@Autowired
+private KafkaReceiver<String,String> receiver;
+
 
 public static void main(String[] args) {
 SpringApplication.run(SpringReactiveApplication.class, args);
 }
+
+public void onApplicationEvent(ApplicationReadyEvent event){
+receiver.receive().subscribe(r -> {
+System.out.printf("Received message: %s\n", r); 
+r.receiverOffset().acknowledge();
+}); 
+}
+
 }

src/main/java/com/aardizio/config/KafkaConsumerConfig.java

@@ -37,10 +37,6 @@ public KafkaReceiver<String,String> simpleConsumer() {
  ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create(consumerProps)
  .subscription(Collections.singleton("prova"));
  KafkaReceiver<String,String> receiver = KafkaReceiver.create(receiverOptions);
- receiver.receive().subscribe(r -> {
- System.out.printf("Received message: %s\n", r); 
- r.receiverOffset().acknowledge();
- }); 
  return receiver;
  }
 

src/main/java/com/aardizio/config/KafkaProducerConfig.java

@@ -37,6 +37,4 @@ public KafkaSender<String,String> simpleProducer(){
  }
 
 
-
-
 }