(Near) realtime update of statistics charts

4 files changed, 59 insertions, 10 deletions

diff --git a/src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt b/src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt
index ed3be5d..e4a97ee 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/config/AppConfiguration.kt
@@ -27,12 +27,16 @@ import dev.dnpm.etl.processor.pseudonym.AnonymizingGenerator
 import dev.dnpm.etl.processor.pseudonym.Generator
 import dev.dnpm.etl.processor.pseudonym.GpasPseudonymGenerator
 import dev.dnpm.etl.processor.pseudonym.PseudonymizeService
+import org.reactivestreams.Publisher
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.kafka.core.KafkaTemplate
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Sinks
 import java.net.URI
+import java.time.Duration
 
 @Configuration
 @EnableConfigurationProperties(
@@ -78,5 +82,10 @@ class AppConfiguration {
         return KafkaMtbFileSender(kafkaTemplate, objectMapper)
     }
 
+    @Bean
+    fun statisticsUpdateProducer(): Sinks.Many<Any> {
+        return Sinks.many().multicast().directBestEffort()
+    }
+
 }
 
diff --git a/src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt b/src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt
index 835f3de..9cbb52a 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/web/MtbFileController.kt
@@ -34,13 +34,15 @@ import org.springframework.http.ResponseEntity
 import org.springframework.web.bind.annotation.PostMapping
 import org.springframework.web.bind.annotation.RequestBody
 import org.springframework.web.bind.annotation.RestController
+import reactor.core.publisher.Sinks
 
 @RestController
 class MtbFileController(
     private val pseudonymizeService: PseudonymizeService,
     private val senders: List<MtbFileSender>,
     private val requestRepository: RequestRepository,
-    private val objectMapper: ObjectMapper
+    private val objectMapper: ObjectMapper,
+    private val statisticsUpdateProducer: Sinks.Many<Any>
 ) {
 
     private val logger = LoggerFactory.getLogger(MtbFileController::class.java)
@@ -63,6 +65,7 @@ class MtbFileController(
                     report = Report("Duplikat erkannt - keine Daten weitergeleitet")
                 )
             )
+            statisticsUpdateProducer.emitNext("", Sinks.EmitFailureHandler.FAIL_FAST)
             return ResponseEntity.noContent().build()
         }
 
@@ -110,6 +113,8 @@ class MtbFileController(
             )
         )
 
+        statisticsUpdateProducer.emitNext("", Sinks.EmitFailureHandler.FAIL_FAST)
+
         return if (requestStatus == RequestStatus.ERROR) {
             ResponseEntity.unprocessableEntity().build()
         } else {
diff --git a/src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsController.kt b/src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsController.kt
index 05b84ff..adc1e2b 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsController.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsController.kt
@@ -20,15 +20,18 @@
 package dev.dnpm.etl.processor.web
 
 import org.springframework.stereotype.Controller
+import org.springframework.ui.Model
 import org.springframework.web.bind.annotation.GetMapping
 import org.springframework.web.bind.annotation.RequestMapping
+import java.time.Instant
 
 @Controller
 @RequestMapping(path = ["/statistics"])
 class StatisticsController {
 
     @GetMapping
-    fun index(): String {
+    fun index(model: Model): String {
+        model.addAttribute("now", Instant.now())
         return "statistics"
     }
 
diff --git a/src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsRestController.kt b/src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsRestController.kt
index 8d5cb0e..2741fd3 100644
--- a/src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsRestController.kt
+++ b/src/main/kotlin/dev/dnpm/etl/processor/web/StatisticsRestController.kt
@@ -21,9 +21,15 @@ package dev.dnpm.etl.processor.web
 
 import dev.dnpm.etl.processor.monitoring.RequestRepository
 import dev.dnpm.etl.processor.monitoring.RequestStatus
+import org.reactivestreams.Publisher
+import org.springframework.http.MediaType
+import org.springframework.http.codec.ServerSentEvent
 import org.springframework.web.bind.annotation.GetMapping
 import org.springframework.web.bind.annotation.RequestMapping
 import org.springframework.web.bind.annotation.RestController
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Sinks
+import reactor.kotlin.core.publisher.toFlux
 import java.time.Instant
 import java.time.Month
 import java.time.ZoneId
@@ -34,6 +40,7 @@ import java.time.temporal.TemporalUnit
 @RestController
 @RequestMapping(path = ["/statistics"])
 class StatisticsRestController(
+    private val statisticsUpdateProducer: Sinks.Many<Any>,
     private val requestRepository: RequestRepository
 ) {
 
@@ -68,13 +75,15 @@ class StatisticsRestController(
                     .toMap()
                 Pair(
                     it.key.toString(),
-                    DateNameValues(it.key.toString(), NameValues(
-                        error = requestList[RequestStatus.ERROR] ?: 0,
-                        warning = requestList[RequestStatus.WARNING] ?: 0,
-                        success = requestList[RequestStatus.SUCCESS] ?: 0,
-                        duplication = requestList[RequestStatus.DUPLICATION] ?: 0,
-                        unknown = requestList[RequestStatus.UNKNOWN] ?: 0,
-                    ))
+                    DateNameValues(
+                        it.key.toString(), NameValues(
+                            error = requestList[RequestStatus.ERROR] ?: 0,
+                            warning = requestList[RequestStatus.WARNING] ?: 0,
+                            success = requestList[RequestStatus.SUCCESS] ?: 0,
+                            duplication = requestList[RequestStatus.DUPLICATION] ?: 0,
+                            unknown = requestList[RequestStatus.UNKNOWN] ?: 0,
+                        )
+                    )
                 )
             }.toMap()
 
@@ -86,10 +95,33 @@ class StatisticsRestController(
             .sortedBy { it.date }
     }
 
+    @GetMapping(path = ["events"], produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
+    fun updater(): Flux<ServerSentEvent<Any>> {
+        return statisticsUpdateProducer.asFlux().flatMap {
+            Flux.fromIterable(
+                listOf(
+                    ServerSentEvent.builder<Any>()
+                        .event("requeststates").id("none").data(this.requestStates())
+                        .build(),
+                    ServerSentEvent.builder<Any>()
+                        .event("requestslastmonth").id("none").data(this.requestsLastMonth())
+                        .build()
+                )
+            )
+
+        }
+    }
+
 }
 
 data class NameValue(val name: String, val value: Int, val color: String)
 
 data class DateNameValues(val date: String, val nameValues: NameValues)
 
-data class NameValues(val error: Int = 0, val warning: Int = 0, val success: Int = 0, val duplication: Int = 0, val unknown: Int = 0)
\ No newline at end of file
+data class NameValues(
+    val error: Int = 0,
+    val warning: Int = 0,
+    val success: Int = 0,
+    val duplication: Int = 0,
+    val unknown: Int = 0
+)
\ No newline at end of file