kafka.patch

Index: src/main/java/com/handson/searchengine/config/KafkaTopicConfig.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/main/java/com/handson/searchengine/config/KafkaTopicConfig.java b/src/main/java/com/handson/searchengine/config/KafkaTopicConfig.java
new file mode 100644
--- /dev/null	(date 1645314637143)
+++ b/src/main/java/com/handson/searchengine/config/KafkaTopicConfig.java	(date 1645314637143)
@@ -0,0 +1,30 @@
+package com.handson.searchengine.config;
+
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.core.KafkaAdmin;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static com.handson.searchengine.kafka.Producer.APP_TOPIC;
+
+public class KafkaTopicConfig {
+
+    @Value( "${spring.kafka.bootstrap-servers}")
+    private String bootstrapAddress;
+
+    @Bean
+    public KafkaAdmin kafkaAdmin() {
+        Map<String, Object> configs = new HashMap<>();
+        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        return new KafkaAdmin(configs);
+    }
+
+    @Bean
+    public NewTopic topic1() {
+        return new NewTopic(APP_TOPIC, 1, (short) 1);
+    }
+}
Index: src/main/java/com/handson/searchengine/kafka/Consumer.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/main/java/com/handson/searchengine/kafka/Consumer.java b/src/main/java/com/handson/searchengine/kafka/Consumer.java
new file mode 100644
--- /dev/null	(date 1645314311859)
+++ b/src/main/java/com/handson/searchengine/kafka/Consumer.java	(date 1645314311859)
@@ -0,0 +1,31 @@
+package com.handson.searchengine.kafka;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.handson.searchengine.crawler.Crawler;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.Optional;
+
+import static com.handson.searchengine.kafka.Producer.APP_TOPIC;
+
+@Component
+public class Consumer {
+
+
+    @KafkaListener(topics = {APP_TOPIC})
+    public void listen(ConsumerRecord<?, ?> record){
+
+        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
+
+        if (kafkaMessage.isPresent()) {
+
+            Object message = kafkaMessage.get();
+            System.out.println("---->" + record);
+            System.out.println("---->" + message);
+
+        }
+    }
+}
Index: src/main/java/com/handson/searchengine/kafka/Producer.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/main/java/com/handson/searchengine/kafka/Producer.java b/src/main/java/com/handson/searchengine/kafka/Producer.java
new file mode 100644
--- /dev/null	(date 1645314311862)
+++ b/src/main/java/com/handson/searchengine/kafka/Producer.java	(date 1645314311862)
@@ -0,0 +1,24 @@
+package com.handson.searchengine.kafka;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+
+@Component
+public class Producer {
+
+    public static final String APP_TOPIC = "searchengine";
+    @Autowired
+    ObjectMapper om;
+
+    @Autowired
+    private KafkaTemplate kafkaTemplate;
+
+    public void send(Object message) throws JsonProcessingException {
+        kafkaTemplate.send(APP_TOPIC, om.writeValueAsString(message));
+    }
+
+}