IC-40 Supports mapping file with multiple values

Author: fernandomora

build.sbt

@@ -15,18 +15,19 @@ lazy val root = (project in file("."))
     name := "action-processor-integrations",
     Compile / packageDoc / mappings := Seq(),
     Compile / packageSrc / mappings := Seq(),
-    testFrameworks += new TestFramework("weaver.framework.TestFramework"),
+    testFrameworks += new TestFramework("weaver.framework.CatsEffect"),
     scalafmtOnCompile := true,
     libraryDependencies ++= Seq(
-      "co.fs2" %% "fs2-core" % "2.4.4",
-      "co.fs2" %% "fs2-io" % "2.4.4",
+      "co.fs2" %% "fs2-core" % "2.5.3",
+      "co.fs2" %% "fs2-io" % "2.5.3",
       "com.propensive" %% "magnolia" % "0.17.0",
       "de.siegmar" % "fastcsv" % "1.0.3",
       "org.mapdb" % "mapdb" % "3.0.8",
       "org.tpolecat" %% "doobie-core" % "0.9.0",
       "org.tpolecat" %% "doobie-hikari" % "0.9.0",
       "com.google.guava" % "guava" % "30.0-jre",
-      "com.disneystreaming" %% "weaver-framework" % "0.5.0" % "test",
-      "org.tpolecat" %% "doobie-h2" % "0.9.0" % "test"
+      "com.disneystreaming" %% "weaver-core" % "0.6.2" % Test,
+      "com.disneystreaming" %% "weaver-cats" % "0.6.2" % Test,
+      "org.tpolecat" %% "doobie-h2" % "0.9.0" % Test
     )
   )

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

@@ -36,13 +36,15 @@ object Aggregate {
       )
   }
 
-  def aggregateByKey[I, K](feedContext: FeedContext[IO], key: I => K, counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
+  def aggregateByKeys[I, K](feedContext: FeedContext[IO], keys: I => List[K], counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
     sourceStream => {
 
       def put(aggRepository: ConcurrentMap[K, Long], o: I): IO[Unit] =
         IO.delay {
-          val previousCounter = aggRepository.getOrDefault(key(o), 0L)
-          aggRepository.put(key(o), counter(o) + previousCounter)
+          keys(o).foreach { value =>
+            val previousCounter = aggRepository.getOrDefault(value, 0L)
+            aggRepository.put(value, counter(o) + previousCounter)
+          }
         }.void
 
       def streamKeyValue(aggRepository: ConcurrentMap[K, Long]): fs2.Stream[IO, (K, Long)] =
@@ -61,4 +63,7 @@ object Aggregate {
         }.drain ++ streamKeyValue(aggRepository)
       }
     }
+
+  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)
 }

src/main/scala/com/intenthq/action_processor/integrations/feeds/FeedContext.scala

@@ -6,4 +6,4 @@ import com.intenthq.action_processor.integrations.config.MapDbSettings
 import com.intenthq.embeddings.Mapping
 
 case class FeedFilter(date: Option[LocalDate], time: Option[LocalTime])
-case class FeedContext[F[_]](embeddings: Option[Mapping[String, Array[Int], F]], filter: FeedFilter, mapDbSettings: MapDbSettings)
+case class FeedContext[F[_]](embeddings: Option[Mapping[String, List[String], F]], filter: FeedFilter, mapDbSettings: MapDbSettings)

src/test/scala/com/intenthq/action_processor/integrations/FeedAggregatedSpec.scala

@@ -4,43 +4,85 @@ import java.nio.charset.StandardCharsets
 import java.time.LocalDateTime
 
 import cats.effect.IO
+import cats.implicits.catsSyntaxApplicativeId
+import fs2.Pipe
+
 import com.intenthq.action_processor.integrations.aggregations.Aggregate
 import com.intenthq.action_processor.integrations.feeds.{Feed, FeedContext}
 import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
-import fs2.Pipe
+import com.intenthq.embeddings.Mapping
+
 import weaver.SimpleIOSuite
 
 object FeedAggregatedSpec extends SimpleIOSuite {
 
   test("should return a stream of aggregated csv feed rows") {
     val aggregatedFeed = new PersonsAggregatedByScoreFeed(
-      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)
+      Person("Peter", LocalDateTime.parse("2000-01-01T00:00:00"), "music.com", 3),
+      Person("Peter", LocalDateTime.parse("2000-01-01T00:00:00"), "rap.com", 5),
+      Person("Peter", LocalDateTime.parse("2000-01-01T00:00:00"), "rock.com", 7),
+      Person("Gabriela", LocalDateTime.parse("2000-01-01T00:00:00"), "music.com", 2),
+      Person("Gabriela", LocalDateTime.parse("2000-01-01T00:00:00"), "rap.com", 10),
+      Person("Gabriela", LocalDateTime.parse("2000-01-01T00:00:00"), "unknown.com", 10),
+      Person("Jolie", LocalDateTime.parse("2000-01-01T00:00:00"), "rap.com", 1),
+      Person("Jolie", LocalDateTime.parse("2000-01-01T00:00:00"), "rock.com", 7)
     )
 
+    val mapping = new Mapping[String, List[String], IO] {
+      private val map = Map(
+        "music.com" -> List("Music"),
+        "rap.com" -> List("Music", "Rap"),
+        "rock.com" -> List("Music", "Rock")
+      )
+      override def get(key: String): IO[Option[List[String]]] = map.get(key).pure
+    }
+
     val expectedResult: Set[String] = Set(
-      "Peter,2001-01-01T00:00:00,11",
-      "Gabriela,2002-01-01T00:00:00,7",
-      "Jolie,2003-01-01T00:00:00,4"
+      "Peter,2000-01-01T00:00:00,Music,15",
+      "Peter,2000-01-01T00:00:00,Rap,5",
+      "Peter,2000-01-01T00:00:00,Rock,7",
+      "Gabriela,2000-01-01T00:00:00,Music,12",
+      "Gabriela,2000-01-01T00:00:00,Rap,10",
+      "Jolie,2000-01-01T00:00:00,Music,8",
+      "Jolie,2000-01-01T00:00:00,Rock,7",
+      "Jolie,2000-01-01T00:00:00,Rap,1"
     ).map(_ + '\n')
 
     for {
-      feedStreamLinesBytes <- aggregatedFeed.stream(TestDefaults.feedContext).compile.toList
+      feedStreamLinesBytes <- aggregatedFeed.stream(TestDefaults.feedContext.copy(embeddings = Some(mapping))).compile.toList
       feedStreamLines = feedStreamLinesBytes.map(bytes => new String(bytes, StandardCharsets.UTF_8)).toSet
     } yield expect(feedStreamLines == expectedResult)
   }
 }
 
-case class Person(name: String, bornDate: LocalDateTime, score: Int)
-case class AggregatedPerson(name: String, bornDate: LocalDateTime)
+case class Person(name: String, timestamp: LocalDateTime, domain: String, count: Int)
+case class MappedPerson(name: String, timestamp: LocalDateTime, interests: List[String], count: Int)
+case class AggregatedPerson(name: String, timestamp: LocalDateTime, interest: String)
+
+class PersonsAggregatedByScoreFeed(persons: Person*) extends Feed[MappedPerson, AggregatedPerson] {
+  override def inputStream(feedContext: FeedContext[IO]): fs2.Stream[IO, MappedPerson] =
+    fs2
+      .Stream(persons: _*)
+      .through(mapPersons(feedContext))
 
-class PersonsAggregatedByScoreFeed(persons: Person*) extends Feed[Person, AggregatedPerson] {
-  override def inputStream(feedContext: FeedContext[IO]): fs2.Stream[IO, Person] = fs2.Stream(persons: _*).covary[IO]
+  private def mapPersons(feedContext: FeedContext[IO]): Pipe[IO, Person, MappedPerson] = { in =>
+    fs2.Stream.eval(IO.fromOption(feedContext.embeddings)(new RuntimeException("Mapping required"))).flatMap { mappings =>
+      in.evalMap { person =>
+        mappings
+          .get(person.domain)
+          .map(
+            _.map(interests => MappedPerson(person.name, person.timestamp, interests, person.count))
+          )
+      }.unNone
+    }
+  }
 
-  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 transform(feedContext: FeedContext[IO]): Pipe[IO, MappedPerson, (AggregatedPerson, Long)] =
+    Aggregate.aggregateByKeys[MappedPerson, AggregatedPerson](
+      feedContext,
+      person => person.interests.map(AggregatedPerson(person.name, person.timestamp, _)),
+      _.count.toLong
+    )
 
   override def serialize(o: AggregatedPerson, counter: Long): Array[Byte] = CsvSerialization.serialize((o, counter))
 }

src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala

@@ -5,18 +5,19 @@ import java.time.{Instant, LocalDate, LocalTime}
 
 import cats.effect.{Async, Blocker, Bracket, ContextShift, IO, Resource}
 import cats.implicits.toFunctorOps
+
 import com.intenthq.action_processor.integrations.aggregations.NoAggregate
 import com.intenthq.action_processor.integrations.feeds.{FeedContext, SQLFeed}
 import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
+
 import doobie.h2.H2Transactor
 import doobie.implicits.{toConnectionIOOps, toSqlInterpolator}
 import doobie.util.query.Query0
 import doobie.util.transactor.Transactor
 import doobie.util.update.Update
 import doobie.implicits.javatime._
 import weaver.IOSuite
-
-import scala.concurrent.ExecutionContextExecutor
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
 
 object SQLFeedSpec extends IOSuite with SQLFeedSpecResources {
 
@@ -55,7 +56,7 @@ trait SQLFeedSpecResources { self: IOSuite =>
     for {
       blocker <- Blocker[IO]
       // Given a transactor for an exanpleRows fixture populated database
-      transactor <- transactorResource[IO](blocker, ec)
+      transactor <- transactorResource[IO](blocker, ExecutionContext.global)
     } yield transactor
 
   private def transactorResource[F[_]: Async: ContextShift](blocker: Blocker, ec: ExecutionContextExecutor): Resource[F, H2Transactor[F]] = {

src/test/scala/com/intenthq/action_processor/integrations/serializations/csv/CsvSerializationSpec.scala

@@ -20,24 +20,24 @@ object CsvSerializationSpec extends SimpleIOSuite {
     checkLine(Test("a"), "a")
   }
 
-  simpleTest("Serialize several fields case class") {
+  pureTest("Serialize several fields case class") {
     case class Test(a: String, b: String, c: String)
     checkLine(Test("a", "b", "c"), "a,b,c")
   }
 
-  simpleTest("Serialize nested case class") {
+  pureTest("Serialize nested case class") {
     case class Test1(a1: String, b1: Test2, c1: String)
     case class Test2(a2: String, b2: String, c2: String)
     checkLine(Test1("a1", Test2("a2", "b2", "c2"), "c1"), "a1,a2,b2,c2,c1")
   }
 
-  simpleTest("Serialize optional fields case class") {
+  pureTest("Serialize optional fields case class") {
     case class Test(a: String, b: Option[String], c: String)
     checkLine(Test("a", Some("b"), "c"), "a,b,c") |+|
       checkLine(Test("a", None, "c"), "a,,c")
   }
 
-  simpleTest("Serialize primitive types") {
+  pureTest("Serialize primitive types") {
     checkLine("a", "a") |+|
       checkLine(1, "1") |+|
       checkLine(1L, "1") |+|