add keycloak and keycloak kafka controller

Author: xcodeassociated

docker/kafka/docker-compose.yml

@@ -1,5 +1,6 @@
 version: "3"
 services:
+  ### KAFKA
   dev-kafka-zookeeper:
     container_name: dev-kafka-zookeeper
     hostname: dev-kafka-zookeeper
@@ -25,7 +26,45 @@ services:
     volumes:
       - kafka-broker-data:/kafka
 
+  ### KEYCLOAK
+  keycloak:
+    image: keycloak-kafka:0.1.1
+    ports:
+      - "8090:8080"
+    depends_on:
+      - postgres-keycloak
+      - dev-kafka-broker
+    environment:
+      KEYCLOAK_ADMIN: admin
+      KEYCLOAK_ADMIN_PASSWORD: admin
+      KAFKA_TOPIC: keycloak-events
+      KAFKA_CLIENT_ID: keycloak
+      KAFKA_BOOTSTRAP_SERVERS: dev-kafka-broker:9092
+      DB_ADDR: postgres-keycloak
+      DB_PORT: 5433
+      DB_VENDOR: POSTGRES
+      DB_DATABASE: keycloak
+      DB_USER: keycloak
+      DB_SCHEMA: public
+      DB_PASSWORD: keycloak
+      PROXY_ADDRESS_FORWARDING: "true"
+      KAFKA_EVENTS: "REGISTER,LOGIN,LOGOUT"
+      KAFKA_ADMIN_TOPIC: keycloak-admin-events
+    command: start-dev
+
+  postgres-keycloak:
+    image: postgres:latest
+    volumes:
+      - postgres-data:/var/lib/postgresql/data
+    environment:
+      POSTGRES_DB: keycloak
+      POSTGRES_USER: keycloak
+      POSTGRES_PASSWORD: keycloak
+    ports:
+      - "5433:5432"
+
 volumes:
+  ### KAFKA
   kafka-zookeeper-data:
     driver: local
     driver_opts:
@@ -38,4 +77,12 @@ volumes:
     driver_opts:
       o: bind
       type: none
-      device: ./volumes/kafka-broker
+      device: ./volumes/kafka-broker
+
+  ### KEYCLOAK
+  postgres-data:
+    driver: local
+    driver_opts:
+      o: bind
+      type: none
+      device: ./volumes/postgres

http/coroutine.http

@@ -30,7 +30,7 @@ GET http://localhost:8080/coroutine/users?page=0&size=10&sort=id&direction=ASC
 POST http://localhost:8080/coroutine/users
 Content-Type: application/json
 
-{"name": "user test coroutine", "email":  "4445test[email protected]", "role": []}
+{"name": "user test coroutine", "email":  "test[email protected]", "role": []}
 
 ###
 PUT http://localhost:8080/coroutine/users/63c6a6c0686f32099dfebd88

notes/keycloak-kafka-samples/login.json

@@ -0,0 +1,19 @@
+{
+  "id": "d567e184-963e-4836-80d5-fedcb07870e5",
+  "time": 1674955961622,
+  "type": "LOGIN",
+  "realmId": "49b390fd-9e31-4931-a55f-34ff3e0776bb",
+  "clientId": "frontend",
+  "userId": "cd85a973-3f13-4219-827b-0c31044e241f",
+  "sessionId": "c71a9003-5165-4370-b98e-4d295297ecc0",
+  "ipAddress": "172.29.0.1",
+  "error": null,
+  "details": {
+    "auth_method": "openid-connect",
+    "auth_type": "code",
+    "redirect_uri": "http://localhost:3000/login",
+    "consent": "no_consent_required",
+    "code_id": "c71a9003-5165-4370-b98e-4d295297ecc0",
+    "username": "test"
+  }
+}

notes/keycloak-kafka-samples/logout.json

@@ -0,0 +1,14 @@
+{
+  "id": "2c908efd-858a-47e5-b899-e8e68559ab72",
+  "time": 1674955930173,
+  "type": "LOGOUT",
+  "realmId": "49b390fd-9e31-4931-a55f-34ff3e0776bb",
+  "clientId": null,
+  "userId": "8693efc7-736d-45c2-ab93-759ed2dd27c0",
+  "sessionId": "2cce54f3-28d5-4727-ac65-ab8f7c70a8c3",
+  "ipAddress": "172.29.0.1",
+  "error": null,
+  "details": {
+    "redirect_uri": "http://localhost:3000/"
+  }
+}

notes/keycloak-kafka-samples/register.json

@@ -0,0 +1,22 @@
+{
+  "id": "4ae0d28e-a117-473a-b9ac-83b285874098",
+  "time": 1674955961567,
+  "type": "REGISTER",
+  "realmId": "49b390fd-9e31-4931-a55f-34ff3e0776bb",
+  "clientId": "frontend",
+  "userId": "cd85a973-3f13-4219-827b-0c31044e241f",
+  "sessionId": null,
+  "ipAddress": "172.29.0.1",
+  "error": null,
+  "details": {
+    "auth_method": "openid-connect",
+    "auth_type": "code",
+    "register_method": "form",
+    "last_name": "test",
+    "redirect_uri": "http://localhost:3000/login",
+    "first_name": "test",
+    "code_id": "c71a9003-5165-4370-b98e-4d295297ecc0",
+    "email": "[email protected]",
+    "username": "test"
+  }
+}

src/main/kotlin/com/softeno/template/dto/Objects.kt renamed to src/main/kotlin/com/softeno/template/dto/dto.kt

@@ -1,7 +1,7 @@
 package com.softeno.template.event
 
 import com.softeno.template.kafka.KafkaMessage
-import com.softeno.template.kafka.ReactiveKafkaProducerService
+import com.softeno.template.kafka.ReactiveKafkaSampleProducer
 import com.softeno.template.websocket.Message
 import com.softeno.template.websocket.ReactiveMessageService
 import org.apache.commons.logging.LogFactory
@@ -14,14 +14,14 @@ data class AppEvent(val source: String) : ApplicationEvent(source)
 @Component
 class SampleApplicationEventPublisher(
     private val reactiveMessageService: ReactiveMessageService,
-    private val reactiveKafkaProducerService: ReactiveKafkaProducerService
+    private val reactiveKafkaProducer: ReactiveKafkaSampleProducer
 ) : ApplicationListener<AppEvent> {
     private val log = LogFactory.getLog(javaClass)
 
     override fun onApplicationEvent(event: AppEvent) {
         log.info("[event handler]: Received event: $event")
         reactiveMessageService.broadcast(event.toMessage())
-        reactiveKafkaProducerService.send(event.toKafkaMessage())
+        reactiveKafkaProducer.send(event.toKafkaMessage())
     }
 
 }

src/main/kotlin/com/softeno/template/event/AppEvent.kt

@@ -0,0 +1,66 @@
+package com.softeno.template.kafka
+
+import com.fasterxml.jackson.databind.JsonNode
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties
+import org.springframework.boot.context.properties.ConfigurationProperties
+import org.springframework.boot.context.properties.ConstructorBinding
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate
+import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate
+import org.springframework.kafka.support.converter.ByteArrayJsonMessageConverter
+import org.springframework.kafka.support.converter.JsonMessageConverter
+import reactor.kafka.receiver.ReceiverOptions
+import reactor.kafka.sender.SenderOptions
+import java.util.*
+
+
+@ConfigurationProperties(prefix = "com.softeno.kafka")
+@ConstructorBinding
+data class KafkaApplicationProperties(val tx: String, val rx: String, val keycloak: String)
+
+@Configuration
+class JsonMessageConverterConfig {
+    @Bean
+    fun jsonMessageConverter(): JsonMessageConverter {
+        return ByteArrayJsonMessageConverter()
+    }
+}
+
+@Configuration
+class ReactiveKafkaSampleConsumerConfig {
+    @Bean(value = ["kafkaSampleOptions"])
+    fun kafkaReceiverOptions(kafkaProperties: KafkaProperties, props: KafkaApplicationProperties): ReceiverOptions<String, JsonNode> {
+        val basicReceiverOptions: ReceiverOptions<String, JsonNode> = ReceiverOptions.create(kafkaProperties.buildConsumerProperties())
+        return basicReceiverOptions.subscription(Collections.singletonList(props.rx))
+    }
+
+    @Bean(value = ["kafkaSampleConsumerTemplate"])
+    fun reactiveKafkaConsumerTemplate(@Qualifier(value = "kafkaSampleOptions") kafkaReceiverOptions: ReceiverOptions<String, JsonNode>): ReactiveKafkaConsumerTemplate<String, JsonNode> {
+        return ReactiveKafkaConsumerTemplate(kafkaReceiverOptions)
+    }
+}
+
+@Configuration
+class ReactiveKafkaSampleProducerConfig {
+    @Bean(value = ["kafkaSampleProducerTemplate"])
+    fun reactiveKafkaProducerTemplate(properties: KafkaProperties): ReactiveKafkaProducerTemplate<String, KafkaMessage> {
+        val props = properties.buildProducerProperties()
+        return ReactiveKafkaProducerTemplate<String, KafkaMessage>(SenderOptions.create(props))
+    }
+}
+
+@Configuration
+class ReactiveKafkaKeycloakConsumerConfig {
+    @Bean(value = ["kafkaKeycloakOptions"])
+    fun kafkaReceiverOptions(kafkaProperties: KafkaProperties, props: KafkaApplicationProperties): ReceiverOptions<String, JsonNode> {
+        val basicReceiverOptions: ReceiverOptions<String, JsonNode> = ReceiverOptions.create(kafkaProperties.buildConsumerProperties())
+        return basicReceiverOptions.subscription(Collections.singletonList(props.keycloak))
+    }
+
+    @Bean(value = ["kafkaKeycloakConsumerTemplate"])
+    fun reactiveKafkaConsumerTemplate(@Qualifier(value = "kafkaKeycloakOptions") kafkaReceiverOptions: ReceiverOptions<String, JsonNode>): ReactiveKafkaConsumerTemplate<String, JsonNode> {
+        return ReactiveKafkaConsumerTemplate(kafkaReceiverOptions)
+    }
+}

src/main/kotlin/com/softeno/template/kafka/Kafka.kt

@@ -0,0 +1,40 @@
+package com.softeno.template.kafka
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.apache.commons.logging.LogFactory
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.boot.CommandLineRunner
+import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate
+import org.springframework.stereotype.Service
+import reactor.core.publisher.Flux
+
+@Service
+class ReactiveKafkaKeycloakController(
+    @Qualifier(value = "kafkaKeycloakConsumerTemplate") private val reactiveKafkaConsumerTemplate: ReactiveKafkaConsumerTemplate<String, JsonNode>,
+    private val objectMapper: ObjectMapper
+): CommandLineRunner {
+    private val log = LogFactory.getLog(javaClass)
+
+    private fun consumeKafkaMessage(): Flux<JsonNode> {
+        return reactiveKafkaConsumerTemplate
+            .receiveAutoAck()
+            .doOnNext { consumerRecord: ConsumerRecord<String, JsonNode> ->
+                log.debug("[kafka] rx keycloak: ConsumerRecord: key=${consumerRecord.key()}, value=${consumerRecord.value()} from topic=${consumerRecord.topic()}, offset=${consumerRecord.offset()}")
+            }
+            .map { obj: ConsumerRecord<String, JsonNode> -> obj.value() }
+            .doOnNext { message: JsonNode ->
+                val dto: KeycloakUserEvent = objectMapper.readValue(message.toString(), KeycloakUserEvent::class.java)
+                log.info("[kafka] rx keycloak: $dto")
+            }
+            .doOnError { throwable: Throwable ->
+                log.error("[kafka] keycloak: ${throwable.message}")
+            }
+    }
+
+    override fun run(vararg args: String) {
+        log.info("[kafka]: keycloak consumer starts")
+        consumeKafkaMessage().subscribe()
+    }
+}