refactor: use event handlers defined at the application level

Author: AndreaGiulianelli

src/main/kotlin/infrastructure/events/KafkaClient.kt

@@ -10,37 +10,38 @@ package infrastructure.events
 
 import application.controller.MedicalTechnologyController
 import application.controller.RoomController
-import application.presenter.event.model.MedicalTechnologyEvent
-import application.presenter.event.model.MedicalTechnologyEventKey
-import application.presenter.event.model.RoomEvent
-import application.presenter.event.model.RoomEventKey
-import application.presenter.event.model.RoomEventPayloads
-import application.presenter.event.serialization.EventSerialization.toHumidity
-import application.presenter.event.serialization.EventSerialization.toLuminosity
-import application.presenter.event.serialization.EventSerialization.toPresence
-import application.presenter.event.serialization.EventSerialization.toTemperature
-import application.service.MedicalTechnologyService
-import application.service.RoomService
+import application.handler.EventHandler
+import application.handler.EventHandlers
+import application.presenter.event.serialization.EventSerialization.toEvent
 import com.fasterxml.jackson.databind.ObjectMapper
-import entity.medicaltechnology.MedicalTechnologyID
-import entity.zone.RoomEnvironmentalData
-import entity.zone.RoomID
 import infrastructure.provider.ManagerProvider
-import kotlinx.serialization.decodeFromString
-import kotlinx.serialization.json.Json
+import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.clients.consumer.KafkaConsumer
 import java.time.Duration
-import java.time.Instant
 import java.time.format.DateTimeParseException
 
 /**
  * This class manage the Kafka client needed to consume events.
  * @param[provider] the provider of managers.
  */
 class KafkaClient(private val provider: ManagerProvider) {
+    private val eventHandlers: List<EventHandler>
+
     init {
         checkNotNull(System.getenv(bootstrapServerUrlVariable)) { "kafka bootstrap server url required" }
         checkNotNull(System.getenv(schemaRegistryUrlVariable)) { "kafka schema registry url required" }
+        val roomController = RoomController(provider.roomDigitalTwinManager, provider.roomDatabaseManager)
+        val medicalTechnologyController = MedicalTechnologyController(
+            provider.medicalTechnologyDigitalTwinManager,
+            provider.medicalTechnologyDatabaseManager
+        )
+        eventHandlers = listOf(
+            EventHandlers.TemperatureEventHandler(roomController),
+            EventHandlers.HumidityEventHandler(roomController),
+            EventHandlers.LuminosityEventHandler(roomController),
+            EventHandlers.PresenceEventHandler(roomController),
+            EventHandlers.MedicalTechnologyEventHandler(medicalTechnologyController)
+        )
     }
 
     private val kafkaConsumer = KafkaConsumer<String, String>(
@@ -58,7 +59,7 @@ class KafkaClient(private val provider: ManagerProvider) {
             while (true) {
                 kafkaConsumer.poll(Duration.ofMillis(pollingTime)).forEach { event ->
                     try {
-                        consumeEvent(event.key(), ObjectMapper().writeValueAsString(event.value()))
+                        consumeEvent(event)
                     } catch (e: IllegalArgumentException) {
                         println("ERROR: Invalid Event Schema. Event discarded! - $e")
                     } catch (e: DateTimeParseException) {
@@ -69,58 +70,11 @@ class KafkaClient(private val provider: ManagerProvider) {
         }
     }
 
-    private fun consumeEvent(eventKey: String, event: String) {
-        fun updateRoomEnvironmentData(event: RoomEvent<*>, environmentalData: RoomEnvironmentalData) =
-            RoomService.UpdateRoomEnvironmentData(
-                RoomID(event.roomId),
-                environmentalData,
-                Instant.parse(event.dateTime),
-                RoomController(provider.roomDigitalTwinManager, provider.roomDatabaseManager)
-            ).execute()
-
-        when (eventKey) {
-            RoomEventKey.TEMPERATURE_EVENT -> {
-                val deserializedEvent = Json.decodeFromString<RoomEvent<RoomEventPayloads.TemperaturePayload>>(event)
-                updateRoomEnvironmentData(
-                    deserializedEvent,
-                    RoomEnvironmentalData(temperature = deserializedEvent.data.toTemperature())
-                )
-            }
-            RoomEventKey.HUMIDITY_EVENT -> {
-                val deserializedEvent = Json.decodeFromString<RoomEvent<RoomEventPayloads.HumidityPayload>>(event)
-                updateRoomEnvironmentData(
-                    deserializedEvent,
-                    RoomEnvironmentalData(humidity = deserializedEvent.data.toHumidity())
-                )
-            }
-            RoomEventKey.LUMINOSITY_EVENT -> {
-                val deserializedEvent = Json.decodeFromString<RoomEvent<RoomEventPayloads.LuminosityPayload>>(event)
-                updateRoomEnvironmentData(
-                    deserializedEvent,
-                    RoomEnvironmentalData(luminosity = deserializedEvent.data.toLuminosity())
-                )
-            }
-            RoomEventKey.PRESENCE_EVENT -> {
-                val deserializedEvent = Json.decodeFromString<RoomEvent<RoomEventPayloads.PresencePayload>>(event)
-                updateRoomEnvironmentData(
-                    deserializedEvent,
-                    RoomEnvironmentalData(presence = deserializedEvent.data.toPresence())
-                )
-            }
-            MedicalTechnologyEventKey.USAGE_EVENT -> {
-                val deserializedEvent = Json.decodeFromString<MedicalTechnologyEvent>(event)
-                MedicalTechnologyService.UpdateMedicalTechnologyUsage(
-                    MedicalTechnologyID(deserializedEvent.data.medicalTechnologyID),
-                    deserializedEvent.data.isInUse,
-                    Instant.parse(deserializedEvent.dateTime),
-                    MedicalTechnologyController(
-                        provider.medicalTechnologyDigitalTwinManager,
-                        provider.medicalTechnologyDatabaseManager
-                    )
-                )
-            }
-            else -> throw IllegalArgumentException("Event not supported")
-        }
+    private fun consumeEvent(event: ConsumerRecord<String, String>) {
+        val deserializedEvent = ObjectMapper().writeValueAsString(event.value()).toEvent(event.key())
+        this@KafkaClient.eventHandlers
+            .filter { it.canHandle(deserializedEvent) }
+            .forEach { it.consume(deserializedEvent) }
     }
 
     companion object {