refactor: use FlamingockSpringbootTestSupport and beta.6 (#5)

Author: dieppa

inventory-orders-service/build.gradle.kts

@@ -28,7 +28,7 @@ repositories {
 group = "io.flamingock"
 version = "1.0-SNAPSHOT"
 
-val flamingockVersion = "1.0.0-beta.5"
+val flamingockVersion = "1.0.0-beta.6"
 logger.lifecycle("Building with flamingock version: $flamingockVersion")
 
 val mongodbVersion = "5.5.1"
@@ -73,12 +73,16 @@ dependencies {
 //    Others dependencies needed for this example
 //    implementation("org.slf4j:slf4j-simple:2.0.6")  // Commented out - Spring Boot provides logging
 
+    testImplementation("org.springframework.boot:spring-boot-starter-test")
+    testImplementation("io.flamingock:flamingock-springboot-test-support")
     testImplementation("org.junit.jupiter:junit-jupiter-api:5.9.2")
     testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine:5.9.2")
 
-    testImplementation("org.testcontainers:testcontainers-mongodb:2.0.2")
-    testImplementation("org.testcontainers:testcontainers-kafka:2.0.2")
-    testImplementation("org.testcontainers:testcontainers-junit-jupiter:2.0.2")
+    testImplementation("org.springframework.boot:spring-boot-testcontainers")
+    testImplementation("org.testcontainers:testcontainers:1.21.4")
+    testImplementation("org.testcontainers:mongodb:1.21.4")
+    testImplementation("org.testcontainers:kafka:1.21.4")
+    testImplementation("org.testcontainers:junit-jupiter:1.21.4")
 }
 
 application {

inventory-orders-service/gradlew

@@ -1,20 +1,46 @@
 package io.flamingock.examples.inventory;
 
 import com.mongodb.client.MongoClient;
+import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
+import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.flamingock.community.mongodb.sync.driver.MongoDBSyncAuditStore;
+import io.flamingock.examples.inventory.util.KafkaSchemaManager;
+import io.flamingock.examples.inventory.util.LaunchDarklyClient;
 import io.flamingock.examples.inventory.util.MongoDBUtil;
 import io.flamingock.internal.core.store.CommunityAuditStore;
 import io.flamingock.targetsystem.nontransactional.NonTransactionalTargetSystem;
 import io.flamingock.targetystem.mongodb.sync.MongoDBSyncTargetSystem;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import jakarta.annotation.PreDestroy;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+
 @Configuration
 public class FlamingockConfig {
 
-    @Bean
+    @Value("${mongodb.uri:mongodb://localhost:27017/}")
+    private String mongodbUri;
+
+    @Value("${kafka.bootstrap-servers:localhost:9092}")
+    private String kafkaBootstrapServers;
+
+    @Value("${kafka.schema-registry-url:http://localhost:8081}")
+    private String schemaRegistryUrl;
+
+    @Value("${launchdarkly.api-url:http://localhost:8765/api/v2}")
+    private String launchDarklyApiUrl;
+
+    private AdminClient kafkaAdminClient;
+
+    @Bean(destroyMethod = "close")
     public MongoClient mongoClient() {
-        return MongoDBUtil.getMongoClient("mongodb://localhost:27017/");
+        return MongoDBUtil.getMongoClient(mongodbUri);
     }
 
     @Bean
@@ -24,12 +50,29 @@ public MongoDBSyncTargetSystem mongoDBSyncTargetSystem(MongoClient mongoClient)
 
     @Bean
     public NonTransactionalTargetSystem kafkaTargetSystem() throws Exception {
-        return TargetSystems.kafkaTargetSystem();
+        SchemaRegistryClient schemaRegistryClient = new CachedSchemaRegistryClient(
+                Collections.singletonList(schemaRegistryUrl),
+                100
+        );
+
+        Properties kafkaProps = new Properties();
+        kafkaProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBootstrapServers);
+        this.kafkaAdminClient = AdminClient.create(kafkaProps);
+
+        KafkaSchemaManager schemaManager = new KafkaSchemaManager(schemaRegistryClient, kafkaAdminClient);
+        schemaManager.createTopicIfNotExists("order-created", 3, (short) 1);
+        return new NonTransactionalTargetSystem(TargetSystems.KAFKA_TARGET_SYSTEM).addDependency(schemaManager);
     }
 
     @Bean
     public NonTransactionalTargetSystem toggleTargetSystem() {
-        return TargetSystems.toggleTargetSystem();
+        LaunchDarklyClient launchDarklyClient = new LaunchDarklyClient(
+                "demo-token",
+                "inventory-service",
+                "production",
+                launchDarklyApiUrl
+        );
+        return new NonTransactionalTargetSystem(TargetSystems.FEATURE_FLAG_TARGET_SYSTEM).addDependency(launchDarklyClient);
     }
 
 
@@ -38,4 +81,11 @@ public NonTransactionalTargetSystem toggleTargetSystem() {
     public CommunityAuditStore auditStore(MongoDBSyncTargetSystem mongoDBSyncTargetSystem) {
         return MongoDBSyncAuditStore.from(mongoDBSyncTargetSystem);
     }
+
+    @PreDestroy
+    public void cleanup() {
+        if (kafkaAdminClient != null) {
+            kafkaAdminClient.close(Duration.ofSeconds(2));
+        }
+    }
 }