output-service-rdb/src/Application.kt

97 lines
2.9 KiB
Kotlin

package com.kmalbz
import com.google.gson.Gson
import com.kmalbz.api.model.ApiObject
import com.kmalbz.api.route.OutputServiceRDBServer
import io.ktor.application.*
import io.ktor.response.*
import io.ktor.routing.*
import io.ktor.http.*
import io.ktor.gson.*
import io.ktor.features.*
import org.apache.http.HttpException
import com.kmalbz.database.DatabaseFactory
import com.kmalbz.database.dao.ResultObjects
import com.kmalbz.database.service.ResultObjectService
import io.ktor.util.KtorExperimentalAPI
import com.rabbitmq.client.*
import com.typesafe.config.ConfigFactory
import io.ktor.config.HoconApplicationConfig
import com.viartemev.thewhiterabbit.channel.*
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.launch
import org.jetbrains.exposed.sql.SchemaUtils
import org.jetbrains.exposed.sql.transactions.transaction
import org.koin.ktor.ext.Koin
fun main(args: Array<String>): Unit = io.ktor.server.netty.EngineMain.main(args)
@KtorExperimentalAPI
@Suppress("unused") // Referenced in application.conf
fun Application.module() {
install(ContentNegotiation) {
gson {
}
}
install(Koin) {
printLogger()
modules(com.kmalbz.di.injectionModule)
}
DatabaseFactory.init()
transaction{
SchemaUtils.create(ResultObjects)
}
val appConfig = HoconApplicationConfig(ConfigFactory.load())
val factory = ConnectionFactory()
factory.host = appConfig.property("ktor.mq.host").getString()
factory.username = appConfig.property("ktor.mq.username").getString()
factory.password = appConfig.property("ktor.mq.password").getString()
val connection = factory.newConnection()
val channel = connection.createChannel()
val rabbitExchangeName = appConfig.property("ktor.mq.exchange").getString()
channel.exchangeDeclare(rabbitExchangeName, BuiltinExchangeType.FANOUT)
val queueName = channel.queueDeclare().queue
channel.queueBind(queueName, rabbitExchangeName, "")
GlobalScope.launch(Dispatchers.Default) {
connection.channel {
consume(queueName) {
consumeMessageWithConfirm({
val resultObjectService = ResultObjectService()
val rawJson = String(it.body)
val gson = Gson()
val apiObject = gson.fromJson(rawJson,ApiObject::class.java)
resultObjectService.addOne(apiObject)
})
}
}
}
routing {
install(StatusPages) {
exception<HttpException> {
call.respond(HttpStatusCode.BadRequest)
}
exception<IllegalStateException> {
call.respond(HttpStatusCode.NotAcceptable)
}
}
OutputServiceRDBServer().apply {
registerOutput()
}
}
}
class AuthenticationException : RuntimeException()
class AuthorizationException : RuntimeException()