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

manual ack
continuous-integration/drone/push Build is passing Details

This commit is contained in:
Torma Kristóf 2021-08-17 17:38:09 +02:00
parent a719addee4
commit 79b68af64c
Signed by: tormakris
GPG Key ID: DC83C4F2C41B1047
2 changed files with 13 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/kotlin/mq/ConsumerWrapper.kt
View File

@ -20,6 +20,6 @@ class ConsumerWrapper {
val inputQueueName = inputChannel.queueDeclare().queue val inputQueueName = inputChannel.queueDeclare().queue
inputChannel.queueBind(inputQueueName, envConfig.mqExchange, "") inputChannel.queueBind(inputQueueName, envConfig.mqExchange, "")
inputChannel.basicConsume(inputQueueName, true, DatabaseConsumer()) inputChannel.basicConsume(inputQueueName, false, DatabaseConsumer(inputChannel))
} }
} }

20
src/main/kotlin/mq/DatabaseConsumer.kt
View File

@ -2,17 +2,15 @@ package mq
import api.ApiObject import api.ApiObject
import com.google.gson.Gson import com.google.gson.Gson
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.IResultObjectService import database.service.IResultObjectService
import org.koin.core.component.KoinComponent import org.koin.core.component.KoinComponent
import org.koin.core.component.inject import org.koin.core.component.inject
class DatabaseConsumer: Consumer, KoinComponent { class DatabaseConsumer(channel: Channel): Consumer, KoinComponent {
private val resultObjectService : IResultObjectService by inject() private val resultObjectService : IResultObjectService by inject()
private val gson = Gson() 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?) {
@ -30,8 +28,14 @@ class DatabaseConsumer: Consumer, KoinComponent {
} }
override fun handleDelivery(consumerTag : String?, envelope : Envelope?, basicProperties : AMQP.BasicProperties?, body : ByteArray?) { override fun handleDelivery(consumerTag : String?, envelope : Envelope?, basicProperties : AMQP.BasicProperties?, body : ByteArray?) {
val rawJson = body!!.toString(Charsets.UTF_8) try {
val apiObject = gson.fromJson(rawJson, ApiObject::class.java) val rawJson = body!!.toString(Charsets.UTF_8)
resultObjectService.addOne(apiObject) val apiObject = gson.fromJson(rawJson, ApiObject::class.java)
resultObjectService.addOne(apiObject)
basicChannel.basicAck(envelope!!.deliveryTag, false)
} catch (e: Exception) {
println(e.message)
basicChannel.basicNack(envelope!!.deliveryTag, false, true)
}
} }
} }