remove consumer and update deps
	
		
			
	
		
	
	
		
	
		
			All checks were successful
		
		
	
	
		
			
				
	
				continuous-integration/drone/push Build is passing
				
			
		
		
	
	
				
					
				
			
		
			All checks were successful
		
		
	
	continuous-integration/drone/push Build is passing
				
			This commit is contained in:
		@@ -1,7 +1,6 @@
 | 
			
		||||
package com.kmalbz
 | 
			
		||||
 | 
			
		||||
import com.kmalbz.api.route.OutputServiceRDBServer
 | 
			
		||||
import com.kmalbz.consumer.DatabaseConsumer
 | 
			
		||||
import io.ktor.application.*
 | 
			
		||||
import io.ktor.response.*
 | 
			
		||||
import io.ktor.routing.*
 | 
			
		||||
@@ -12,9 +11,6 @@ import org.apache.http.HttpException
 | 
			
		||||
import com.kmalbz.database.DatabaseFactory
 | 
			
		||||
import com.kmalbz.database.dao.ResultObjects
 | 
			
		||||
import io.ktor.util.KtorExperimentalAPI
 | 
			
		||||
import com.rabbitmq.client.*
 | 
			
		||||
import com.typesafe.config.ConfigFactory
 | 
			
		||||
import io.ktor.config.HoconApplicationConfig
 | 
			
		||||
import org.jetbrains.exposed.sql.SchemaUtils
 | 
			
		||||
import org.jetbrains.exposed.sql.transactions.transaction
 | 
			
		||||
import org.koin.ktor.ext.Koin
 | 
			
		||||
@@ -39,24 +35,6 @@ fun Application.module() {
 | 
			
		||||
        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, "")
 | 
			
		||||
 | 
			
		||||
    channel.basicConsume(queueName, true, DatabaseConsumer())
 | 
			
		||||
 | 
			
		||||
    routing {
 | 
			
		||||
        install(StatusPages) {
 | 
			
		||||
            exception<HttpException> {
 | 
			
		||||
 
 | 
			
		||||
@@ -1,35 +0,0 @@
 | 
			
		||||
package com.kmalbz.consumer
 | 
			
		||||
 | 
			
		||||
import com.google.gson.Gson
 | 
			
		||||
import com.kmalbz.api.model.ApiObject
 | 
			
		||||
import com.kmalbz.database.service.ResultObjectService
 | 
			
		||||
import com.rabbitmq.client.AMQP.BasicProperties
 | 
			
		||||
import com.rabbitmq.client.Consumer
 | 
			
		||||
import com.rabbitmq.client.Envelope
 | 
			
		||||
import com.rabbitmq.client.ShutdownSignalException
 | 
			
		||||
 | 
			
		||||
class DatabaseConsumer : Consumer {
 | 
			
		||||
    private val resultObjectService = ResultObjectService()
 | 
			
		||||
 | 
			
		||||
    private val gson = Gson()
 | 
			
		||||
    override fun handleConsumeOk(consumerTag : String?) {
 | 
			
		||||
    }
 | 
			
		||||
    override fun handleCancelOk(p0 : String?) {
 | 
			
		||||
        throw UnsupportedOperationException()
 | 
			
		||||
    }
 | 
			
		||||
    override fun handleRecoverOk(p0 : String?) {
 | 
			
		||||
        throw UnsupportedOperationException()
 | 
			
		||||
    }
 | 
			
		||||
    override fun handleCancel(p0 : String?) {
 | 
			
		||||
        throw UnsupportedOperationException()
 | 
			
		||||
    }
 | 
			
		||||
    override fun handleDelivery(consumerTag : String?, envelope : Envelope?, basicProperties : BasicProperties?, body : ByteArray?) {
 | 
			
		||||
        val rawJson = body!!.toString(Charsets.UTF_8)
 | 
			
		||||
        val apiObject = gson.fromJson(rawJson, ApiObject::class.java)
 | 
			
		||||
 | 
			
		||||
        resultObjectService.addOne(apiObject)
 | 
			
		||||
    }
 | 
			
		||||
    override fun handleShutdownSignal(p0 : String?, p1 : ShutdownSignalException?) {
 | 
			
		||||
        println("got shutdown signal")
 | 
			
		||||
    }
 | 
			
		||||
}
 | 
			
		||||
		Reference in New Issue
	
	Block a user