Fix opex orders

auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/proxy/KeycloakProxy.kt

@@ -174,7 +174,7 @@ class KeycloakProxy(
                     ).apply {
                         if (username.type == UsernameType.MOBILE)
                             put("mobile", username.value)
-                        put(Attributes.OTP, username.type.otpType.name)
+                        put(Attributes.OTP, OTPType.NONE.name)
                     }
                 ).apply { if (username.type == UsernameType.EMAIL) put("email", username.value) }
             )

matching-engine/matching-engine-app/src/main/kotlin/co/nilin/opex/matching/engine/app/bl/ExchangeEventHandler.kt

@@ -3,15 +3,19 @@ package co.nilin.opex.matching.engine.app.bl
 import co.nilin.opex.matching.engine.app.config.AppSchedulers
 import co.nilin.opex.matching.engine.core.eventh.EventDispatcher
 import co.nilin.opex.matching.engine.core.eventh.events.*
+import co.nilin.opex.matching.engine.core.inout.OrderBookUpdateEvent
 import co.nilin.opex.matching.engine.core.spi.OrderBookPersister
 import co.nilin.opex.matching.engine.ports.kafka.submitter.service.EventsSubmitter
+import co.nilin.opex.matching.engine.ports.kafka.submitter.service.OrderBookUpdateSubmitter
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.launch
 import org.springframework.stereotype.Component
 
 @Component
 class ExchangeEventHandler(
-    eventsSubmitter: EventsSubmitter, orderBookPersister: OrderBookPersister
+    private val eventsSubmitter: EventsSubmitter,
+    private val orderBookPersister: OrderBookPersister,
+    private val orderBookSubmitter: OrderBookUpdateSubmitter
 ) {
     fun register() {
         EventDispatcher.register(CreateOrderEvent::class.java, handler)
@@ -32,6 +36,12 @@ class ExchangeEventHandler(
     val localHandler: (OrderBookPublishedEvent) -> Unit = {
         CoroutineScope(AppSchedulers.generalExecutor).launch {
             orderBookPersister.storeLastState(it.persistentOrderBook)
+            val orderBookUpdate = OrderBookUpdateEvent(
+                it.persistentOrderBook.pair,
+                System.currentTimeMillis(),
+                it.persistentOrderBook.orders ?: emptyList()
+            )
+            orderBookSubmitter.submit(orderBookUpdate)
         }
     }
 

matching-engine/matching-engine-app/src/main/kotlin/co/nilin/opex/matching/engine/app/bl/OrderBooks.kt

@@ -6,12 +6,13 @@ import co.nilin.opex.matching.engine.core.model.Pair
 import co.nilin.opex.matching.engine.core.model.PersistentOrderBook
 
 object OrderBooks {
+
     private val orderBooks = mutableMapOf<String, OrderBook>()
 
     fun createOrderBook(pair: String) {
         println("Going to add order book:" + pair + ", current order books#" + orderBooks.size)
         if (orderBooks.containsKey(pair))
-            throw IllegalArgumentException("${pair} has an order book right now!")
+            throw IllegalArgumentException("$pair has an order book right now!")
         val pairs = pair.split("_")
         orderBooks[pair] = OrderBookFactory.createOrderBook(Pair(pairs[0], pairs[1]))
         println("order book:" + pair + " added, current order books#" + orderBooks.size)

matching-engine/matching-engine-core/src/main/kotlin/co/nilin/opex/matching/engine/core/eventh/EventDispatcher.kt

@@ -18,7 +18,6 @@ object EventDispatcher {
         eventsHandler.getOrPut(type, { LinkedList() }).add(listener)
     }
 
-
     fun emit(event: CoreEvent) {
         var type: Class<*>? = event::class.java
         while (type != null) {
@@ -31,7 +30,6 @@ object EventDispatcher {
         }
     }
 
-
     open class EventListener<T>(
         val lambda: (T) -> Unit
     ) {

matching-engine/matching-engine-core/src/main/kotlin/co/nilin/opex/matching/engine/core/inout/OrderBookUpdateEvent.kt

@@ -0,0 +1,10 @@
+package co.nilin.opex.matching.engine.core.inout
+
+import co.nilin.opex.matching.engine.core.model.Pair
+import co.nilin.opex.matching.engine.core.model.PersistentOrder
+
+data class OrderBookUpdateEvent(
+    val pair: Pair,
+    val time: Long,
+    val orders: List<PersistentOrder>
+)

matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/config/EventsKafkaConfig.kt

@@ -1,6 +1,7 @@
 package co.nilin.opex.matching.engine.ports.kafka.submitter.config
 
 import co.nilin.opex.matching.engine.core.eventh.events.CoreEvent
+import co.nilin.opex.matching.engine.core.inout.OrderBookUpdateEvent
 import co.nilin.opex.matching.engine.core.inout.OrderRequestEvent
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.serialization.StringSerializer
@@ -11,6 +12,7 @@ import org.springframework.context.annotation.Configuration
 import org.springframework.kafka.core.DefaultKafkaProducerFactory
 import org.springframework.kafka.core.KafkaTemplate
 import org.springframework.kafka.core.ProducerFactory
+import org.springframework.kafka.support.serializer.JsonDeserializer
 import org.springframework.kafka.support.serializer.JsonSerializer
 
 @Configuration
@@ -25,7 +27,9 @@ class EventsKafkaConfig {
             ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,
             ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
             ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to JsonSerializer::class.java,
-            ProducerConfig.ACKS_CONFIG to "all"
+            ProducerConfig.ACKS_CONFIG to "all",
+            JsonDeserializer.TRUSTED_PACKAGES to "co.nilin.opex.*",
+            JsonDeserializer.TYPE_MAPPINGS to "orderBookUpdate:co.nilin.opex.matching.engine.core.inout.OrderBookUpdateEvent"
         )
     }
 
@@ -49,4 +53,9 @@ class EventsKafkaConfig {
         return KafkaTemplate(producerFactory)
     }
 
+    @Bean("orderBookUpdateTemplate")
+    fun orderBookUpdateTemplate(@Qualifier("producerConfigs") producerConfigs: Map<String, Any>): KafkaTemplate<String?, OrderBookUpdateEvent> {
+        return KafkaTemplate(DefaultKafkaProducerFactory(producerConfigs))
+    }
+
 }

matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/config/KafkaTopicConfig.kt

@@ -48,6 +48,13 @@ class KafkaTopicConfig {
                         .build()
                 })
             }
+
+        applicationContext.registerBean("topic_orderBookUpdate", NewTopic::class.java, Supplier {
+            TopicBuilder.name("orderBookUpdate")
+                .partitions(10)
+                .replicas(1)
+                .build()
+        })
     }
 
 }

matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/service/OrderBookUpdateSubmitter.kt

@@ -0,0 +1,34 @@
+package co.nilin.opex.matching.engine.ports.kafka.submitter.service
+
+import co.nilin.opex.matching.engine.core.inout.OrderBookUpdateEvent
+import org.slf4j.LoggerFactory
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.stereotype.Component
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
+import kotlin.coroutines.suspendCoroutine
+
+@Component
+class OrderBookUpdateSubmitter(private val kafkaTemplate: KafkaTemplate<String?, OrderBookUpdateEvent>) {
+
+    private val logger = LoggerFactory.getLogger(OrderBookUpdateSubmitter::class.java)
+    private var lastSent = 0L
+
+    suspend fun submit(event: OrderBookUpdateEvent): Unit = suspendCoroutine { cont ->
+        val now = System.currentTimeMillis()
+        if (now - lastSent < 1000) // send every second not more
+            return@suspendCoroutine
+
+        kafkaTemplate.send("orderBookUpdate", event).addCallback(
+            {
+                logger.info("Orderbook update event sent")
+                lastSent = System.currentTimeMillis()
+                cont.resume(Unit)
+            },
+            {
+                logger.warn("Orderbook update event failed", it)
+                cont.resumeWithException(it)
+            }
+        )
+    }
+}