Profile, KYC and config modules added

.github/workflows/dev.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        java: [ 11 ]
+        java: [ 17 ]
     name: Build OPEX and run tests with java ${{ matrix.java }}
     steps:
       - name: Checkout Source Code

.github/workflows/main.yml

@@ -10,7 +10,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        java: [ 11 ]
+        java: [ 17 ]
     name: Build OPEX and run tests with java ${{ matrix.java }}
     steps:
       - name: Checkout Source Code

.github/workflows/pr.yml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        java: [ 11 ]
+        java: [ 17 ]
     name: Build OPEX and run tests with java ${{ matrix.java }}
     steps:
       - name: Checkout Source Code

accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/config/AppConfig.kt

@@ -1,5 +1,6 @@
 package co.nilin.opex.accountant.app.config
 
+import co.nilin.opex.accountant.app.listener.*
 import co.nilin.opex.accountant.app.listener.AccountantEventListener
 import co.nilin.opex.accountant.app.listener.AccountantTempEventListener
 import co.nilin.opex.accountant.app.listener.AccountantTradeListener
@@ -12,11 +13,11 @@ import co.nilin.opex.accountant.core.service.FinancialActionJobManagerImpl
 import co.nilin.opex.accountant.core.service.OrderManagerImpl
 import co.nilin.opex.accountant.core.service.TradeManagerImpl
 import co.nilin.opex.accountant.core.spi.*
+import co.nilin.opex.accountant.ports.kafka.listener.consumer.*
 import co.nilin.opex.accountant.ports.kafka.listener.consumer.EventKafkaListener
 import co.nilin.opex.accountant.ports.kafka.listener.consumer.OrderKafkaListener
 import co.nilin.opex.accountant.ports.kafka.listener.consumer.TempEventKafkaListener
 import co.nilin.opex.accountant.ports.kafka.listener.consumer.TradeKafkaListener
-import com.fasterxml.jackson.databind.ObjectMapper
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
@@ -113,6 +114,10 @@ class AppConfig {
         return AccountantTempEventListener(orderManager, tradeManager)
     }
 
+
+
+
+
     @Autowired
     fun configureOrderListener(orderKafkaListener: OrderKafkaListener, orderListener: OrderListener) {
         orderKafkaListener.addListener(orderListener)
@@ -128,16 +133,21 @@ class AppConfig {
 
     @Autowired
     fun configureEventListener(
-        eventKafkaListener: EventKafkaListener,
-        accountantEventListener: AccountantEventListener
+            eventKafkaListener: EventKafkaListener,
+            accountantEventListener: AccountantEventListener,
+            kycLevelUpdatedKafkaListener: KycLevelUpdatedKafkaListener,
+            kycLevelUpdatedEventListener: KycLevelUpdatedListener
     ) {
         eventKafkaListener.addListener(accountantEventListener)
+        kycLevelUpdatedKafkaListener.addEventListener(kycLevelUpdatedEventListener)
+
     }
 
     @Autowired
     fun configureTempEventListener(
         tempEventKafkaListener: TempEventKafkaListener,
-        accountantTempEventListener: AccountantTempEventListener
+        accountantTempEventListener: AccountantTempEventListener,
+
     ) {
         tempEventKafkaListener.addListener(accountantTempEventListener)
     }

accountant/accountant-app/src/main/kotlin/co/nilin/opex/accountant/app/listener/KycLevelUpdatedListener.kt

@@ -0,0 +1,32 @@
+package co.nilin.opex.accountant.app.listener
+
+import co.nilin.opex.accountant.core.inout.KycLevelUpdatedEvent
+import co.nilin.opex.accountant.ports.kafka.listener.spi.KycLevelUpdatedEventListener
+import co.nilin.opex.accountant.ports.postgres.impl.UserLevelLoaderImpl
+import org.slf4j.LoggerFactory
+import org.springframework.stereotype.Component
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+
+@Component
+class KycLevelUpdatedListener(val userLevelLoaderImpl: UserLevelLoaderImpl) : KycLevelUpdatedEventListener {
+
+    private val logger = LoggerFactory.getLogger(KycLevelUpdatedListener::class.java)
+    val scope = CoroutineScope(Dispatchers.IO)
+    override fun id(): String {
+        return "KycLevelUpdatedListener"
+    }
+
+    override fun onEvent(event: KycLevelUpdatedEvent,
+                         partition: Int, offset: Long, timestamp: Long, eventId: String) {
+        logger.info("==========================================================================")
+        logger.info("Incoming UserLevelUpdated event: $event")
+        logger.info("==========================================================================")
+        scope.launch {
+            userLevelLoaderImpl.update(event.userId, event.kycLevel)
+        }
+    }
+
+
+}

accountant/accountant-core/pom.xml

@@ -58,9 +58,13 @@
             <artifactId>mockk</artifactId>
         </dependency>
         <dependency>
-            <groupId>com.google.code.gson</groupId>
-            <artifactId>gson</artifactId>
-            <scope>test</scope>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+            <version>2.15.2</version>
         </dependency>
     </dependencies>
 </project>

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/inout/KycLevelUpdatedEvent.kt

@@ -0,0 +1,7 @@
+package co.nilin.opex.accountant.core.inout
+
+
+import co.nilin.opex.accountant.core.model.KycLevel
+import java.time.LocalDateTime
+
+data class KycLevelUpdatedEvent(var userId: String, var kycLevel: KycLevel, var updateDate: LocalDateTime)

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/model/KycLevel.kt

@@ -0,0 +1,6 @@
+package co.nilin.opex.accountant.core.model
+
+enum class KycLevel {
+    Level1, Level2,
+}
+

accountant/accountant-core/src/main/kotlin/co/nilin/opex/accountant/core/spi/UserLevelLoader.kt

@@ -1,7 +1,11 @@
 package co.nilin.opex.accountant.core.spi
 
+import co.nilin.opex.accountant.core.model.KycLevel
+
 interface UserLevelLoader {
 
     suspend fun load(uuid: String): String
 
+    suspend fun update(uuid: String,userLevel:KycLevel)
+
 }

accountant/accountant-core/src/test/kotlin/co/nilin/opex/accountant/core/service/JsonMapperTestImpl.kt

@@ -1,20 +1,23 @@
 package co.nilin.opex.accountant.core.service
 
 import co.nilin.opex.accountant.core.spi.JsonMapper
-import com.google.gson.Gson
-import org.springframework.boot.json.GsonJsonParser
+import com.fasterxml.jackson.databind.ObjectMapper
 
 class JsonMapperTestImpl : JsonMapper {
 
+    private val objectMapper = ObjectMapper().apply {
+        findAndRegisterModules()
+    }
+
     override fun serialize(input: Any): String? {
-        return Gson().toJson(input)
+        return objectMapper.writeValueAsString(input)
     }
 
     override fun <T> deserialize(input: String, t: Class<T>): T {
-        return Gson().fromJson(input, t)
+        return objectMapper.readValue(input, t)
     }
 
     override fun toMap(input: Any): Map<String, Any> {
-        return GsonJsonParser().parseMap(serialize(input))
+        return objectMapper.convertValue(input, Map::class.java) as Map<String, Any>
     }
 }

accountant/accountant-ports/accountant-eventlistener-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/listener/config/AccountantKafkaConfig.kt

@@ -1,9 +1,7 @@
 package co.nilin.opex.accountant.ports.kafka.listener.config
 
-import co.nilin.opex.accountant.ports.kafka.listener.consumer.EventKafkaListener
-import co.nilin.opex.accountant.ports.kafka.listener.consumer.OrderKafkaListener
-import co.nilin.opex.accountant.ports.kafka.listener.consumer.TempEventKafkaListener
-import co.nilin.opex.accountant.ports.kafka.listener.consumer.TradeKafkaListener
+import co.nilin.opex.accountant.core.inout.KycLevelUpdatedEvent
+import co.nilin.opex.accountant.ports.kafka.listener.consumer.*
 import co.nilin.opex.matching.engine.core.eventh.events.CoreEvent
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.common.TopicPartition
@@ -14,9 +12,7 @@ import org.springframework.beans.factory.annotation.Value
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
-import org.springframework.kafka.core.ConsumerFactory
-import org.springframework.kafka.core.DefaultKafkaConsumerFactory
-import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.kafka.core.*
 import org.springframework.kafka.listener.*
 import org.springframework.kafka.support.serializer.JsonDeserializer
 import org.springframework.util.backoff.FixedBackOff
@@ -39,7 +35,7 @@ class AccountantKafkaConfig {
             ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
             ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to JsonDeserializer::class.java,
             JsonDeserializer.TRUSTED_PACKAGES to "co.nilin.opex.*",
-            JsonDeserializer.TYPE_MAPPINGS to "order_request_event:co.nilin.opex.accountant.ports.kafka.listener.inout.OrderRequestEvent,order_request_submit:co.nilin.opex.accountant.ports.kafka.listener.inout.OrderSubmitRequestEvent,order_request_cancel:co.nilin.opex.accountant.ports.kafka.listener.inout.OrderCancelRequestEvent"
+            JsonDeserializer.TYPE_MAPPINGS to "order_request_event:co.nilin.opex.accountant.ports.kafka.listener.inout.OrderRequestEvent,order_request_submit:co.nilin.opex.accountant.ports.kafka.listener.inout.OrderSubmitRequestEvent,order_request_cancel:co.nilin.opex.accountant.ports.kafka.listener.inout.OrderCancelRequestEvent,kyc_level_updated_event:co.nilin.opex.accountant.core.inout.KycLevelUpdatedEvent"
         )
     }
 
@@ -48,6 +44,11 @@ class AccountantKafkaConfig {
         return DefaultKafkaConsumerFactory(consumerConfigs)
     }
 
+    @Bean("KycConsumerFactory")
+    fun kycConsumerFactory(@Qualifier("consumerConfig") consumerConfigs: Map<String, Any?>): ConsumerFactory<String, KycLevelUpdatedEvent> {
+        return DefaultKafkaConsumerFactory(consumerConfigs)
+    }
+
     @Autowired
     @ConditionalOnBean(TradeKafkaListener::class)
     fun configureTradeListener(
@@ -108,6 +109,31 @@ class AccountantKafkaConfig {
         container.start()
     }
 
+    @Bean("kycLevelUpdatedProducerFactory")
+    fun producerFactory(@Qualifier("consumerConfig") producerConfigs: Map<String, Any>): ProducerFactory<String, KycLevelUpdatedEvent> {
+        return DefaultKafkaProducerFactory(producerConfigs)
+    }
+
+    @Bean("kycLevelUpdatedKafkaTemplate")
+    fun kafkaTemplate(@Qualifier("kycLevelUpdatedProducerFactory") producerFactory: ProducerFactory<String, KycLevelUpdatedEvent>): KafkaTemplate<String, KycLevelUpdatedEvent> {
+        return KafkaTemplate(producerFactory)
+    }
+    @Autowired
+    @ConditionalOnBean(KycLevelUpdatedKafkaListener::class)
+    fun configureKycLevelUpdatedListener(
+            listener: KycLevelUpdatedKafkaListener,
+            @Qualifier("kycLevelUpdatedKafkaTemplate") template: KafkaTemplate<String, KycLevelUpdatedEvent>,
+            @Qualifier("KycConsumerFactory") consumerFactory: ConsumerFactory<String, KycLevelUpdatedEvent>
+    ) {
+        val containerProps = ContainerProperties(Pattern.compile("kyc_level_updated"))
+        containerProps.messageListener = listener
+        val container = ConcurrentMessageListenerContainer(consumerFactory, containerProps)
+        container.setBeanName("KycLevelUpdatedKafkaListenerContainer")
+        container.commonErrorHandler = createConsumerErrorHandler(template, "kyc_level_updated.DLT")
+        container.start()
+    }
+
+
     private fun createConsumerErrorHandler(kafkaTemplate: KafkaTemplate<*, *>, dltTopic: String): CommonErrorHandler {
         val recoverer = DeadLetterPublishingRecoverer(kafkaTemplate) { cr, _ ->
             cr.headers().add("dlt-origin-module", "ACCOUNTANT".toByteArray())

accountant/accountant-ports/accountant-eventlistener-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/listener/consumer/KycLevelUpdatedKafkaListener.kt

@@ -0,0 +1,35 @@
+package co.nilin.opex.accountant.ports.kafka.listener.consumer
+
+
+import co.nilin.opex.accountant.core.inout.KycLevelUpdatedEvent
+import co.nilin.opex.accountant.ports.kafka.listener.spi.KycLevelUpdatedEventListener
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.slf4j.LoggerFactory
+import org.springframework.kafka.listener.MessageListener
+import org.springframework.stereotype.Component
+
+@Component
+class KycLevelUpdatedKafkaListener : MessageListener<String, KycLevelUpdatedEvent> {
+    val eventListeners = arrayListOf<KycLevelUpdatedEventListener>()
+    private val logger = LoggerFactory.getLogger(KycLevelUpdatedKafkaListener::class.java)
+    override fun onMessage(data: ConsumerRecord<String, KycLevelUpdatedEvent>) {
+
+        eventListeners.forEach { tl ->
+            logger.info("incoming new event " + tl.id())
+            tl.onEvent(data.value(), data.partition(), data.offset(), data.timestamp(), tl.id())
+        }
+    }
+
+    fun addEventListener(tl: KycLevelUpdatedEventListener) {
+        eventListeners.add(tl)
+    }
+
+    fun removeEventListener(tl: KycLevelUpdatedEventListener) {
+        eventListeners.removeIf { item ->
+            item.id() == tl.id()
+        }
+
+    }
+
+
+}

accountant/accountant-ports/accountant-eventlistener-kafka/src/main/kotlin/co/nilin/opex/accountant/ports/kafka/listener/spi/KycLevelUpdatedEventListener.kt

@@ -0,0 +1,10 @@
+package co.nilin.opex.accountant.ports.kafka.listener.spi
+
+import co.nilin.opex.accountant.core.inout.KycLevelUpdatedEvent
+
+
+interface KycLevelUpdatedEventListener {
+    fun id(): String
+    fun onEvent(event: KycLevelUpdatedEvent, partition: Int, offset: Long, timestamp: Long, eventId: String)
+
+}

accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/dao/UserLevelRepository.kt

@@ -9,6 +9,7 @@ import reactor.core.publisher.Mono
 @Repository
 interface UserLevelRepository : ReactiveCrudRepository<UserLevelModel, String> {
 
+    fun findByLevel(level: String): Mono<UserLevelModel>
     @Query("insert into user_level (level) values (:level) on conflict do nothing")
     fun insert(level: String): Mono<Void>
 

accountant/accountant-ports/accountant-persister-postgres/src/main/kotlin/co/nilin/opex/accountant/ports/postgres/impl/TempEventPersisterImpl.kt

@@ -5,7 +5,8 @@ import co.nilin.opex.accountant.core.spi.TempEventPersister
 import co.nilin.opex.accountant.ports.postgres.dao.TempEventRepository
 import co.nilin.opex.accountant.ports.postgres.model.TempEventModel
 import co.nilin.opex.matching.engine.core.eventh.events.CoreEvent
-import com.google.gson.Gson
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.module.kotlin.readValue
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.reactive.awaitFirstOrNull
@@ -16,15 +17,18 @@ import org.springframework.stereotype.Component
 import java.time.LocalDateTime
 
 @Component
-class TempEventPersisterImpl(private val tempEventRepository: TempEventRepository) : TempEventPersister {
+class TempEventPersisterImpl(
+    private val tempEventRepository: TempEventRepository,
+    private val objectMapper: ObjectMapper
+) : TempEventPersister {
 
     override suspend fun saveTempEvent(ouid: String, event: CoreEvent) {
         tempEventRepository.save(
             TempEventModel(
                 null,
                 ouid,
                 event.javaClass.name,
-                Gson().toJson(event),
+                objectMapper.writeValueAsString(event),
                 LocalDateTime.now()
             )
         ).awaitSingleOrNull()
@@ -33,7 +37,7 @@ class TempEventPersisterImpl(private val tempEventRepository: TempEventRepositor
     override suspend fun loadTempEvents(ouid: String): List<CoreEvent> {
         return tempEventRepository
             .findByOuid(ouid)
-            .map { Gson().fromJson(it.eventBody, Class.forName(it.eventType)) as CoreEvent }
+            .map { objectMapper.readValue<CoreEvent>(it.eventBody) }
             .toList()
     }
 
@@ -52,7 +56,7 @@ class TempEventPersisterImpl(private val tempEventRepository: TempEventRepositor
                 TempEvent(
                     it.id!!,
                     it.ouid,
-                    Gson().fromJson(it.eventBody, Class.forName(it.eventType)) as CoreEvent,
+                    objectMapper.readValue(it.eventBody),
                     it.eventDate
                 )
             }