[FIX] Implement actual ring buffer to prevent memory leaks when influx is down

Author: Jan Boroń

build.sbt

@@ -5,6 +5,7 @@ val defaultScalaVersion = "3.3.1"
 val scalaVersions = Seq("2.11.12", "2.12.18", "2.13.12", defaultScalaVersion)
 
 val asyncHttpClientV    = "2.12.3"
+val commonsCollectionsV = "4.4"
 val dropwizardMetricsV  = "4.0.2"
 val findbugsV           = "3.0.1"
 val hikariCPV           = "3.2.0"
@@ -96,6 +97,7 @@ lazy val core = project.in(file("core"))
         "io.dropwizard.metrics"        % "metrics-core"                 % dropwizardMetricsV,
         "org.scala-lang.modules"      %% "scala-collection-compat"      % scalaCompatV,
         "com.typesafe.scala-logging"  %% "scala-logging"                % scalaLoggingV,
+        "org.apache.commons"           % "commons-collections4"         % commonsCollectionsV,
 
         "ch.qos.logback"               % "logback-classic"              % logbackV              % Test,
         "org.scalatest"               %% "scalatest"                    % scalaTestV            % Test,

core/src/main/scala/influxdbreporter/core/WriterDataBuffer.scala

@@ -16,31 +16,29 @@
 package influxdbreporter.core
 
 import influxdbreporter.core.writers.WriterData
+import org.apache.commons.collections4.queue.CircularFifoQueue
 
-import scala.collection.immutable.List
-import scala.collection.mutable.ListBuffer
+import scala.jdk.CollectionConverters.{CollectionHasAsScala, SeqHasAsJava}
 
 trait WriterDataBuffer[T] {
 
-  def update(add: List[WriterData[T]] = Nil, remove: List[WriterData[T]] = Nil): List[WriterData[T]]
+  def update(add: List[WriterData[T]] = Nil, remove: List[WriterData[T]] = Nil): Unit
 
   def get(): List[WriterData[T]]
 }
 
 class FixedSizeWriterDataBuffer[T](maxSize: Int)
   extends WriterDataBuffer[T] {
 
-  private var ringBuffer: ListBuffer[WriterData[T]] = ListBuffer.empty
+  private val buffer: CircularFifoQueue[WriterData[T]] = new CircularFifoQueue[WriterData[T]](maxSize)
 
-  override def update(add: List[WriterData[T]] = Nil, remove: List[WriterData[T]] = Nil): List[WriterData[T]] = {
-    if (add.nonEmpty || remove.nonEmpty) synchronized {
-      ringBuffer --= remove
-      add ++=: ringBuffer
-      ringBuffer = ringBuffer.distinct.take(maxSize)
+  override def update(add: List[WriterData[T]] = Nil, remove: List[WriterData[T]] = Nil): Unit = {
+    if(add.nonEmpty || remove.nonEmpty) {
+      buffer.removeAll(remove.asJava)
+      buffer.addAll(add.asJava)
     }
-    ringBuffer.toList
   }
 
-  override def get(): List[WriterData[T]] = ringBuffer.toList
+  override def get(): List[WriterData[T]] = buffer.asScala.toList
 
 }

core/src/test/scala/influxdbreporter/core/FixedSizeWriterDataBufferTests.scala

@@ -30,27 +30,27 @@ class FixedSizeWriterDataBufferTests extends AnyWordSpec {
         buffer.get()
       }
 
-      assertResult(wd(4) :: wd(5) :: wd(1) :: wd(2) :: Nil) {
+      assertResult(wd(2) :: wd(3) :: wd(4) :: wd(5) :: Nil) {
         buffer.update(add = wd(4) :: wd(5) :: Nil)
         buffer.get()
       }
 
-      assertResult(wd(4) :: wd(5) :: wd(2) :: Nil) {
-        buffer.update(remove = wd(1) :: Nil)
+      assertResult(wd(3) :: wd(4) :: wd(5) :: Nil) {
+        buffer.update(remove = wd(2) :: Nil)
         buffer.get()
       }
 
-      assertResult(wd(6) :: wd(7) :: wd(4) :: wd(2) :: Nil) {
+      assertResult(wd(3) :: wd(4) :: wd(6) :: wd(7) :: Nil) {
         buffer.update(add = wd(6) :: wd(7) :: Nil, remove = wd(5) :: Nil)
         buffer.get()
       }
 
-      assertResult(wd(7) :: wd(4) :: Nil)  {
-        buffer.update(add = wd(7) :: Nil, remove = wd(6) :: wd(2) :: Nil)
+      assertResult(wd(3) :: wd(4) :: wd(7) :: wd(7) :: Nil) {
+        buffer.update(add = wd(7) :: Nil, remove = wd(6) :: Nil)
         buffer.get()
       }
     }
   }
 
   private def wd(num: Int) = new WriterData[String](num.toString)
-}
+}