clear backlog before connecting to message stream

1.12.2/src/main/kotlin/matterlink/MatterLink.kt

@@ -42,7 +42,7 @@ object MatterLink : IMatterLink() {
 
     @Mod.EventHandler
     fun serverAboutToStart(event: FMLServerAboutToStartEvent) {
-        MessageHandler.start()
+        MessageHandler.start(clear = true)
     }
 
     @Mod.EventHandler
@@ -50,12 +50,12 @@ object MatterLink : IMatterLink() {
         logger.debug("Registering server commands")
         event.registerServerCommand(CommandMatterlink())
 
-        MessageHandler.rcvQueue.clear()
+        connect()
     }
 
     @Mod.EventHandler
     fun serverStopping(event: FMLServerStoppingEvent) {
-        MessageHandler.stop()
+        disconnect()
     }
 
     //FORGE-DEPENDENT

1.12.2/src/main/kotlin/matterlink/command/CommandMatterlink.kt

@@ -4,6 +4,7 @@ import com.google.common.collect.Lists
 import matterlink.MODID
 import matterlink.bridge.MessageHandler
 import matterlink.bridge.ServerChatHandler
+import matterlink.instance
 import matterlink.logger
 import net.minecraft.command.CommandBase
 import net.minecraft.command.ICommandSender
@@ -40,17 +41,10 @@ class CommandMatterlink : CommandBase() {
         val cmd = args[0].toLowerCase()
         when (cmd) {
             "connect" -> {
-                if (MessageHandler.start()) {
+                instance.connect()
-                    MessageHandler.rcvQueue.clear()
-                    logger.info("Connected to matterbridge relay")
-                    ServerChatHandler.processMessages = true
-                } else {
-                    logger.error("Connection to matterbridge relay failed.")
-                }
             }
             "disconnect" -> {
-                MessageHandler.stop()
+                instance.disconnect()
-                ServerChatHandler.processMessages = false
             }
         }
     }

build.gradle

@@ -6,7 +6,7 @@ subprojects {
     apply plugin: 'idea'
 
     group = 'matterlink'
-    version = '1.1.0'
+    version = '1.1.1'
 
     idea {
         module {

core/src/main/kotlin/matterlink/IMatterLink.kt

@@ -1,5 +1,7 @@
 package matterlink
 
+import matterlink.bridge.MessageHandler
+import matterlink.bridge.ServerChatHandler
 import org.apache.logging.log4j.Level
 import org.apache.logging.log4j.Logger
 import org.apache.logging.log4j.message.SimpleMessageFactory
@@ -25,16 +27,28 @@ abstract class IMatterLink {
     abstract fun wrappedSendToPlayers(msg: String)
 
     abstract fun wrappedPlayerList(): Array<String>
+
+    fun connect() {
+        if (MessageHandler.start(clear = true)) {
+            logger.info("Connected to matterbridge relay")
+        } else {
+            logger.error("Connection to matterbridge relay failed.")
+        }
+    }
+
+    fun disconnect  () {
+        MessageHandler.stop()
+    }
 }
 
 
 class DummyLink : IMatterLink() {
     override fun wrappedPlayerList(): Array<String> {
-        TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
+        TODO("not implemented, make sure the MatterLink implementations is loaded") //To change body of created functions use File | Settings | File Templates.
     }
 
     override fun wrappedSendToPlayers(msg: String) {
-        TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
+        TODO("not implemented, make sure the MatterLink implementations is loaded") //To change body of created functions use File | Settings | File Templates.
     }
 
 }

core/src/main/kotlin/matterlink/bridge/HttpStreamConnection.kt

@@ -8,16 +8,23 @@ import java.net.SocketException
 
 val BUFFER_SIZE = 1000
 
-class HttpStreamConnection(private val getClosure: () -> HttpGet, private val mhandler: (String) -> Unit, private val onClose: () -> Unit) : Thread() {
+class HttpStreamConnection(getClosure: () -> HttpGet, clearClosure: () -> HttpGet, private val mhandler: (String) -> Unit, private val onClose: () -> Unit, private val clear: Boolean = true) : Thread() {
     private val client = HttpClients.createDefault()
     private var stream: InputStream? = null
 
     val get = getClosure()
+    private val clearGet = clearClosure()
     var cancelled: Boolean = false
         private set
 
 
     override fun run() {
+        if(clear) {
+            val r = client.execute(clearGet)
+            r.entity.content.bufferedReader().forEachLine {
+                logger.debug("skipping $it")
+            }
+        }
         val response = client.execute(get)
         val content = response.entity.content.buffered()
         stream = content

core/src/main/kotlin/matterlink/bridge/MessageHandler.kt

@@ -25,12 +25,12 @@ object MessageHandler {
         connected = true
     }
 
-    fun HttpRequestBase.authorize() {
+    private fun HttpRequestBase.authorize() {
         if (cfg!!.connect.authToken.isNotEmpty() && getHeaders("Authorization").isEmpty())
             setHeader("Authorization", "Bearer " + cfg!!.connect.authToken)
     }
 
-    private fun createThread(): HttpStreamConnection {
+    private fun createThread(clear: Boolean = true): HttpStreamConnection {
         logger.info("Attempting to open bridge connection.")
         return HttpStreamConnection(
                 {
@@ -38,6 +38,11 @@ object MessageHandler {
                         authorize()
                     }
                 },
+                {
+                    HttpGet(cfg!!.connect.url + "/api/messages").apply {
+                        authorize()
+                    }
+                },
                 {
                     rcvQueue.add(
                             ApiMessage.decode(it)
@@ -47,7 +52,8 @@ object MessageHandler {
                 {
                     logger.info("Bridge connection closed!")
                     connected = false
-                }
+                },
+                clear
         )
     }
 
@@ -64,9 +70,9 @@ object MessageHandler {
         streamConnection.close()
     }
 
-    fun start(): Boolean {
+    fun start(clear: Boolean = true): Boolean {
         if (!connected)
-            streamConnection = createThread()
+            streamConnection = createThread(clear)
         if (!streamConnection.isAlive) {
             streamConnection.start()
 //            MessageHandler.transmit(ApiMessage(text="bridge connected", username="Server"))

core/src/main/kotlin/matterlink/bridge/ServerChatHandler.kt

@@ -6,13 +6,11 @@ import matterlink.bridge.command.BridgeCommandRegistry
 import matterlink.cfg
 
 object ServerChatHandler {
-    var processMessages: Boolean = true
 
     /**
      * This method must be called every server tick with no arguments.
      */
     fun writeIncomingToChat() {
-        if (!processMessages) return
         if (MessageHandler.rcvQueue.isNotEmpty())
             logger.debug("incoming: " + MessageHandler.rcvQueue.toString())
         val nextMessage = MessageHandler.rcvQueue.poll()