add coroutine sse and clean up

Author: xcodeassociated

http/server-events.http

@@ -2,8 +2,3 @@
 GET {{host}}/coroutine/currency-rate/current
 Accept: text/event-stream
 Authorization: Bearer {{oauthToken}}
-
-### Get server event stream of user notification
-GET {{host}}/coroutine/notification/user
-Accept: text/event-stream
-Authorization: Bearer {{oauthToken}}

http/update.http

@@ -1,4 +1,7 @@
 ###
 GET {{host}}/update/user
 Content-Type: text/event-stream
-#Authorization: Bearer {{oauthToken}}
+
+###
+GET {{host}}/update/user/reactive
+Content-Type: text/event-stream

src/main/kotlin/com/softeno/template/SoftenoReactiveMongoApp.kt

@@ -1,13 +1,6 @@
 package com.softeno.template
 
 import com.softeno.template.playground.CoroutinePlayground
-import com.softeno.template.sample.http.internal.serverevents.UserNotificationService
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.launchIn
-import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.runBlocking
 import org.slf4j.LoggerFactory
 import org.springframework.boot.autoconfigure.SpringBootApplication
 import org.springframework.boot.context.event.ApplicationReadyEvent
@@ -32,22 +25,12 @@ fun main(args: Array<String>) {
 
 @Component
 @Profile(value = ["!integration"])
-class SpringApplicationReadyEventListener(private val userNotificationService: UserNotificationService) {
+class SpringApplicationReadyEventListener {
     private val logger = LoggerFactory.getLogger(this::class.java)
 
     @EventListener(ApplicationReadyEvent::class)
     fun onApplicationReady() {
-        logger.info(">> Application Ready")
-
-        runBlocking {
-            val subscribingScope = CoroutineScope(SupervisorJob())
-            userNotificationService.stream()
-                .onEach { logger.info(">> UserNotificationService: $it") }
-                .launchIn(subscribingScope)
-
-            logger.info(">> UserNotificationService: Subscribed to stream")
-            while (true) { delay(Long.MAX_VALUE) }
-        }
+        logger.info("Application Ready")
     }
 }
 

src/main/kotlin/com/softeno/template/app/common/PrincipalHandler.kt

@@ -10,7 +10,7 @@ import java.security.Principal
 interface PrincipalHandler {
     suspend fun showPrincipal(log: Log, monoPrincipal: Mono<Principal>){
         val principal = monoPrincipal.awaitSingleOrNull()
-        log.info("principal: $principal, name: ${principal?.name}")
+        log.debug("principal: $principal, name: ${principal?.name}")
         val authentication = ReactiveSecurityContextHolder.getContext().map { it.authentication }.awaitSingleOrNull()
         if (authentication != null) {
             val token = (authentication as JwtAuthenticationToken).token

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

@@ -0,0 +1,34 @@
+package com.softeno.template.app.event
+
+import com.softeno.template.app.kafka.KafkaMessage
+import com.softeno.template.app.kafka.ReactiveKafkaSampleProducer
+import kotlinx.coroutines.DelicateCoroutinesApi
+import org.apache.commons.logging.LogFactory
+import org.slf4j.MDC
+import org.springframework.context.ApplicationEvent
+import org.springframework.context.ApplicationListener
+import org.springframework.stereotype.Component
+
+data class UserAction(val source: String, val traceId: String? = null, val spanId: String? = null) : ApplicationEvent(source)
+
+@Component
+class SampleApplicationEventPublisher(
+
+    private val reactiveKafkaProducer: ReactiveKafkaSampleProducer,
+) : ApplicationListener<UserAction> {
+    private val log = LogFactory.getLog(javaClass)
+
+    @OptIn(DelicateCoroutinesApi::class)
+    override fun onApplicationEvent(event: UserAction) {
+        // note: propagate traceId and spanId in MDC context
+        if (!event.spanId.isNullOrBlank() && !event.traceId.isNullOrBlank()) {
+            MDC.put("traceId", event.traceId)
+            MDC.put("spanId", event.spanId)
+        }
+        log.debug("[app event handler]: Received event: $event")
+        reactiveKafkaProducer.send(event.toKafkaMessage())
+    }
+
+}
+
+fun UserAction.toKafkaMessage() = KafkaMessage(content = this.source, traceId = this.traceId, spanId = this.spanId)

src/main/kotlin/com/softeno/template/app/event/UserAction.kt

@@ -1,8 +1,9 @@
 package com.softeno.template.app.kafka
 
+import com.fasterxml.jackson.annotation.JsonCreator
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.ObjectMapper
-import com.softeno.template.app.kafka.dto.KeycloakUserEvent
 import org.apache.commons.logging.LogFactory
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.springframework.beans.factory.annotation.Qualifier
@@ -38,4 +39,31 @@ class ReactiveKafkaKeycloakController(
         log.info("[kafka]: keycloak consumer starts")
         consumeKafkaMessage().subscribe()
     }
-}
+}
+
+enum class KeycloakEventType {
+    LOGIN,
+    LOGOUT,
+    REGISTER;
+
+    companion object {
+        @JsonCreator
+        fun byString(input: String): KeycloakEventType? {
+            return KeycloakEventType.entries.firstOrNull { it.name.equals(input, true) }
+        }
+    }
+}
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class KeycloakUserEvent(
+    val id: String,
+    val time: Long,
+    val type: KeycloakEventType,
+    val realmId: String,
+    val clientId: String?,
+    val userId: String,
+    val sessionId: String?,
+    val ipAddress: String,
+    val error: String?,
+    val details: Map<String, String>
+)

src/main/kotlin/com/softeno/template/app/kafka/KafkaKeycloakController.kt

@@ -1,13 +1,17 @@
-package com.softeno.template.sample.kafka
+package com.softeno.template.app.kafka
 
 
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.softeno.template.app.kafka.config.KafkaApplicationProperties
-import com.softeno.template.app.kafka.dto.KafkaMessage
-import io.micrometer.observation.ObservationRegistry
+import com.softeno.template.app.user.notification.CoroutineUserUpdateEmitter
+import com.softeno.template.app.user.notification.ReactiveUserUpdateEmitter
+import com.softeno.template.sample.websocket.Message
+import com.softeno.template.sample.websocket.ReactiveMessageService
 import io.micrometer.tracing.Span
 import io.micrometer.tracing.Tracer
+import kotlinx.coroutines.DelicateCoroutinesApi
 import org.apache.commons.lang3.RandomUtils
 import org.apache.commons.logging.LogFactory
 import org.apache.kafka.clients.consumer.ConsumerRecord
@@ -18,20 +22,25 @@ import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate
 import org.springframework.stereotype.Controller
 import org.springframework.stereotype.Service
 import reactor.core.publisher.Flux
-import reactor.kafka.receiver.ReceiverOptions
 import reactor.kafka.sender.SenderResult
 
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class KafkaMessage(val content: String, val traceId: String? = null, val spanId: String? = null)
+
+fun KafkaMessage.toMessage() = Message(from = "SYSTEM", to = "ALL", content = this.content)
 
 @Controller
 class ReactiveKafkaSampleController(
     @Qualifier(value = "kafkaSampleConsumerTemplate") private val reactiveKafkaConsumerTemplate: ReactiveKafkaConsumerTemplate<String, JsonNode>,
     private val objectMapper: ObjectMapper,
     private val tracer: Tracer,
-    private val observationRegistry: ObservationRegistry,
-    @Qualifier(value = "kafkaSampleOptions") private val kafkaReceiverOptions: ReceiverOptions<String, JsonNode>
+    private val reactiveMessageService: ReactiveMessageService,
+    private val reactiveUserUpdateEmitter: ReactiveUserUpdateEmitter,
+    private val userUpdateEmitter: CoroutineUserUpdateEmitter,
 ) : CommandLineRunner {
     private val log = LogFactory.getLog(javaClass)
 
+    @OptIn(DelicateCoroutinesApi::class)
     private fun consumeKafkaMessage(): Flux<JsonNode> {
         return reactiveKafkaConsumerTemplate
             .receiveAutoAck()
@@ -56,7 +65,9 @@ class ReactiveKafkaSampleController(
                     val span = tracer.nextSpan().name("kafka-consumer")
                     tracer.withSpan(span.start()).use {
                         log.info("[kafka] rx sample: $kafkaMessage")
-                        // additional processing ...
+                        reactiveMessageService.broadcast(kafkaMessage.toMessage())
+                        reactiveUserUpdateEmitter.broadcast(kafkaMessage.toMessage())
+                        userUpdateEmitter.broadcast(kafkaMessage.toMessage())
                     }
                     span.end()
                 }

src/main/kotlin/com/softeno/template/sample/kafka/KafkaSampleHandler.kt renamed to src/main/kotlin/com/softeno/template/app/kafka/KafkaSampleHandler.kt

@@ -1,7 +1,7 @@
 package com.softeno.template.app.kafka.config
 
 import com.fasterxml.jackson.databind.JsonNode
-import com.softeno.template.app.kafka.dto.KafkaMessage
+import com.softeno.template.app.kafka.KafkaMessage
 import io.micrometer.observation.ObservationRegistry
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties