upload to docker hub

.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:

Dockerfile

@@ -11,4 +11,4 @@ USER $APPLICATION_USER
 COPY ./build/libs/sample-service-consumer.jar /app/sample-service-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", "sample-service-consumer.jar"]
+CMD ["java", "-server", "-XX:+UnlockExperimentalVMOptions", "-XX:+UseG1GC", "-XX:MaxGCPauseMillis=100", "-XX:+UseStringDeduplication", "-jar", "sample-service-consumer.jar"]

build.gradle.kts

@@ -1,6 +1,7 @@
 plugins {
     application
     kotlin("jvm") version "1.5.10"
+    kotlin("plugin.serialization") version "1.5.10"
     id("com.github.johnrengelman.shadow") version "7.0.0"
 }
 
@@ -16,7 +17,19 @@ repositories {
     jcenter()
 }
 
+buildscript {
+    repositories { mavenCentral() }
+
+    dependencies {
+        val kotlinVersion = "1.5.10"
+        classpath(kotlin("gradle-plugin", version = kotlinVersion))
+        classpath(kotlin("serialization", version = kotlinVersion))
+    }
+}
+
 dependencies {
+    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.2.2")
+    implementation("org.jetbrains.kotlinx:kotlinx-datetime:0.2.1")
     implementation("com.rabbitmq:amqp-client:5.13.0")
     implementation("com.viartemev:the-white-rabbit:0.0.6")
     implementation("com.zaxxer:HikariCP:5.0.0")
@@ -26,7 +39,6 @@ dependencies {
     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 {

docker-compose.yml

@@ -0,0 +1,24 @@
+version: '3'
+
+services:
+
+  output-service-postgres:
+    image: "postgres:13"
+    restart: "always"
+    volumes:
+      - "sample-service-data:/var/lib/postgresql/data"
+    ports:
+      - "127.0.0.1:54321:5432"
+    environment:
+      POSTGRES_USER: "sample-service"
+      POSTGRES_PASSWORD: "sample-service"
+      POSTGRES_DB: "sample-service"
+
+  rabbitmq:
+    image: "registry.kmlabz.com/birbnetes/rabbitmq-federation"
+    ports:
+      - "127.0.0.1:15672:15672"
+      - "127.0.0.1:5672:5672"
+
+volumes:
+  ktor-data:

src/main/kotlin/Main.kt

@@ -1,5 +1,9 @@
+import database.DatabaseFactory
+import database.dao.SampleObjects
 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(SampleObjects)
+    }
+
     val consumerWrapper = ConsumerWrapper();
     consumerWrapper.recieve();
 }

src/main/kotlin/api/ApiObject.kt

@@ -1,10 +1,11 @@
 package api
 
-import com.google.gson.annotations.SerializedName
+import kotlinx.datetime.LocalDateTime
-import java.time.LocalDate
+import kotlinx.serialization.Serializable
 
+@Serializable
 data class ApiObject(
-    @SerializedName("tag") val tag: String,
+    val tag: String,
-    @SerializedName("device_id") val device_id: Int,
+    val device_id: Int,
-    @SerializedName("device_date") val device_date: LocalDate
+    val device_date: LocalDateTime
 )

src/main/kotlin/config/EnvConfig.kt

@@ -2,10 +2,10 @@ package config
 
 data class EnvConfig (
     var mqHost: String = System.getenv("MQ_HOST") ?: "localhost",
-    var mqUserName: String = System.getenv("MQ_USERNAME") ?: "rabbitmq",
+    var mqUserName: String = System.getenv("MQ_USERNAME") ?: "user",
     var mqPassWord: String = System.getenv("MQ_PASSWORD") ?: "rabbitmq",
-    var mqExchange: String = System.getenv("MQ_EXCHANGE") ?: "rabbitmq",
+    var mqExchange: String = System.getenv("MQ_EXCHANGE") ?: "sample",
-    var dbJdbc: String = System.getenv("DB_JDBC") ?: "input",
+    var dbJdbc: String = System.getenv("DB_JDBC") ?: "jdbc:postgresql://localhost:5432/sample-service",
-    var dbUsername: String = System.getenv("DB_USERNAME") ?: "output",
+    var dbUsername: String = System.getenv("DB_USERNAME") ?: "sample-service",
-    var dbPassowrd: String = System.getenv("DB_PASSOWRD") ?: "output"
+    var dbPassowrd: String = System.getenv("DB_PASSWORD") ?: "sample-service"
 )

src/main/kotlin/database/dao/SampleObjects.kt

@@ -2,13 +2,13 @@ package database.dao
 
 import org.jetbrains.exposed.dao.id.IntIdTable
 import org.jetbrains.exposed.sql.*
-import org.jetbrains.exposed.sql.`java-time`.date
+import org.jetbrains.exposed.sql.`java-time`.datetime
-import java.time.LocalDate
+import java.time.LocalDateTime
 
 object SampleObjects : IntIdTable() {
     val tag: Column<String> = varchar("tag", 32)
-    val timestamp: Column<LocalDate> = date("timestamp").default(LocalDate.now())
+    val timestamp: Column<LocalDateTime> = datetime("timestamp").default(LocalDateTime.now())
     val device_id: Column<Int> = integer("device_id")
-    val device_date: Column<LocalDate> = date("device_date")
+    val device_date: Column<LocalDateTime> = datetime("device_date")
     override val primaryKey = PrimaryKey(id, name = "PK_SampleObject_Id")
 }

src/main/kotlin/database/service/SampleObjectService.kt

@@ -2,6 +2,7 @@ package database.service
 
 import database.model.SampleObject
 import api.ApiObject
+import kotlinx.datetime.toJavaLocalDateTime
 import org.jetbrains.exposed.sql.transactions.transaction
 
 class SampleObjectService : ISampleObjectService {
@@ -10,7 +11,7 @@ class SampleObjectService : ISampleObjectService {
         transaction {
             SampleObject.new {
                 tag = apiObject.tag
-                device_date = apiObject.device_date
+                device_date = apiObject.device_date.toJavaLocalDateTime()
                 device_id = apiObject.device_id
             }
         }

src/main/kotlin/mq/ConsumerWrapper.kt

@@ -16,10 +16,10 @@ class ConsumerWrapper {
         val inputConnection = factory.newConnection()
         val inputChannel = inputConnection.createChannel()
 
-        inputChannel.exchangeDeclare(envConfig.mqExchange, BuiltinExchangeType.FANOUT)
+        inputChannel.exchangeDeclare(envConfig.mqExchange, BuiltinExchangeType.DIRECT)
         val inputQueueName = inputChannel.queueDeclare().queue
-        inputChannel.queueBind(inputQueueName, envConfig.mqExchange, "")
+        inputChannel.queueBind(inputQueueName, envConfig.mqExchange, "meta")
 
-        inputChannel.basicConsume(inputQueueName, true, DatabaseConsumer())
+        inputChannel.basicConsume(inputQueueName, false, DatabaseConsumer(inputChannel))
     }
 }

src/main/kotlin/mq/DatabaseConsumer.kt

@@ -1,27 +1,28 @@
 package mq
 
 import api.ApiObject
-import com.google.gson.Gson
+import kotlinx.serialization.json.*
-import com.rabbitmq.client.AMQP
+import com.rabbitmq.client.*
-import com.rabbitmq.client.Consumer
-import com.rabbitmq.client.Envelope
-import com.rabbitmq.client.ShutdownSignalException
 import database.service.ISampleObjectService
+import kotlinx.serialization.decodeFromString
 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 : ISampleObjectService by inject()
+    private val resultObjectService: ISampleObjectService by inject()
-    private val gson = Gson()
+    private val basicChannel = channel
-    override fun handleConsumeOk(consumerTag : String?) {
+    override fun handleConsumeOk(consumerTag: String?) {
     }
-    override fun handleCancelOk(p0 : String?) {
+
+    override fun handleCancelOk(p0: String?) {
         throw UnsupportedOperationException()
     }
-    override fun handleRecoverOk(p0 : String?) {
+
+    override fun handleRecoverOk(p0: String?) {
         throw UnsupportedOperationException()
     }
-    override fun handleCancel(p0 : String?) {
+
+    override fun handleCancel(p0: String?) {
         throw UnsupportedOperationException()
     }
 
@@ -29,9 +30,20 @@ class DatabaseConsumer: Consumer, KoinComponent {
         println("got shutdown signal")
     }
 
-    override fun handleDelivery(consumerTag : String?, envelope : Envelope?, basicProperties : AMQP.BasicProperties?, body : ByteArray?) {
+    override fun handleDelivery(
-        val rawJson = body!!.toString(Charsets.UTF_8)
+        consumerTag: String?,
-        val apiObject = gson.fromJson(rawJson, ApiObject::class.java)
+        envelope: Envelope?,
-        resultObjectService.addOne(apiObject)
+        basicProperties: AMQP.BasicProperties?,
+        body: ByteArray?
+    ) {
+        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)
         }
+    }
 }