Revert "Revert "Improve concurrency (#62)""

This reverts commit 9383602.

Author: tipsy

src/main/kotlin/app/GhService.kt

@@ -7,7 +7,9 @@ import org.eclipse.egit.github.core.service.RepositoryService
 import org.eclipse.egit.github.core.service.UserService
 import org.eclipse.egit.github.core.service.WatcherService
 import org.slf4j.LoggerFactory
-import java.util.*
+import java.io.IOException
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
 
 object GhService {
@@ -16,8 +18,11 @@ object GhService {
 
     private val log = LoggerFactory.getLogger(GhService.javaClass)
 
+    // Allows for parallel iteration and O(1) put/remove
+    private val clientSessions = ConcurrentHashMap<WsSession, Boolean>()
+
     private val tokens = Config.getApiTokens()?.split(",") ?: listOf("") // empty token is limited to 60 requests
-    private val clients = tokens.map { token -> GitHubClient().apply { setOAuth2Token(token) } }
+    private val clients = tokens.map { GitHubClient().apply { setOAuth2Token(it) } }
     private val repoServices = clients.map { RepositoryService(it) }
     private val commitServices = clients.map { CommitService(it) }
     private val userServices = clients.map { UserService(it) }
@@ -30,9 +35,15 @@ object GhService {
 
     val remainingRequests: Int get() = clients.sumBy { it.remainingRequests }
 
-    init { // create timer to ping clients every other minute to make sure remainingRequests is correct
-        Timer().scheduleAtFixedRate(object : TimerTask() {
-            override fun run() {
+    fun registerClient(ws: WsSession) = clientSessions.put(ws, true) == true
+
+    fun unregisterClient(ws: WsSession) = clientSessions.remove(ws) == true
+
+    init {
+        Executors.newScheduledThreadPool(2).apply {
+
+            // ping clients every other minute to make sure remainingRequests is correct
+            scheduleAtFixedRate({
                 repoServices.forEach {
                     try {
                         it.getRepository("tipsy", "github-profile-summary")
@@ -41,18 +52,22 @@ object GhService {
                         log.info("Pinged client ${clients.indexOf(it.client)} - was rate-limited")
                     }
                 }
-            }
-        }, 0, TimeUnit.MILLISECONDS.convert(2, TimeUnit.MINUTES))
-    }
+            }, 0, 2, TimeUnit.MINUTES)
+
+            // update all connected clients with remainingRequests twice per second
+            scheduleAtFixedRate({
+                val payload = remainingRequests.toString()
+                clientSessions.forEachKey(1) {
+                    try {
+                        if (it.isOpen)
+                            it.send(payload)
+                    } catch (e: IOException) {
+                        log?.error(e.toString())
+                    }
+                }
+            }, 0, 500, TimeUnit.MILLISECONDS)
 
-    fun broadcastRemainingRequests(session: WsSession) = object : TimerTask() {
-        override fun run() {
-            if (session.isOpen) {
-                return session.send(GhService.remainingRequests.toString())
-            }
-            this.cancel()
         }
     }
 
 }
-

src/main/kotlin/app/Main.kt

@@ -12,7 +12,6 @@ import org.eclipse.jetty.server.Server
 import org.eclipse.jetty.server.ServerConnector
 import org.eclipse.jetty.util.thread.QueuedThreadPool
 import org.slf4j.LoggerFactory
-import java.util.*
 
 fun main(args: Array<String>) {
 
@@ -62,9 +61,9 @@ fun main(args: Array<String>) {
         }
 
         ws("/rate-limit-status") { ws ->
-            ws.onConnect { session ->
-                Timer().scheduleAtFixedRate(GhService.broadcastRemainingRequests(session), 0, 1000)
-            }
+            ws.onConnect { GhService.registerClient(it) }
+            ws.onClose { session, _, _ -> GhService.unregisterClient(session) }
+            ws.onError { session, _ -> GhService.unregisterClient(session) }
         }
 
     }