finish consumer

src/main/kotlin/Main.kt

@@ -1,4 +1,5 @@
 import di.databasemodule
+import mq.ConsumerWrapper
 import org.koin.core.context.GlobalContext.startKoin
 
 fun main(vararg args: String) {
@@ -8,5 +9,6 @@ fun main(vararg args: String) {
         modules(databasemodule)
     }
 
-
+    val consumerWrapper = ConsumerWrapper();
+    consumerWrapper.recieve();
 }

src/main/kotlin/config/EnvConfig.kt

@@ -4,6 +4,7 @@ data class EnvConfig (
     var mqHost: String = System.getenv("MQ_HOST") ?: "localhost",
     var mqUserName: String = System.getenv("MQ_USERNAME") ?: "rabbitmq",
     var mqPassWord: String = System.getenv("MQ_PASSWORD") ?: "rabbitmq",
+    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"

src/main/kotlin/di/DatabaseModule.kt

@@ -5,5 +5,5 @@ import database.service.ResultObjectService
 import org.koin.dsl.module
 
 val databasemodule = module(createdAtStart = true) {
-    single { ResultObjectService() as IResultObjectService }
+    single<IResultObjectService> { ResultObjectService() }
 }

src/main/kotlin/mq/ConsumerWrapper.kt

@@ -0,0 +1,25 @@
+package mq
+
+import com.rabbitmq.client.BuiltinExchangeType
+import com.rabbitmq.client.ConnectionFactory
+import config.EnvConfig
+
+class ConsumerWrapper {
+    private val envConfig = EnvConfig()
+
+    fun recieve(){
+        val factory = ConnectionFactory()
+        factory.host = envConfig.mqHost
+        factory.username = envConfig.mqUserName
+        factory.password = envConfig.mqPassWord
+
+        val inputConnection = factory.newConnection()
+        val inputChannel = inputConnection.createChannel()
+
+        inputChannel.exchangeDeclare(envConfig.mqExchange, BuiltinExchangeType.FANOUT)
+        val inputQueueName = inputChannel.queueDeclare().queue
+        inputChannel.queueBind(inputQueueName, envConfig.mqExchange, "")
+
+        inputChannel.basicConsume(inputQueueName, true, DatabaseConsumer())
+    }
+}

src/main/kotlin/mq/DatabaseConsumer.kt

@@ -0,0 +1,37 @@
+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 database.service.IResultObjectService
+import org.koin.core.component.KoinComponent
+import org.koin.core.component.inject
+
+class DatabaseConsumer: Consumer, KoinComponent {
+    private val resultObjectService : IResultObjectService by inject()
+    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 handleShutdownSignal(consumerTag: String?, sig: ShutdownSignalException?) {
+        println("got shutdown signal")
+    }
+
+    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)
+        }
+}