removed background thread + scalariform

Author: simplesteph

src/main/scala/com/github/simplesteph/ksm/AclSynchronizer.scala

@@ -43,13 +43,13 @@ object AclSynchronizer {
 class AclSynchronizer(
   authorizer: Authorizer,
   sourceAcl: SourceAcl,
-  notification: Notification) extends Runnable {
+  notification: Notification) {
 
   import AclSynchronizer._
 
   private var sourceAclsCache: SourceAclResult = _
 
-  override def run(): Unit = {
+  def run(): Unit = {
 
     // flatten the Kafka ACL
     val kafkaAcls: Set[(Resource, Acl)] = flattenKafkaAcls(authorizer.getAcls())

src/main/scala/com/github/simplesteph/ksm/KafkaSecurityManager.scala

@@ -1,6 +1,6 @@
 package com.github.simplesteph.ksm
 
-import java.util.concurrent.{ Executors, ScheduledFuture, TimeUnit }
+import java.util.concurrent.atomic.AtomicBoolean
 
 import com.github.simplesteph.ksm.parser.CsvAclParser
 import com.typesafe.config.ConfigFactory
@@ -13,6 +13,8 @@ object KafkaSecurityManager extends App {
   val config = ConfigFactory.load()
   val appConfig: AppConfig = new AppConfig(config)
 
+  var isCancelled: AtomicBoolean = new AtomicBoolean(false)
+
   if (appConfig.KSM.extract) {
     new ExtractAcl(appConfig.Authorizer.authorizer, CsvAclParser).extract()
   } else {
@@ -21,20 +23,18 @@ object KafkaSecurityManager extends App {
       appConfig.Source.sourceAcl,
       appConfig.Notification.notification)
 
-    val executor = Executors.newScheduledThreadPool(1)
-    val f: ScheduledFuture[_] = executor.scheduleAtFixedRate(aclSynchronizer, 1000,
-      appConfig.KSM.refreshFrequencyMs, TimeUnit.MILLISECONDS)
-
     Runtime.getRuntime.addShutdownHook(new Thread() {
       override def run(): Unit = {
+        log.info("Received stop signal, Kafka Security Manager is shutting down...")
+        isCancelled = new AtomicBoolean(true)
         aclSynchronizer.close()
-        log.info("Kafka Security Manager is shutting down...")
-        f.cancel(false)
-        log.info("Waiting for thread to cleanly shutdown (10 seconds maximum)")
-        executor.shutdown()
-        executor.awaitTermination(10, TimeUnit.SECONDS)
-        log.info("Kafka Security Manager has shut down...")
       }
     })
+
+    while (!isCancelled.get()) {
+      aclSynchronizer.run()
+      Thread.sleep(appConfig.KSM.refreshFrequencyMs)
+    }
+
   }
 }

src/main/scala/com/github/simplesteph/ksm/source/GitHubSourceAcl.scala

@@ -62,14 +62,14 @@ class GitHubSourceAcl extends SourceAcl {
     })
 
     // we use this header for the 304
-    lastModified.foreach( header => request.header("If-Modified-Since", header))
+    lastModified.foreach(header => request.header("If-Modified-Since", header))
     val response: Response = HTTP.get(request)
 
     response.status match {
       case 200 =>
         lastModified = response.header("Last-Modified")
         val b64encodedContent = objectMapper.readTree(response.textBody).get("content").asText()
-        val data = new String(Base64.getDecoder.decode(b64encodedContent.replace("\n","").replace("\r","")), Charset.forName("UTF-8"))
+        val data = new String(Base64.getDecoder.decode(b64encodedContent.replace("\n", "").replace("\r", "")), Charset.forName("UTF-8"))
         // use the CSV Parser
         Some(CsvAclParser.aclsFromReader(new StringReader(data)))
       case 304 =>