basic 2-way communication

src/main/kotlin/org/foxarmy/chatProxy/ChatListener.kt

@@ -1,4 +1,25 @@
 package org.foxarmy.chatProxy
 
-class ChatListener {
+import com.google.gson.Gson
+import org.bukkit.event.EventHandler
+import org.bukkit.event.Listener
+import org.bukkit.event.player.AsyncPlayerChatEvent
+import org.bukkit.plugin.java.JavaPlugin
+import java.time.Instant
+
+class ChatListener(private val kafkaClient: KafkaClient) : Listener{
+
+    @EventHandler
+    fun onChat(event: AsyncPlayerChatEvent) {
+        val player = event.player
+        val message = event.message
+        val timestamp = Instant.now().epochSecond
+
+        val data = MessageJson(player.name, message, "server", timestamp)
+        val gson = Gson()
+        val gsonData = gson.toJson(data)
+
+        kafkaClient.sendData(gsonData)
+    }
+
 }

src/main/kotlin/org/foxarmy/chatProxy/ChatProxy.kt

@@ -5,7 +5,15 @@ import org.bukkit.plugin.java.JavaPlugin
 class ChatProxy : JavaPlugin() {
 
     override fun onEnable() {
-        // Plugin startup logic
+        saveDefaultConfig()
+        reloadConfig()
+        val kafkaClient = KafkaClient(this, config.getStringList("kafkaBrokers"))
+        server.pluginManager.registerEvents(ChatListener(kafkaClient), this)
+
+        this.getCommand("chatproxy")?.setExecutor(Commands(this))
+
+
+
     }
 
     override fun onDisable() {

src/main/kotlin/org/foxarmy/chatProxy/Commands.kt

@@ -1,4 +1,35 @@
 package org.foxarmy.chatProxy
 
-class Commands {
+import org.bukkit.command.Command
+import org.bukkit.command.CommandExecutor
+import org.bukkit.command.CommandSender
+import org.bukkit.plugin.java.JavaPlugin
+import java.util.*
+
+class Commands(private val plugin: JavaPlugin) : CommandExecutor {
+    override fun onCommand(sender: CommandSender, command: Command, label: String, args: Array<out String>): Boolean {
+        if (!sender.hasPermission("chatproxy.use")) {
+            sender.sendMessage("You do not have permission to use this command.")
+            return true
+        }
+        if (args.isEmpty()) {
+            sender.sendMessage("Usage: /chatproxy <subcommand>")
+            return true
+        }
+
+        when (args[0].lowercase(Locale.getDefault())) {
+            "reload" -> {
+                reloadConfig(sender)
+                return true
+            }
+            else -> {
+                sender.sendMessage("Unknown subcommand. Usage: /chatproxy <reload>")
+                return true
+            }
+        }
+    }
+    private fun reloadConfig(sender: CommandSender) {
+        plugin.reloadConfig()
+        sender.sendMessage("Configuration reloaded successfully.")
+    }
 }

src/main/kotlin/org/foxarmy/chatProxy/KafkaClient.kt

@@ -1,4 +1,60 @@
 package org.foxarmy.chatProxy
 
-class KafkaClient {
+import com.google.gson.Gson
+import org.apache.kafka.clients.consumer.Consumer
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.kafka.clients.consumer.ConsumerRecords
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.Producer
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.apache.kafka.common.serialization.StringSerializer
+import org.bukkit.Bukkit
+import org.bukkit.entity.Player
+import org.bukkit.event.player.AsyncPlayerChatEvent
+import org.bukkit.plugin.java.JavaPlugin
+import java.time.Duration
+import java.util.*
+
+class KafkaClient(plugin: JavaPlugin, brokers: List<String>) {
+    private val consumer: Consumer<String, String>
+    private val producer: Producer<String, String>
+
+    init {
+        Thread.currentThread().contextClassLoader = null;
+        val props = Properties()
+
+        props["bootstrap.servers"] = brokers
+        props["group.id"] = "chat-listener"
+        props["key.deserializer"] = StringDeserializer::class.java.name
+        props["value.deserializer"] = StringDeserializer::class.java.name
+        props["key.serializer"] = StringSerializer::class.java.canonicalName
+        props["value.serializer"] = StringSerializer::class.java.canonicalName
+
+        producer = KafkaProducer<String, String>(props)
+        consumer = KafkaConsumer<String, String>(props)
+
+        consumer.subscribe(listOf("chatMessage"))
+        listenForMessages(plugin)
+    }
+
+    private fun listenForMessages(plugin: JavaPlugin) {
+        Bukkit.getScheduler().runTaskTimer(
+            plugin, Runnable {
+                val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofMillis(100))
+                for (record: ConsumerRecord<String, String> in records) {
+                    val recordJSON = Gson().fromJson(record.value(), MessageJson::class.java)
+                    if (recordJSON.origin == "server") continue
+                    Bukkit.broadcastMessage("<${recordJSON.author}> ${recordJSON.content}")
+                    recordJSON.origin = "server"
+                    sendData(recordJSON.toString())
+                }
+            }, 0L, 20L
+        )
+    }
+
+    fun sendData(data: String) {
+        val futureResult = producer.send(ProducerRecord<String, String>("chatMessage", data))
+    }
 }

src/main/kotlin/org/foxarmy/chatProxy/MessageJson.kt

@@ -1,4 +1,3 @@
 package org.foxarmy.chatProxy
 
-class MessageJson {
-}
+data class MessageJson(val author: String, val content: String, var origin: String, val timestamp: Long)

src/main/resources/config.yml

@@ -0,0 +1,3 @@
+kafkaBrokers:
+  - localhost:9092
+  - kafka:9092

src/main/resources/plugin.yml

@@ -6,3 +6,15 @@ prefix: Chatproxy
 authors: [ leca ]
 description: Plugin for proxying chat messages from mc server to site
 website: https://git.foxarmy.org/leca/Chatproxy
+commands:
+  chatproxy:
+    description: Chat proxy commands
+    usage: /<command> <subcommand>
+    aliases: [ cprox ]
+    permission: chatproxy.use
+    permission-message: You do not have permission to use this command.
+    reload:
+      description: Reload plugin's configuration
+      usage: /<command> reload
+      permission: chatproxy.reload
+      permission-message: You do not have permission to reload configuration.