birbnetes/output-service-rdb
4
0
Fork 0
Code Pull Requests Releases Wiki Activity

add db interaction layer
Some checks failed
continuous-integration/drone/push Build is failing
Details

Browse Source
This commit is contained in:
Torma Kristóf
2020-04-04 23:21:17 +02:00
parent 97b5923c1e
commit af4307c6a4
11 changed files with 158 additions and 425 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
build.gradle
View File

@@ -32,6 +32,13 @@ repositories {
}
dependencies {
compile 'org.postgresql:postgresql:42.2.2'
compile 'org.jetbrains.exposed:exposed-core:0.23.1'
compile 'org.jetbrains.exposed:exposed-dao:0.23.1'
compile 'org.jetbrains.exposed:exposed-jdbc:0.23.1'
compile 'org.jetbrains.exposed:exposed-java-time:0.23.1'
compile 'com.rabbitmq:amqp-client:2.7.1'
compile 'com.zaxxer:HikariCP:2.7.8'
implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk8:$kotlin_version"
implementation "io.ktor:ktor-server-netty:$ktor_version"
implementation "ch.qos.logback:logback-classic:$logback_version"
@@ -41,7 +48,6 @@ dependencies {
implementation "io.ktor:ktor-client-core:$ktor_version"
implementation "io.ktor:ktor-client-core-jvm:$ktor_version"
implementation "io.ktor:ktor-client-apache:$ktor_version"
implementation "io.ktor:ktor-jackson:$ktor_version"
implementation "io.ktor:ktor-auth:$ktor_version"
testImplementation "io.ktor:ktor-server-tests:$ktor_version"
}

7
resources/application.conf
View File

@@ -6,4 +6,9 @@ ktor {
application {
modules = [ com.kmalbz.ApplicationKt.module ]
}
}
db {
jdbcUrl = ${DB_URL} //jdbc:postgresql://localhost:${db_port}/${db_name}
dbUser = ${DB_USER}
dbPassword = ${DB_PASSWORD}
}
}

17
src/Application.kt
View File

@@ -9,12 +9,10 @@ import io.ktor.gson.*
import io.ktor.features.*
import io.ktor.client.*
import io.ktor.client.engine.apache.*
import com.fasterxml.jackson.databind.*
import io.ktor.jackson.*
import io.ktor.auth.*
import kotlin.reflect.*
import java.util.*
import io.ktor.swagger.experimental.*
import org.apache.http.HttpException
fun main(args: Array<String>): Unit = io.ktor.server.netty.EngineMain.main(args)
@@ -24,10 +22,6 @@ fun Application.module(testing: Boolean = false) {
install(ContentNegotiation) {
gson {
}
jackson {
enable(SerializationFeature.INDENT_OUTPUT)
}
}
val client = HttpClient(Apache) {
@@ -49,19 +43,16 @@ fun Application.module(testing: Boolean = false) {
exception<AuthenticationException> { cause ->
call.respond(HttpStatusCode.Unauthorized)
}
exception<AuthorizationException> { cause ->
call.respond(HttpStatusCode.Forbidden)
}
exception<HttpException> { cause ->
call.respond(cause.code, cause.description)
exception<HttpException> { cause ->
call.respond(HttpStatusCode.BadRequest)
}
}
get("/json/jackson") {
call.respond(mapOf("hello" to "world"))
}
OutputServiceRDBServer().apply {
registerOutput()
}

6
src/OutputServiceRDB.kt
View File

@@ -1,14 +1,12 @@
package com.kmalbz
import java.util.*
import io.ktor.http.*
import io.ktor.request.*
import io.ktor.swagger.experimental.*
data class OutputObject(
val tag: String,
val decison: Boolean,
val date: Date
val date: Date,
val confidence: Double
)
data class ApiResponse(

1
src/OutputServiceRDBClient.kt
View File

@@ -2,6 +2,7 @@ package com.kmalbz
import io.ktor.client.*
import io.ktor.client.request.*
import java.util.*
/**
* Output Service - RDB Client

21
src/OutputServiceRDBServer.kt
View File

@@ -4,8 +4,6 @@ import io.ktor.application.*
import io.ktor.response.*
import io.ktor.routing.*
import java.util.*
import io.ktor.swagger.experimental.*
import io.ktor.auth.*
import io.ktor.http.*
/**
@@ -19,42 +17,43 @@ class OutputServiceRDBServer() {
*/
fun Routing.registerOutput() {
get("/output/filter/negative") {
if (false) httpException(HttpStatusCode.NotFound)
if (false) error(HttpStatusCode.NotFound)
call.respond(listOf())
}
get("/output/filter/positive") {
if (false) httpException(HttpStatusCode.NotFound)
if (false) error(HttpStatusCode.NotFound)
call.respond(listOf())
}
get("/output/after/{dateAfter}") {
val dateAfter = call.getPath<Date>("dateAfter")
val dateAfter = call.parameters["dateAfter"]
if (false) httpException(HttpStatusCode.NotFound)
if (false) error(HttpStatusCode.NotFound)
call.respond(listOf())
}
get("/output/before/{dateBefore}") {
val dateBefore = call.getPath<Date>("dateBefore")
val dateBefore = call.parameters["dateBefore"]
if (false) httpException(HttpStatusCode.NotFound)
if (false) error(HttpStatusCode.NotFound)
call.respond(listOf())
}
get("/output/{tagID}") {
val tagID = call.getPath<Int>("tagID")
val tagID = call.parameters["tagID"]
if (false) httpException(HttpStatusCode.NotFound)
if (false) error(HttpStatusCode.NotFound)
call.respond(OutputObject(
tag = "tag",
decison = false,
date = Date()
date = Date(),
confidence = 0.0
))
}
}

46
src/database/DatabaseFactory.kt Normal file
View File

@@ -0,0 +1,46 @@
package com.kmalbz.database
import com.typesafe.config.ConfigFactory
import com.zaxxer.hikari.*
import io.ktor.config.HoconApplicationConfig
import io.ktor.util.KtorExperimentalAPI
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import org.jetbrains.exposed.sql.*
import org.jetbrains.exposed.sql.transactions.transaction
object DatabaseFactory {
@KtorExperimentalAPI
private val appConfig = HoconApplicationConfig(ConfigFactory.load())
@KtorExperimentalAPI
private val dbUrl = appConfig.property("db.jdbcUrl").getString()
@KtorExperimentalAPI
private val dbUser = appConfig.property("db.dbUser").getString()
@KtorExperimentalAPI
private val dbPassword = appConfig.property("db.dbPassword").getString()
@KtorExperimentalAPI
fun init() {
Database.connect(hikari())
}
@KtorExperimentalAPI
private fun hikari(): HikariDataSource {
val config = HikariConfig()
config.driverClassName = "org.postgresql.Driver"
config.jdbcUrl = dbUrl
config.username = dbUser
config.password = dbPassword
config.maximumPoolSize = 3
config.isAutoCommit = false
config.transactionIsolation = "TRANSACTION_REPEATABLE_READ"
config.validate()
return HikariDataSource(config)
}
suspend fun <T> dbQuery(block: () -> T): T =
withContext(Dispatchers.IO) {
transaction { block() }
}
}

14
src/database/dao/ResultObjects.kt Normal file
View File

@@ -0,0 +1,14 @@
package com.kmalbz.database.dao
import org.jetbrains.exposed.sql.*
import org.jetbrains.exposed.sql.`java-time`.date
import java.time.LocalDate
object ResultObjects : Table() {
val id: Column<Int> = integer("id").autoIncrement()
val tag: Column<String> = varchar("tag",32)
val date: Column<LocalDate> = date("date")
val decision: Column<Boolean> = bool("decision")
val confidence: Column<Double> = double("confidence")
override val primaryKey = PrimaryKey(id, name = "PK_ResultObject_Id")
}

11
src/database/model/ResultObject.kt Normal file
View File

@@ -0,0 +1,11 @@
package com.kmalbz.database.model
import java.time.LocalDate
data class ResultObject(
val id: Int,
val tag: String,
val date: LocalDate,
val decision: Boolean,
val confidence: Double
)

57
src/database/service/ResultObjectService.kt Normal file
View File

@@ -0,0 +1,57 @@
package com.kmalbz.database.service
import com.kmalbz.database.DatabaseFactory.dbQuery
import com.kmalbz.database.model.ResultObject
import com.kmalbz.database.dao.ResultObjects
import org.jetbrains.exposed.sql.ResultRow
import org.jetbrains.exposed.sql.select
import org.jetbrains.exposed.sql.selectAll
import java.time.LocalDate
class ResultObjectService {
suspend fun getAllResultObjects(): List<ResultObject> = dbQuery {
ResultObjects.selectAll().map { toResultObject(it) }
}
suspend fun getResultObjectbyTag(tag: String): ResultObject? = dbQuery {
ResultObjects.select {
(ResultObjects.tag eq tag)
}.mapNotNull { toResultObject(it) }
.singleOrNull()
}
suspend fun getResultObjectbyDate(date: LocalDate): List<ResultObject>? = dbQuery {
ResultObjects.select {
(ResultObjects.date eq date)
}.mapNotNull { toResultObject(it) }
}
suspend fun getResultObjectbeforeDate(date: LocalDate): List<ResultObject>? = dbQuery {
ResultObjects.select {
(ResultObjects.date less date)
}.mapNotNull { toResultObject(it) }
}
suspend fun getResultObjectafterDate(date: LocalDate): List<ResultObject>? = dbQuery {
ResultObjects.select {
(ResultObjects.date greater date)
}.mapNotNull { toResultObject(it) }
}
suspend fun getResultObjectbyDecision(decision: Boolean): List<ResultObject>? = dbQuery {
ResultObjects.select {
(ResultObjects.decision eq decision)
}.mapNotNull { toResultObject(it) }
}
private fun toResultObject(row: ResultRow): ResultObject =
ResultObject(
id = row[ResultObjects.id],
tag = row[ResultObjects.tag],
date = row[ResultObjects.date],
decision = row[ResultObjects.decision],
confidence = row[ResultObjects.confidence]
)
}

395
src/io/ktor/swagger/experimental/SwaggerUtils.kt
View File

@@ -1,395 +0,0 @@
package io.ktor.swagger.experimental
import com.fasterxml.jackson.module.kotlin.*
import kotlin.coroutines.*
import kotlin.coroutines.intrinsics.*
import kotlinx.coroutines.*
import io.ktor.application.*
import io.ktor.auth.authenticate
import io.ktor.client.*
import io.ktor.client.call.*
import io.ktor.client.request.*
import io.ktor.client.response.*
import io.ktor.content.*
import io.ktor.http.*
import io.ktor.request.*
import io.ktor.response.*
import io.ktor.routing.*
import io.ktor.util.*
import java.lang.reflect.*
import java.lang.reflect.Type
class HttpException(val code: HttpStatusCode, val description: String = code.description) : RuntimeException(description)
fun httpException(code: HttpStatusCode, message: String = code.description): Nothing = throw HttpException(code, message)
fun httpException(code: Int, message: String = "Error $code"): Nothing = throw HttpException(HttpStatusCode(code, message))
@Suppress("unused")
inline fun <T> T.verifyParam(name: String, callback: (T) -> Boolean): T {
if (!callback(this)) throw IllegalArgumentException("$name"); return this
}
inline fun <T> T.checkRequest(cond: Boolean, callback: () -> String) {
if (!cond) httpException(HttpStatusCode.BadRequest, callback())
}
interface SwaggerBaseApi {
}
interface SwaggerBaseServer {
}
class ApplicationCallContext(val call: ApplicationCall) : CoroutineContext.Element {
object KEY : CoroutineContext.Key<ApplicationCallContext>
override val key: CoroutineContext.Key<*> = KEY
}
@Suppress("unused")
suspend fun SwaggerBaseServer.call(): ApplicationCall {
return coroutineContext[ApplicationCallContext.KEY]?.call ?: error("ApplicationCall not available")
}
annotation class Method(val method: String)
annotation class Body(val name: String)
annotation class Header(val name: String)
annotation class Query(val name: String)
annotation class Path(val name: String) // Reused
annotation class FormData(val name: String)
annotation class Auth(vararg val auths: String)
//interface FeatureClass
//annotation class Feature(val clazz: KClass<out FeatureClass>)
inline fun <reified T : SwaggerBaseApi> createClient(client: HttpClient, rootUrl: String): T =
createClient(T::class.java, client, rootUrl)
fun <T : SwaggerBaseApi> createClient(clazz: Class<T>, client: HttpClient, rootUrl: String): T {
val rootUrlTrim = rootUrl.trimEnd('/')
val apiClass = ApiClass.parse(clazz)
var authContext = LinkedHashMap<String, String>()
return Proxy.newProxyInstance(clazz.classLoader, arrayOf(clazz)) { proxy, method, args ->
val info = apiClass.getInfo(method) ?: error("Can't find method $method")
val rparams = info.params.zip(args.slice(0 until info.params.size)).map { ApiClass.ApiParamInfoValue<Any?>(it.first as ApiClass.ApiParamInfo<Any?>, it.second) }.associateBy { it.name }
//val params = method.parameters
val cont = args.lastOrNull() as? Continuation<Any>?
?: throw RuntimeException("Just implemented suspend functions")
val continuationReturnType = method.genericParameterTypes.firstOrNull()?.extractFirstGenericType()
val realReturnType = continuationReturnType ?: method.returnType
val pathPattern = info.path
val pathReplaced = pathPattern.replace { "${rparams[it]?.value}" }
kotlinx.coroutines.GlobalScope.apply {
launch {
try {
val fullUrl = "$rootUrlTrim/$pathReplaced"
val res = client.call(fullUrl) {
this.method = HttpMethod(info.httpMethod)
val body = linkedMapOf<String, Any?>()
val formData = linkedMapOf<String, Any?>()
for (param in rparams.values) {
when (param.source) {
Source.QUERY -> parameter(param.name, "${param.value}")
Source.HEADER -> header(param.name, "${param.value}")
Source.BODY -> body[param.name] = param.value
Source.FORM_DATA -> formData[param.name] = param.value
}
}
if (body.isNotEmpty()) {
this.contentType(io.ktor.http.ContentType.Application.Json)
this.body = ByteArrayContent(Json.stringify(body).toByteArray(Charsets.UTF_8))
}
if (formData.isNotEmpty()) {
this.contentType(io.ktor.http.ContentType.Application.FormUrlEncoded)
this.body = ByteArrayContent(formData.map { it.key to it.value.toString() }.formUrlEncode().toByteArray(Charsets.UTF_8))
}
}
if (res.response.status.value < 400) {
cont.resume(Json.parse(res.response.readText(), realReturnType))
} else {
throw HttpExceptionWithContent(res.response.status, res.response.readText())
}
} catch (e: Throwable) {
cont.resumeWithException(e)
}
}
}
COROUTINE_SUSPENDED
} as T
}
class HttpExceptionWithContent(val code: HttpStatusCode, val content: String) :
RuntimeException("HTTP ERROR $code : $content")
fun Routing.registerRoutes(server: SwaggerBaseServer) {
val clazz = ApiClass.parse(server::class.java)
for (method in clazz.methods) {
authenticateIfNotEmpty(method.auths) {
route(method.path.pathPattern, HttpMethod(method.httpMethod)) {
handle {
val args = arrayListOf<Any?>()
for (param in method.params) {
args += param.get(call)
}
withContext(ApplicationCallContext(call)) {
val result = method.method.invokeSuspend(server, args)
call.respondText(Json.stringify(result ?: Any()), ContentType.Application.Json)
}
}
}
}
}
}
enum class Source {
BODY, QUERY, FORM_DATA, HEADER, PATH
}
fun Route.authenticateIfNotEmpty(configurations: List<String>, optional: Boolean = false, build: Route.() -> Unit): Route {
return if (configurations.isEmpty()) {
build()
this
} else {
authenticate(*configurations.toTypedArray(), optional = optional, build = build)
}
}
class ApiClass(val clazz: Class<*>, val methods: List<ApiMethodInfo>) {
val methodsBySignature = methods.associateBy { it.methodSignature }
fun getInfo(method: java.lang.reflect.Method) = methodsBySignature[method.signature]
companion object {
fun parse(clazz: Class<*>): ApiClass {
val imethods = arrayListOf<ApiMethodInfo>()
for (method in clazz.methods) {
val path = method.getAnnotationInAncestors(Path::class.java)?.name
val httpMethod = method.getAnnotationInAncestors(Method::class.java)?.method ?: "GET"
//println("METHOD: $method, $path")
if (path != null) {
val params = arrayListOf<ApiParamInfo<*>>()
for ((ptype, annotations) in method.parameterTypes.zip(method.parameterAnnotationsInAncestors)) {
// Skip the continuation last argument!
if (ptype.isAssignableFrom(Continuation::class.java)) continue
val body = annotations.filterIsInstance<Body>().firstOrNull()?.name
val query = annotations.filterIsInstance<Query>().firstOrNull()?.name
val formData = annotations.filterIsInstance<FormData>().firstOrNull()?.name
val header = annotations.filterIsInstance<Header>().firstOrNull()?.name
val ppath = annotations.filterIsInstance<Path>().firstOrNull()?.name
val source = when {
body != null -> Source.BODY
query != null -> Source.QUERY
formData != null -> Source.FORM_DATA
header != null -> Source.HEADER
ppath != null -> Source.PATH
else -> Source.QUERY
}
val rname = body ?: query ?: formData ?: header ?: ppath ?: "unknown"
//println(" - $ptype, ${annotations.toList()}")
params += ApiParamInfo(source, rname, ptype)
}
//println("METHOD: $instance, $method, $httpMethod, $path")
//for (param in params) println(" - $param")
val auths = method.getAnnotationInAncestors(Auth::class.java)?.auths?.toList() ?: listOf()
imethods += ApiMethodInfo(method, PathPattern(path.trim('/')), httpMethod, auths, params)
}
}
return ApiClass(clazz, imethods)
}
}
class ApiMethodInfo(val method: java.lang.reflect.Method, val path: PathPattern, val httpMethod: String, val auths: List<String>, val params: List<ApiParamInfo<*>>) {
val methodSignature = method.signature
}
data class ApiParamInfo<T>(val source: Source, val name: String, val type: Class<T>) {
suspend fun get(call: ApplicationCall): T {
return call.getTyped(source, name, type)
}
}
data class ApiParamInfoValue<T>(val info: ApiParamInfo<T>, val value: T) {
val source get() = info.source
val type get() = info.type
val name get() = info.name
}
class PathPattern(val pathPattern: String) {
companion object {
val PARAM_REGEX = Regex("\\{(\\w*)\\}")
}
val pathNames by lazy { PARAM_REGEX.findAll(pathPattern).map { it.groupValues[1] }.toList() }
val pathRegex by lazy { Regex(replace { "(\\w+)" }) }
fun replace(replacer: (name: String) -> String): String {
return pathPattern.replace(PARAM_REGEX) { mr -> replacer(mr.groupValues[1]) }
}
fun extract(path: String): List<String> {
return pathRegex.find(path)?.groupValues?.drop(1) ?: listOf()
}
}
}
data class MethodSignature(val name: String, val types: List<Class<*>>)
val java.lang.reflect.Method.signature get() = MethodSignature(name, parameterTypes.toList())
object Json {
@PublishedApi
internal val objectMapper = jacksonObjectMapper()
fun <T> convert(value: Any?, clazz: Class<T>): T = objectMapper.convertValue(value, clazz)
fun <T> parse(str: String, clazz: Class<T>): T = objectMapper.readValue(str, clazz)
fun <T> stringify(value: T): String = objectMapper.writeValueAsString(value)
}
//inline fun <reified T> ApplicationCall.getTyped(source: String, name: String): T =
// objectMapper.convertValue(getRaw(source, name), T::class.java)
suspend fun <T> ApplicationCall.getTyped(source: Source, name: String, clazz: Class<T>): T {
return Json.convert(getRaw(source, name), clazz)
}
suspend fun <T> ApplicationCall.getTypedOrNull(source: Source, name: String, clazz: Class<T>): T? =
getRaw(source, name)?.let { Json.convert(it, clazz) }
suspend fun ApplicationCall.getRaw(source: Source, name: String): Any? {
return when (source) {
Source.PATH -> this.parameters.get(name)
Source.QUERY -> this.request.queryParameters.get(name)
Source.BODY -> this.getCachedUntypedBody()[name]
Source.FORM_DATA -> TODO()
Source.HEADER -> this.request.header(name)
}
}
val CACHED_BODY_KEY = AttributeKey<Any>("CACHED_BODY_KEY")
inline fun <T : Any> Attributes.computeIfAbsentInline(key: AttributeKey<T>, block: () -> T): T {
if (!this.contains(key)) {
this.put(key, block())
}
return this[key]
}
private suspend fun ApplicationCall.getCachedUntypedBody(): Map<String, Any?> {
return attributes.computeIfAbsentInline(CACHED_BODY_KEY) {
Json.parse(this@getCachedUntypedBody.receive(), HashMap::class.java)
} as Map<String, Any?>
}
///////////////////////////////////////////////////
// Reflection Tools
///////////////////////////////////////////////////
val java.lang.Class<*>.allTypes: Set<Class<*>>
get() {
val types = LinkedHashSet<Class<*>>()
val explore = arrayListOf(this)
while (explore.isNotEmpty()) {
val item = explore.removeAt(explore.size - 1) ?: continue
types += item
explore += item.superclass
explore += item.interfaces
}
return types
}
val java.lang.reflect.Method.parameterAnnotationsInAncestors: List<List<Annotation>>
get() {
val allMethods = this.declaringClass.allTypes.map {
try {
it.getDeclaredMethod(name, *parameterTypes) ?: null
} catch (e: NoSuchMethodException) {
null
}
}.filterNotNull()
val out = Array<ArrayList<Annotation>>(parameterTypes.size) { arrayListOf() }.toList()
for (method in allMethods) {
for ((index, annotations) in method.parameterAnnotations.withIndex()) {
out[index] += annotations
}
}
return out
}
suspend fun java.lang.reflect.Method.invokeSuspend(obj: Any?, args: List<Any?>): Any? = suspendCoroutine { c ->
val method = this@invokeSuspend
val lastParam = method.parameterTypes.lastOrNull()
val margs = java.util.ArrayList(args)
if (lastParam != null && lastParam.isAssignableFrom(Continuation::class.java)) {
margs += c
}
try {
val result = method.invoke(obj, *margs.toTypedArray())
if (result != COROUTINE_SUSPENDED) {
c.resume(result)
}
} catch (e: InvocationTargetException) {
c.resumeWithException(e.targetException)
} catch (e: Throwable) {
c.resumeWithException(e)
}
}
fun <T : Annotation> java.lang.reflect.Method.getAnnotationInAncestors(clazz: Class<T>): T? {
val res = this.getAnnotation(clazz) ?: this.getDeclaredAnnotation(clazz)
if (res != null) return res
// Try interfaces
for (ifc in this.declaringClass.interfaces) {
return ignoreErrors { ifc?.getMethod(name, *parameterTypes)?.getAnnotationInAncestors(clazz) } ?: continue
}
// Try ancestor
return ignoreErrors { this.declaringClass.superclass?.getMethod(name, *parameterTypes) }?.getAnnotationInAncestors(
clazz
)
}
inline fun <T> ignoreErrors(callback: () -> T): T? = try {
callback()
} catch (e: Throwable) {
null
}
fun Type.extractFirstGenericType(): Class<*> {
if (this is ParameterizedType) {
return this.actualTypeArguments.first().extractFirstGenericType()
}
if (this is WildcardType) {
val tt = this.lowerBounds.firstOrNull() ?: this.upperBounds.firstOrNull()
?: error("WildcardType without lower/upper bounds")
return tt.extractFirstGenericType()
}
if (this is Class<*>) {
return this
}
error("Couldn't find right generic type")
}
suspend inline fun <reified T : Any> ApplicationCall.getBodyParam(name: String, noinline default: () -> T = { error("mandatory $name") }): T =
getTypedOrNull(Source.BODY, name, T::class.java) ?: default()
suspend inline fun <reified T : Any> ApplicationCall.getPath(name: String, noinline default: () -> T = { error("mandatory $name") }): T =
getTypedOrNull(Source.PATH, name, T::class.java) ?: default()
suspend inline fun <reified T : Any> ApplicationCall.getQuery(name: String, noinline default: () -> T = { error("mandatory $name") }): T =
getTypedOrNull(Source.QUERY, name, T::class.java) ?: default()