diff --git a/auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/proxy/KeycloakProxy.kt b/auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/proxy/KeycloakProxy.kt index c8bd1778f..7797d9ea5 100644 --- a/auth-gateway/auth-gateway-app/src/main/kotlin/co/nilin/opex/auth/proxy/KeycloakProxy.kt +++ b/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) } ) diff --git a/matching-engine/matching-engine-app/src/main/kotlin/co/nilin/opex/matching/engine/app/bl/ExchangeEventHandler.kt b/matching-engine/matching-engine-app/src/main/kotlin/co/nilin/opex/matching/engine/app/bl/ExchangeEventHandler.kt index 251d7c889..e967cdebc 100644 --- a/matching-engine/matching-engine-app/src/main/kotlin/co/nilin/opex/matching/engine/app/bl/ExchangeEventHandler.kt +++ b/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) } } diff --git a/matching-engine/matching-engine-app/src/main/kotlin/co/nilin/opex/matching/engine/app/bl/OrderBooks.kt b/matching-engine/matching-engine-app/src/main/kotlin/co/nilin/opex/matching/engine/app/bl/OrderBooks.kt index af298bbf8..95f0fc30a 100644 --- a/matching-engine/matching-engine-app/src/main/kotlin/co/nilin/opex/matching/engine/app/bl/OrderBooks.kt +++ b/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() 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) diff --git a/matching-engine/matching-engine-core/src/main/kotlin/co/nilin/opex/matching/engine/core/eventh/EventDispatcher.kt b/matching-engine/matching-engine-core/src/main/kotlin/co/nilin/opex/matching/engine/core/eventh/EventDispatcher.kt index 7d6c4d4f4..cc2d31cfc 100644 --- a/matching-engine/matching-engine-core/src/main/kotlin/co/nilin/opex/matching/engine/core/eventh/EventDispatcher.kt +++ b/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( val lambda: (T) -> Unit ) { diff --git a/matching-engine/matching-engine-core/src/main/kotlin/co/nilin/opex/matching/engine/core/inout/OrderBookUpdateEvent.kt b/matching-engine/matching-engine-core/src/main/kotlin/co/nilin/opex/matching/engine/core/inout/OrderBookUpdateEvent.kt new file mode 100644 index 000000000..8e615433f --- /dev/null +++ b/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 +) \ No newline at end of file diff --git a/matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/config/EventsKafkaConfig.kt b/matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/config/EventsKafkaConfig.kt index c11c225e8..667fceeac 100644 --- a/matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/config/EventsKafkaConfig.kt +++ b/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): KafkaTemplate { + return KafkaTemplate(DefaultKafkaProducerFactory(producerConfigs)) + } + } \ No newline at end of file diff --git a/matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/config/KafkaTopicConfig.kt b/matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/config/KafkaTopicConfig.kt index ee8446e29..a11f308fc 100644 --- a/matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/config/KafkaTopicConfig.kt +++ b/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() + }) } } diff --git a/matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/service/OrderBookUpdateSubmitter.kt b/matching-engine/matching-engine-ports/matching-engine-submitter-kafka/src/main/kotlin/co/nilin/opex/matching/engine/ports/kafka/submitter/service/OrderBookUpdateSubmitter.kt new file mode 100644 index 000000000..e617ddb2b --- /dev/null +++ b/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) { + + 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) + } + ) + } +} \ No newline at end of file