data processor messaging #11

core-back-end/build.gradle

@@ -21,8 +21,8 @@ dependencies {
     implementation 'org.springframework.boot:spring-boot-starter-web'
     implementation 'org.springframework.boot:spring-boot-starter-webflux'
     implementation 'org.springframework.boot:spring-boot-starter-security'
-    implementation group: 'org.keycloak', name: 'keycloak-spring-boot-adapter', version: '18.0.2'
-    implementation 'org.keycloak:keycloak-admin-client:23.0.3'
+//    implementation group: 'org.keycloak', name: 'keycloak-spring-boot-adapter', version: '23.0.3'
+//    implementation 'org.keycloak:keycloak-admin-client:23.0.3'
     implementation 'org.springframework.kafka:spring-kafka'
     testImplementation 'org.springframework.boot:spring-boot-starter-test'
     testImplementation 'io.projectreactor:reactor-test'
@@ -31,3 +31,4 @@ dependencies {
 tasks.named('test') {
     useJUnitPlatform()
 }
+

core-back-end/src/main/java/org/student/messaging/KafkaConfig.java

@@ -1,6 +1,5 @@
 package org.student.messaging;
 
-import jakarta.validation.Valid;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.StringSerializer;

data-processor/build.gradle

@@ -1,19 +1,29 @@
 plugins {
     id 'java'
+    id 'org.springframework.boot' version '3.2.0'
+    id 'io.spring.dependency-management' version '1.1.4'
 }
 
-group = 'org.student'
-version = '1.0-SNAPSHOT'
+group = "org.student"
+version = "1.0-SNAPSHOT"
 
-repositories {
-    mavenCentral()
+java {
+    sourceCompatibility = '17'
+}
+
+configurations {
+    compileOnly {
+        extendsFrom annotationProcessor
+    }
 }
 
 dependencies {
-    testImplementation platform('org.junit:junit-bom:5.9.1')
-    testImplementation 'org.junit.jupiter:junit-jupiter'
+    implementation 'org.springframework.boot:spring-boot-starter:3.2.0'
+    implementation 'org.springframework.kafka:spring-kafka'
+    testImplementation 'org.springframework.boot:spring-boot-starter-test'
+    testImplementation 'io.projectreactor:reactor-test'
 }
 
-test {
+tasks.named('test') {
     useJUnitPlatform()
 }

data-processor/src/main/java/org/student/Application.java

@@ -1,7 +1,11 @@
 package org.student;
 
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
 public class Application {
 	public static void main(String[] args) {
-		System.out.println("Hello world!");
+		SpringApplication.run(Application.class, args);
 	}
 }

data-processor/src/main/java/org/student/messaging/KafkaConfig.java

@@ -0,0 +1,35 @@
+package org.student.messaging;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@Configuration
+public class KafkaConfig {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    @Bean
+    public ProducerFactory<String, byte[]> producerFactory() {
+        Map<String, Object> configProp = new HashMap<>();
+        configProp.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        configProp.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        configProp.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+        return new DefaultKafkaProducerFactory<>(configProp);
+    }
+
+    @Bean
+    public KafkaTemplate<String, byte[]> kafkaTemplate() {
+        return new KafkaTemplate<>(producerFactory());
+    }
+}

data-processor/src/main/java/org/student/messaging/MessageConsumer.java

@@ -0,0 +1,13 @@
+package org.student.messaging;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.annotation.TopicPartition;
+import org.springframework.stereotype.Service;
+
+@Service
+public class MessageConsumer {
+    @KafkaListener(id = "data-processor-listener", topicPartitions = @TopicPartition(topic = "artifact-processor-topic", partitions = {"0"}), groupId = "core-file-garbage-group")
+    public void consume(byte[] message) {
+        System.out.println("Consumed message: " + new String(message));
+    }
+}

data-processor/src/main/java/org/student/messaging/MessageProducer.java

@@ -0,0 +1,18 @@
+package org.student.messaging;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+public class MessageProducer {
+    private final KafkaTemplate<String, byte[]> template;
+
+    MessageProducer(KafkaTemplate<String, byte[]> template) {
+        this.template = template;
+    }
+
+    public void send(byte[] message) {
+        template.send(new ProducerRecord<>("artifact-processor-topic", 1, null, message));
+    }
+}

data-processor/src/main/resources/application.yaml

@@ -0,0 +1,11 @@
+spring:
+  kafka:
+    bootstrap-servers: localhost:9092
+    consumer:
+      group-id: core-file-garbage-group
+      auto-offset-reset: earlier
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
+    producer:
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+      value-serializer: org.apache.kafka.common.serialization.ByteArraySerializer