Compare commits
9 Commits
2bfc28555c
...
master
Author | SHA1 | Date | |
---|---|---|---|
f6e664f0f7 | |||
4b57850635 | |||
7b0281af10
|
|||
0809386bc3
|
|||
2eb23dd9a6
|
|||
788b636196 | |||
79b68af64c
|
|||
a719addee4
|
|||
df13c145e7
|
12
.drone.yml
12
.drone.yml
@ -30,6 +30,18 @@ steps:
|
||||
- latest
|
||||
- ${DRONE_BUILD_NUMBER}
|
||||
|
||||
- name: dockerhub
|
||||
image: plugins/docker
|
||||
settings:
|
||||
repo: birbnetes/${DRONE_REPO_NAME}
|
||||
username:
|
||||
from_secret: DOCKERHUB_USER
|
||||
password:
|
||||
from_secret: DOCKERHUB_PASSWORD
|
||||
tags:
|
||||
- latest
|
||||
- ${DRONE_BUILD_NUMBER}
|
||||
|
||||
- name: ms-teams
|
||||
image: kuperiu/drone-teams
|
||||
settings:
|
||||
|
@ -11,4 +11,4 @@ USER $APPLICATION_USER
|
||||
COPY ./build/libs/output-service-rdb-consumer.jar /app/output-service-rdb-consumer.jar
|
||||
WORKDIR /app
|
||||
|
||||
CMD ["java", "-server", "-XX:+UnlockExperimentalVMOptions", "-XX:InitialRAMFraction=2", "-XX:MinRAMFraction=2", "-XX:MaxRAMFraction=2", "-XX:+UseG1GC", "-XX:MaxGCPauseMillis=100", "-XX:+UseStringDeduplication", "-jar", "output-service-rdb-consumer.jar"]
|
||||
CMD ["java", "-server", "-XX:+UnlockExperimentalVMOptions", "-XX:+UseG1GC", "-XX:MaxGCPauseMillis=100", "-XX:+UseStringDeduplication", "-jar", "output-service-rdb-consumer.jar"]
|
@ -1,9 +1,20 @@
|
||||
plugins {
|
||||
application
|
||||
kotlin("jvm") version "1.5.10"
|
||||
kotlin("plugin.serialization") version "1.5.10"
|
||||
id("com.github.johnrengelman.shadow") version "7.0.0"
|
||||
}
|
||||
|
||||
buildscript {
|
||||
repositories { mavenCentral() }
|
||||
|
||||
dependencies {
|
||||
val kotlinVersion = "1.5.10"
|
||||
classpath(kotlin("gradle-plugin", version = kotlinVersion))
|
||||
classpath(kotlin("serialization", version = kotlinVersion))
|
||||
}
|
||||
}
|
||||
|
||||
application {
|
||||
mainClassName = "MainKt"
|
||||
}
|
||||
@ -20,13 +31,14 @@ dependencies {
|
||||
implementation("com.rabbitmq:amqp-client:5.13.0")
|
||||
implementation("com.viartemev:the-white-rabbit:0.0.6")
|
||||
implementation("com.zaxxer:HikariCP:5.0.0")
|
||||
implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.2.2")
|
||||
implementation("org.jetbrains.kotlinx:kotlinx-datetime:0.2.1")
|
||||
implementation("org.jetbrains.exposed:exposed-core:0.33.1")
|
||||
implementation("org.jetbrains.exposed:exposed-dao:0.33.1")
|
||||
implementation("org.jetbrains.exposed:exposed-jdbc:0.33.1")
|
||||
implementation("org.jetbrains.exposed:exposed-java-time:0.33.1")
|
||||
implementation("io.insert-koin:koin-core:3.1.2")
|
||||
implementation("org.postgresql:postgresql:42.2.23")
|
||||
implementation("com.google.code.gson:gson:2.8.7")
|
||||
}
|
||||
|
||||
tasks {
|
||||
|
@ -1,5 +1,9 @@
|
||||
import database.DatabaseFactory
|
||||
import database.dao.ResultObjects
|
||||
import di.databasemodule
|
||||
import mq.ConsumerWrapper
|
||||
import org.jetbrains.exposed.sql.SchemaUtils
|
||||
import org.jetbrains.exposed.sql.transactions.transaction
|
||||
import org.koin.core.context.GlobalContext.startKoin
|
||||
|
||||
fun main(vararg args: String) {
|
||||
@ -9,6 +13,11 @@ fun main(vararg args: String) {
|
||||
modules(databasemodule)
|
||||
}
|
||||
|
||||
DatabaseFactory.init()
|
||||
transaction{
|
||||
SchemaUtils.create(ResultObjects)
|
||||
}
|
||||
|
||||
val consumerWrapper = ConsumerWrapper();
|
||||
consumerWrapper.recieve();
|
||||
}
|
@ -1,8 +1,9 @@
|
||||
package api
|
||||
|
||||
import com.google.gson.annotations.SerializedName
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
@Serializable
|
||||
data class ApiObject(
|
||||
@SerializedName("tag") val tag: String,
|
||||
@SerializedName("probability") val probability: Double
|
||||
val tag: String,
|
||||
val probability: Double
|
||||
)
|
@ -7,5 +7,5 @@ data class EnvConfig (
|
||||
var mqExchange: String = System.getenv("MQ_EXCHANGE") ?: "rabbitmq",
|
||||
var dbJdbc: String = System.getenv("DB_JDBC") ?: "input",
|
||||
var dbUsername: String = System.getenv("DB_USERNAME") ?: "output",
|
||||
var dbPassowrd: String = System.getenv("DB_PASSOWRD") ?: "output"
|
||||
var dbPassowrd: String = System.getenv("DB_PASSWORD") ?: "output"
|
||||
)
|
@ -2,12 +2,12 @@ package database.dao
|
||||
|
||||
import org.jetbrains.exposed.dao.id.IntIdTable
|
||||
import org.jetbrains.exposed.sql.*
|
||||
import org.jetbrains.exposed.sql.`java-time`.date
|
||||
import java.time.LocalDate
|
||||
import org.jetbrains.exposed.sql.`java-time`.datetime
|
||||
import java.time.LocalDateTime
|
||||
|
||||
object ResultObjects : IntIdTable() {
|
||||
val tag: Column<String> = varchar("tag", 32)
|
||||
val date: Column<LocalDate> = date("date").default(LocalDate.now())
|
||||
val date: Column<LocalDateTime> = datetime("date").default(LocalDateTime.now())
|
||||
val probability: Column<Double> = double("probability")
|
||||
override val primaryKey = PrimaryKey(id, name = "PK_ResultObject_Id")
|
||||
}
|
@ -1,14 +1,8 @@
|
||||
package database.service
|
||||
|
||||
import database.DatabaseFactory.dbQuery
|
||||
import database.model.ResultObject
|
||||
import database.dao.ResultObjects
|
||||
import api.ApiObject
|
||||
import org.jetbrains.exposed.sql.ResultRow
|
||||
import org.jetbrains.exposed.sql.select
|
||||
import org.jetbrains.exposed.sql.selectAll
|
||||
import org.jetbrains.exposed.sql.transactions.transaction
|
||||
import java.time.LocalDate
|
||||
|
||||
class ResultObjectService : IResultObjectService {
|
||||
|
||||
|
@ -20,6 +20,6 @@ class ConsumerWrapper {
|
||||
val inputQueueName = inputChannel.queueDeclare().queue
|
||||
inputChannel.queueBind(inputQueueName, envConfig.mqExchange, "")
|
||||
|
||||
inputChannel.basicConsume(inputQueueName, true, DatabaseConsumer())
|
||||
inputChannel.basicConsume(inputQueueName, false, DatabaseConsumer(inputChannel))
|
||||
}
|
||||
}
|
@ -1,18 +1,16 @@
|
||||
package mq
|
||||
|
||||
import api.ApiObject
|
||||
import com.google.gson.Gson
|
||||
import com.rabbitmq.client.AMQP
|
||||
import com.rabbitmq.client.Consumer
|
||||
import com.rabbitmq.client.Envelope
|
||||
import com.rabbitmq.client.ShutdownSignalException
|
||||
import com.rabbitmq.client.*
|
||||
import database.service.IResultObjectService
|
||||
import kotlinx.serialization.decodeFromString
|
||||
import kotlinx.serialization.json.Json
|
||||
import org.koin.core.component.KoinComponent
|
||||
import org.koin.core.component.inject
|
||||
|
||||
class DatabaseConsumer: Consumer, KoinComponent {
|
||||
class DatabaseConsumer(channel: Channel): Consumer, KoinComponent {
|
||||
private val resultObjectService : IResultObjectService by inject()
|
||||
private val gson = Gson()
|
||||
private val basicChannel = channel
|
||||
override fun handleConsumeOk(consumerTag : String?) {
|
||||
}
|
||||
override fun handleCancelOk(p0 : String?) {
|
||||
@ -30,8 +28,14 @@ class DatabaseConsumer: Consumer, KoinComponent {
|
||||
}
|
||||
|
||||
override fun handleDelivery(consumerTag : String?, envelope : Envelope?, basicProperties : AMQP.BasicProperties?, body : ByteArray?) {
|
||||
val rawJson = body!!.toString(Charsets.UTF_8)
|
||||
val apiObject = gson.fromJson(rawJson, ApiObject::class.java)
|
||||
resultObjectService.addOne(apiObject)
|
||||
try {
|
||||
val rawJson = body!!.toString(Charsets.UTF_8)
|
||||
val apiObject = Json{ ignoreUnknownKeys = true }.decodeFromString<ApiObject>(rawJson)
|
||||
resultObjectService.addOne(apiObject)
|
||||
basicChannel.basicAck(envelope!!.deliveryTag, false)
|
||||
} catch (e: Exception) {
|
||||
println(e.message)
|
||||
basicChannel.basicNack(envelope!!.deliveryTag, false, true)
|
||||
}
|
||||
}
|
||||
}
|
Reference in New Issue
Block a user