1 files changed, 38 insertions, 0 deletions
diff --git a/opendc-common/src/main/kotlin/org/opendc/common/utils/Kafka.kt b/opendc-common/src/main/kotlin/org/opendc/common/utils/Kafka.kt
new file mode 100644
index 00000000..81eb6752
--- /dev/null
+++ b/opendc-common/src/main/kotlin/org/opendc/common/utils/Kafka.kt
@@ -0,0 +1,38 @@
+package org.opendc.common.utils
+
+import com.fasterxml.jackson.dataformat.toml.TomlMapper
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.opendc.common.ProtobufMetrics
+import java.util.*
+/**
+ * Represents the Kafka interface.
+ * @constructor `topic` the Kafka topic
+ * @author Mateusz Kwiatkowski
+ * @see <a href=https://kafka.apache.org/23/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html>
+ *     https://kafka.apache.org/23/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html</a>
+ * @see <a href=https://kafka.apache.org/23/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html>
+ *      https://kafka.apache.org/23/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html</a>
+ */
+
+@Suppress("DEPRECATION")
+public class Kafka(private val topic: String) {
+    private var properties : Properties
+
+    init {
+        check(!topic.contains("-"))
+        properties  = TomlMapper().readerFor(Properties().javaClass)
+            .readValue(Kafka::class.java.getResource("/producer.toml"))
+    }
+
+    public fun getSend() : (ProtobufMetrics.ProtoExport) -> Unit {
+        val producer = KafkaProducer<String, ProtobufMetrics.ProtoExport>(properties)
+        return fun (value : ProtobufMetrics.ProtoExport) {
+            try {
+                producer.send(ProducerRecord(this.topic, value))
+            } catch (e: Exception) {
+                println("${e.message}")
+            }
+        }
+    }
+}
+\ No newline at end of file