IC-40 Removes Aggregations progress

Author: fernandomora

src/main/scala/com/intenthq/action_processor/integrations/aggregations/Aggregate.scala

@@ -2,27 +2,23 @@ package com.intenthq.action_processor.integrations.aggregations
 
 import java.util.concurrent.ConcurrentMap
 
-import scala.concurrent.duration.{DurationInt, FiniteDuration, NANOSECONDS}
+import scala.jdk.CollectionConverters._
 
-import cats.effect.{Blocker, ContextShift, IO, Resource, Sync, SyncIO, Timer}
+import cats.effect.{Blocker, ContextShift, IO, Resource, SyncIO}
 
 import com.intenthq.action_processor.integrations.config.MapDbSettings
 import com.intenthq.action_processor.integrations.feeds.FeedContext
 import com.intenthq.action_processor.integrations.repositories.MapDBRepository
 
+import org.mapdb.{DataInput2, DataOutput2, HTreeMap, Serializer}
 import org.mapdb.elsa.{ElsaMaker, ElsaSerializer}
 import org.mapdb.serializer.GroupSerializerObjectArray
-import org.mapdb.{DataInput2, DataOutput2, HTreeMap, Serializer}
-import scala.jdk.CollectionConverters._
-
-import cats.implicits._
 
 object Aggregate {
 
   private lazy val blocker = Blocker[SyncIO].allocated.unsafeRunSync()._1
   private val ec = scala.concurrent.ExecutionContext.global
   implicit private val contextShift: ContextShift[IO] = IO.contextShift(ec)
-  implicit private val timer: Timer[IO] = IO.timer(ec)
 
   def noop[I]: fs2.Pipe[IO, I, (I, Long)] = _.map(_ -> 1L)
 
@@ -45,26 +41,30 @@ object Aggregate {
 
   def aggregateByKeys[I, K](feedContext: FeedContext[IO], keys: I => List[K], counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
     sourceStream => {
+
+      // This pipe aggregates all the elemens and returns a single Map as an aggregate repository
       val aggregateInRepository: fs2.Pipe[IO, I, ConcurrentMap[K, Long]] =
         in => {
           fs2.Stream
             .resource[IO, ConcurrentMap[K, Long]](loadAggRepository(feedContext.mapDbSettings)(blocker))
-            .flatTap { aggRepository =>
+            .flatMap { aggRepository =>
               fs2.Stream.eval_(IO.delay(println("Starting aggregation"))) ++
-                in.evalTap { o =>
+                in.evalMapChunk { o =>
                   IO.delay {
                     keys(o).foreach { value =>
                       val previousCounter = aggRepository.getOrDefault(value, 0L)
                       aggRepository.put(value, counter(o) + previousCounter)
                     }
+                    aggRepository
                   }
-                }.through(AggregationsProgress.showAggregationProgress(5.seconds))
-                  .as(1)
-                  .foldMonoid
-                  .evalMap(n => IO.delay(println(s"Finished aggregation of $n rows")))
+                }
+                  // Returns last aggRepository with the counter of elements
+                  .fold((aggRepository, 0L)) { case ((_, previousRows), aggRepository) => (aggRepository, previousRows + 1) }
+                  .evalMapChunk { case (aggRepository, n) => IO.delay(println(s"Finished aggregation of $n rows")).as(aggRepository) }
             }
         }
 
+      // Streams the givens aggregate repository entries
       val streamAggRepository: fs2.Pipe[IO, ConcurrentMap[K, Long], (K, Long)] =
         _.flatMap(aggRepository => fs2.Stream.iterable(aggRepository.asScala))
 
@@ -74,53 +74,3 @@ object Aggregate {
   def aggregateByKey[I, K](feedContext: FeedContext[IO], key: I => K, counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
     aggregateByKeys(feedContext, key.andThen(List(_)), counter)
 }
-
-object AggregationsProgress {
-  def showAggregationProgress[F[_]: Sync: Timer, O](duration: FiniteDuration): fs2.Pipe[F, O, O] = { in =>
-    val startTime = System.nanoTime()
-    var lastTime = System.nanoTime()
-    var lastRow = 0L
-    def formatTime(duration: FiniteDuration): String = {
-      val durationSecs = duration.toSeconds
-      f"${durationSecs / 3600}%d:${(durationSecs % 3600) / 60}%02d:${durationSecs % 60}%02d"
-    }
-    in.through(showProgress(duration) {
-      case (totalRows, o) =>
-        Sync[F].delay {
-          val now = System.nanoTime()
-          val totalTime = FiniteDuration(now - startTime, NANOSECONDS)
-          val partialTime = FiniteDuration(now - lastTime, NANOSECONDS)
-          val partialRows = totalRows - lastRow
-          lastTime = System.nanoTime()
-          lastRow = totalRows
-
-          println(f"\nRow #$totalRows: ${o.toString} ")
-          println(f"Partial time: ${formatTime(partialTime)}. Total time: ${formatTime(totalTime)}")
-          println(
-            f"Partial speed: ${partialRows.toFloat / partialTime.toSeconds}%.2f rows/sec. Total Speed: ${totalRows.toFloat / totalTime.toSeconds}%.2f rows/sec"
-          )
-
-        }
-    })
-  }
-
-  def showProgress[F[_]: Sync: Timer, O](every: FiniteDuration)(output: (Long, O) => F[Unit]): fs2.Pipe[F, O, O] = { source =>
-    val ticks = fs2.Stream.every[F](every)
-    source
-      // Based on zipWithIndex but starting by 1
-      .scanChunks(1L) { (index, c) =>
-        var idx = index
-        val out = c.map { o =>
-          val r = (o, idx)
-          idx += 1
-          r
-        }
-        (idx, out)
-      }
-      .zipWith(ticks)((_, _))
-      .evalMap {
-        case ((v, index), isTick) =>
-          (if (isTick) output(index, v) else Sync[F].unit) >> Sync[F].pure(v)
-      }
-  }
-}

src/main/scala/com/intenthq/action_processor/integrations/config/MapDbSettings.scala

@@ -8,9 +8,9 @@ object MapDbSettings {
 
   val Default: MapDbSettings = MapDbSettings(
     dbPath = Paths.get("/tmp"),
-    startDbSize = 5L * 1024,
-    incSize = 1L * 1024,
-    segments = 16,
+    startDbSize = 512 * 1024 * 1024, // 512MB
+    incSize = 512 * 1024 * 1024, // 512MB
+    segments = 8,
     nodeSize = 128,
     levels = 4
   )