Add feed context to transform method

Merlin Rabens · merlinrabens · commit c8019544a085 · 2020-11-03T15:28:18.000+01:00
Signed-off-by: Merlin Rabens &lt;merlin.rabens@intenthq.com&gt;
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/aggregations/Aggregate.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/aggregations/Aggregate.scala
@@ -1,20 +1,40 @@
 package com.intenthq.action_processor.integrationsV2.aggregations
 
-import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
+import java.util.concurrent.ConcurrentMap
 
 import cats.effect.{IO, Resource}
+import com.intenthq.action_processor.integrationsV2.config.MapDbSettings
+import com.intenthq.action_processor.integrationsV2.feeds.FeedContext
+import com.intenthq.action_processor.integrationsV2.repositories.MapDBRepository
+import org.mapdb.elsa.{ElsaMaker, ElsaSerializer}
+import org.mapdb.serializer.GroupSerializerObjectArray
+import org.mapdb.{DataInput2, DataOutput2, HTreeMap, Serializer}
 
 import scala.jdk.CollectionConverters._
 
 object Aggregate {
 
   def noop[I]: fs2.Pipe[IO, I, (I, Long)] = _.map(_ -> 1L)
 
-  private def loadAggRepository[K]: Resource[IO, ConcurrentMap[K, Long]] =
-    // TODO: Insert MapDB here ...
-    Resource.liftF(IO.pure(new ConcurrentHashMap[K, Long]()))
+  private def loadAggRepository[K](mapDbSettings: MapDbSettings): Resource[IO, HTreeMap[K, Long]] = {
 
-  def aggregateByKey[I, K](key: I => K, counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
+    val serializer: Serializer[K] = new GroupSerializerObjectArray[K] {
+      val elsaSerializer: ElsaSerializer = new ElsaMaker().make
+      override def deserialize(input: DataInput2, available: Int): K = elsaSerializer.deserialize[K](input)
+      override def serialize(out: DataOutput2, value: K): Unit = elsaSerializer.serialize(out, value)
+    }
+
+    MapDBRepository
+      .load(mapDbSettings)
+      .map(db =>
+        db.hashMap("stuff", serializer, Serializer.LONG)
+          .layout(mapDbSettings.segments, mapDbSettings.nodeSize, mapDbSettings.levels)
+          .createOrOpen()
+          .asInstanceOf[HTreeMap[K, Long]]
+      )
+  }
+
+  def aggregateByKey[I, K](feedContext: FeedContext[IO], key: I => K, counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
     sourceStream => {
 
       def put(aggRepository: ConcurrentMap[K, Long], o: I): IO[Unit] =
@@ -33,7 +53,7 @@ object Aggregate {
           )
           .map(e => (e.getKey, e.getValue))
 
-      fs2.Stream.resource[IO, ConcurrentMap[K, Long]](loadAggRepository).flatMap { aggRepository =>
+      fs2.Stream.resource[IO, ConcurrentMap[K, Long]](loadAggRepository(feedContext.mapDbSettings)).flatMap { aggRepository =>
         sourceStream.evalMap { i =>
           put(aggRepository, i)
         }.drain ++ streamKeyValue(aggRepository)
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/aggregations/NoAggregate.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/aggregations/NoAggregate.scala
@@ -1,9 +1,11 @@
 package com.intenthq.action_processor.integrationsV2.aggregations
 
 import cats.effect.IO
-import com.intenthq.action_processor.integrationsV2.feeds.Feed
+import com.intenthq.action_processor.integrationsV2.feeds.{Feed, FeedContext}
+
+import scala.annotation.unused
 
 trait NoAggregate[I] {
   self: Feed[I, I] =>
-  override def transform: fs2.Pipe[IO, I, (I, Long)] = Aggregate.noop
+  override def transform(@unused feedContext: FeedContext[IO]): fs2.Pipe[IO, I, (I, Long)] = Aggregate.noop
 }
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/config/MapDBSettings.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/config/MapDBSettings.scala
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/config/MapDbSettings.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/config/MapDbSettings.scala
@@ -0,0 +1,5 @@
+package com.intenthq.action_processor.integrationsV2.config
+
+import java.nio.file.Path
+
+case class MapDbSettings(dbPath: Path, startDbSize: Long, incSize: Long, segments: Int, nodeSize: Int, levels: Int)
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/Feed.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/Feed.scala
@@ -5,11 +5,11 @@ import cats.effect.IO
 trait Feed[I, O] {
 
   def inputStream(feedContext: FeedContext[IO]): fs2.Stream[IO, I]
-  def transform: fs2.Pipe[IO, I, (O, Long)]
+  def transform(feedContext: FeedContext[IO]): fs2.Pipe[IO, I, (O, Long)]
   def serialize(o: O, counter: Long): Array[Byte]
 
   final def stream(feedContext: FeedContext[IO]): fs2.Stream[IO, Array[Byte]] =
     inputStream(feedContext)
-      .through(transform)
+      .through(transform(feedContext))
       .map((serialize _).tupled)
 }
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/FeedContext.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/FeedContext.scala
@@ -2,17 +2,8 @@ package com.intenthq.action_processor.integrationsV2.feeds
 
 import java.time.{LocalDate, LocalTime}
 
-import com.intenthq.action_processor.integrationsV2.config.MapDBSettings
+import com.intenthq.action_processor.integrationsV2.config.MapDbSettings
 import com.intenthq.embeddings.Mapping
 
 case class FeedFilter(date: Option[LocalDate], time: Option[LocalTime])
-
-object FeedFilter {
-  val empty: FeedFilter = FeedFilter(None, None)
-}
-
-case class FeedContext[F[_]](embeddings: Option[Mapping[String, Array[Int], F]], filter: FeedFilter, mapDbSettings: MapDBSettings)
-
-object FeedContext {
-  def empty[F[_]]: FeedContext[F] = FeedContext[F](None, FeedFilter.empty, MapDBSettings.Default)
-}
+case class FeedContext[F[_]](embeddings: Option[Mapping[String, Array[Int], F]], filter: FeedFilter, mapDbSettings: MapDbSettings)
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/repositories/MapDBRepository.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/repositories/MapDBRepository.scala
@@ -5,20 +5,20 @@ import java.nio.file.Paths
 import java.time.LocalDateTime
 
 import cats.effect.{IO, Resource}
-import com.intenthq.action_processor.integrationsV2.config.MapDBSettings
+import com.intenthq.action_processor.integrationsV2.config.MapDbSettings
 import org.mapdb.{DB, DBMaker}
 
 object MapDBRepository {
-  def load(mapDBSettings: MapDBSettings): Resource[IO, DB] = {
+  def load(mapDBSettings: MapDbSettings): Resource[IO, DB] = {
     val dbCreated = for {
       now <- IO.delay(LocalDateTime.now())
       dbFile <- IO.delay(new File(Paths.get(mapDBSettings.dbPath.toString, s"dbb-${now.toLocalDate}-${now.toLocalTime}").toUri))
       _ <- IO.delay(dbFile.deleteOnExit())
       createDb <- IO.delay {
         DBMaker
           .fileDB(dbFile.getAbsolutePath)
-          .allocateStartSize(mapDBSettings.startDbSize * 1024 * 1024 * 1024)
-          .allocateIncrement(mapDBSettings.incSize * 1024 * 1024 * 1024)
+          .allocateStartSize(mapDBSettings.startDbSize)
+          .allocateIncrement(mapDBSettings.incSize)
           .fileMmapEnableIfSupported()
           .fileMmapPreclearDisable()
           .closeOnJvmShutdown()
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/CSVFeedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/CSVFeedSpec.scala
@@ -8,7 +8,7 @@ import cats.effect.{IO, Resource}
 import cats.implicits._
 import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
 import com.intenthq.action_processor.integrationsV2.aggregations.NoAggregate
-import com.intenthq.action_processor.integrationsV2.feeds.{FeedContext, LocalFileCSVFeed}
+import com.intenthq.action_processor.integrationsV2.feeds.LocalFileCSVFeed
 import weaver.IOSuite
 
 object CSVFeedSpec extends IOSuite with CSVFeedSpecResources {
@@ -29,7 +29,7 @@ object CSVFeedSpec extends IOSuite with CSVFeedSpecResources {
     ).map(_ + '\n')
 
     for {
-      feedStreamLinesBytes <- resources.csvFeed.stream(FeedContext.empty).compile.toList
+      feedStreamLinesBytes <- resources.csvFeed.stream(Defaults.feedContext).compile.toList
       feedStreamLines = feedStreamLinesBytes.map(bytes => new String(bytes, StandardCharsets.UTF_8)).toSet
     } yield expect(feedStreamLines == expectedResult)
   }
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/Defaults.scala b/src/test/scala/com/intenthq/action_processor/integrations/Defaults.scala
@@ -0,0 +1,22 @@
+package com.intenthq.action_processor.integrations
+
+import java.nio.file.Paths
+
+import com.intenthq.action_processor.integrationsV2.config.MapDbSettings
+import com.intenthq.action_processor.integrationsV2.feeds.{FeedContext, FeedFilter}
+
+object Defaults {
+
+  val mapDbSettings: MapDbSettings = MapDbSettings(
+    dbPath = Paths.get("/tmp"),
+    startDbSize = 1L * 1024 * 1024,
+    incSize = 1L,
+    segments = 16,
+    nodeSize = 128,
+    levels = 4
+  )
+
+  val feedFilter: FeedFilter = FeedFilter(None, None)
+
+  def feedContext[F[_]]: FeedContext[F] = FeedContext[F](None, feedFilter, mapDbSettings)
+}
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/FeedAggregatedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/FeedAggregatedSpec.scala
@@ -1,6 +1,7 @@
 package com.intenthq.action_processor.integrations
 
 import java.nio.charset.StandardCharsets
+import java.time.LocalDateTime
 
 import cats.effect.IO
 import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
@@ -13,33 +14,33 @@ object FeedAggregatedSpec extends SimpleIOSuite {
 
   test("should return a stream of aggregated csv feed rows") {
     val aggregatedFeed = new PersonsAggregatedByScoreFeed(
-      Person("Peter", "Big Street 1", 5),
-      Person("Gabriela", "Big Street 2", 7),
-      Person("Jolie", "Big Street 3", 4),
-      Person("Peter", "Big Street 1", 6)
+      Person("Peter", LocalDateTime.parse("2001-01-01T00:00:00"), 5),
+      Person("Gabriela", LocalDateTime.parse("2002-01-01T00:00:00"), 7),
+      Person("Jolie", LocalDateTime.parse("2003-01-01T00:00:00"), 4),
+      Person("Peter", LocalDateTime.parse("2001-01-01T00:00:00"), 6)
     )
 
     val expectedResult: Set[String] = Set(
-      "Peter,Big Street 1,11",
-      "Gabriela,Big Street 2,7",
-      "Jolie,Big Street 3,4"
+      "Peter,2001-01-01T00:00:00,11",
+      "Gabriela,2002-01-01T00:00:00,7",
+      "Jolie,2003-01-01T00:00:00,4"
     ).map(_ + '\n')
 
     for {
-      feedStreamLinesBytes <- aggregatedFeed.stream(FeedContext.empty).compile.toList
+      feedStreamLinesBytes <- aggregatedFeed.stream(Defaults.feedContext).compile.toList
       feedStreamLines = feedStreamLinesBytes.map(bytes => new String(bytes, StandardCharsets.UTF_8)).toSet
     } yield expect(feedStreamLines == expectedResult)
   }
 }
 
-case class Person(name: String, address: String, score: Int)
-case class AggregatedPerson(name: String, address: String)
+case class Person(name: String, bornDate: LocalDateTime, score: Int)
+case class AggregatedPerson(name: String, bornDate: LocalDateTime)
 
 class PersonsAggregatedByScoreFeed(persons: Person*) extends Feed[Person, AggregatedPerson] {
   override def inputStream(feedContext: FeedContext[IO]): fs2.Stream[IO, Person] = fs2.Stream(persons: _*).covary[IO]
 
-  override def transform: Pipe[IO, Person, (AggregatedPerson, Long)] =
-    Aggregate.aggregateByKey[Person, AggregatedPerson](person => AggregatedPerson(person.name, person.address), _.score.toLong)
+  override def transform(feedContext: FeedContext[IO]): Pipe[IO, Person, (AggregatedPerson, Long)] =
+    Aggregate.aggregateByKey[Person, AggregatedPerson](feedContext, person => AggregatedPerson(person.name, person.bornDate), _.score.toLong)
 
   override def serialize(o: AggregatedPerson, counter: Long): Array[Byte] = CsvSerialization.serialize((o, counter))
 }
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala
@@ -38,7 +38,7 @@ object SQLFeedSpec extends IOSuite with SQLFeedSpecResources {
 
   test("should return a stream of parsed ExampleFeedRow") { _ =>
     for {
-      feedStreamLinesBytes <- ExampleCsvFeed.stream(FeedContext.empty).compile.toList
+      feedStreamLinesBytes <- ExampleCsvFeed.stream(Defaults.feedContext).compile.toList
       feedStreamLines = feedStreamLinesBytes.map(new String(_))
       expectedOutput = exampleRows.map(CsvSerialization.serialize[ExampleCsvFeedRow]).map(new String(_))
     } yield expect(feedStreamLines == expectedOutput)
