Replace SourceContext by FeedContext

Merlin Rabens · merlinrabens · commit cb0d07d4d62c · 2020-11-03T15:28:18.000+01:00
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
@@ -11,6 +11,7 @@ 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]()))
 
   def aggregateByKey[I, K](key: I => K, counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/CSVFeed.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/CSVFeed.scala
@@ -4,7 +4,6 @@ import java.io.StringReader
 
 import cats.effect.{IO, Resource}
 import cats.implicits.catsSyntaxApplicativeId
-import com.intenthq.action_processor.integrations.SourceContext
 import de.siegmar.fastcsv.reader.CsvReader
 import fs2.Stream
 
@@ -24,7 +23,6 @@ trait CSVFeed[O] extends Feed[Iterable[String], O] {
         .pure[IO]
     }
 
-  override def inputStream(sourceContext: SourceContext[IO]): Stream[IO, Iterable[String]] =
+  override def inputStream(feedContext: FeedContext[IO]): Stream[IO, Iterable[String]] =
     rows.evalMap(csvParse)
-
 }
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
@@ -1,16 +1,15 @@
 package com.intenthq.action_processor.integrationsV2.feeds
 
 import cats.effect.IO
-import com.intenthq.action_processor.integrations.SourceContext
 
 trait Feed[I, O] {
 
-  def inputStream(sourceContext: SourceContext[IO]): fs2.Stream[IO, I]
+  def inputStream(feedContext: FeedContext[IO]): fs2.Stream[IO, I]
   def transform: fs2.Pipe[IO, I, (O, Long)]
   def serialize(o: O, counter: Long): Array[Byte]
 
-  final def stream(sourceContext: SourceContext[IO]): fs2.Stream[IO, Array[Byte]] =
-    inputStream(sourceContext)
+  final def stream(feedContext: FeedContext[IO]): fs2.Stream[IO, Array[Byte]] =
+    inputStream(feedContext)
       .through(transform)
       .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
@@ -0,0 +1,18 @@
+package com.intenthq.action_processor.integrationsV2.feeds
+
+import java.time.{LocalDate, LocalTime}
+
+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)
+}
diff --git a/src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/SQLFeed.scala b/src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/SQLFeed.scala
@@ -1,7 +1,7 @@
 package com.intenthq.action_processor.integrationsV2.feeds
 
 import cats.effect.{ContextShift, IO}
-import com.intenthq.action_processor.integrations.{DoobieImplicits, SourceContext}
+import com.intenthq.action_processor.integrations.DoobieImplicits
 import doobie.util.query.Query0
 import doobie.util.transactor.Transactor
 import doobie.util.transactor.Transactor.Aux
@@ -10,7 +10,7 @@ trait SQLFeed[I, O] extends Feed[I, O] with DoobieImplicits {
 
   protected val driver: String
   protected val jdbcUrl: String
-  protected def query(sourceContext: SourceContext[IO]): Query0[I]
+  protected def query(sourceContext: FeedContext[IO]): Query0[I]
 
   implicit private val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
 
@@ -19,8 +19,8 @@ trait SQLFeed[I, O] extends Feed[I, O] with DoobieImplicits {
 
   protected def createTransactor: Aux[IO, Unit] = Transactor.fromDriverManager[IO](driver, jdbcUrl)
 
-  override def inputStream(sourceContext: SourceContext[IO]): fs2.Stream[IO, I] =
-    query(sourceContext)
+  override def inputStream(feedContext: FeedContext[IO]): fs2.Stream[IO, I] =
+    query(feedContext)
       .streamWithChunkSize(chunkSize)
       .transact[IO](transactor)
 }
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.LocalFileCSVFeed
+import com.intenthq.action_processor.integrationsV2.feeds.{FeedContext, 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(SourceContext.empty).compile.toList
+      feedStreamLinesBytes <- resources.csvFeed.stream(FeedContext.empty).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/FeedAggregatedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/FeedAggregatedSpec.scala
@@ -5,7 +5,7 @@ import java.nio.charset.StandardCharsets
 import cats.effect.IO
 import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
 import com.intenthq.action_processor.integrationsV2.aggregations.Aggregate
-import com.intenthq.action_processor.integrationsV2.feeds.Feed
+import com.intenthq.action_processor.integrationsV2.feeds.{Feed, FeedContext}
 import fs2.Pipe
 import weaver.SimpleIOSuite
 
@@ -26,7 +26,7 @@ object FeedAggregatedSpec extends SimpleIOSuite {
     ).map(_ + '\n')
 
     for {
-      feedStreamLinesBytes <- aggregatedFeed.stream(SourceContext.empty).compile.toList
+      feedStreamLinesBytes <- aggregatedFeed.stream(FeedContext.empty).compile.toList
       feedStreamLines = feedStreamLinesBytes.map(bytes => new String(bytes, StandardCharsets.UTF_8)).toSet
     } yield expect(feedStreamLines == expectedResult)
   }
@@ -36,7 +36,7 @@ case class Person(name: String, address: String, score: Int)
 case class AggregatedPerson(name: String, address: String)
 
 class PersonsAggregatedByScoreFeed(persons: Person*) extends Feed[Person, AggregatedPerson] {
-  override def inputStream(sourceContext: SourceContext[IO]): fs2.Stream[IO, Person] = fs2.Stream(persons: _*).covary[IO]
+  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)
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala
@@ -7,7 +7,7 @@ import cats.effect.{Async, Blocker, Bracket, ContextShift, 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.SQLFeed
+import com.intenthq.action_processor.integrationsV2.feeds.{FeedContext, SQLFeed}
 import doobie.h2.H2Transactor
 import doobie.implicits._
 import doobie.implicits.javatime._
@@ -38,7 +38,7 @@ object SQLFeedSpec extends IOSuite with SQLFeedSpecResources {
 
   test("should return a stream of parsed ExampleFeedRow") { _ =>
     for {
-      feedStreamLinesBytes <- ExampleCsvFeed.stream(SourceContext.empty).compile.toList
+      feedStreamLinesBytes <- ExampleCsvFeed.stream(FeedContext.empty).compile.toList
       feedStreamLines = feedStreamLinesBytes.map(new String(_))
       expectedOutput = exampleRows.map(CsvSerialization.serialize[ExampleCsvFeedRow]).map(new String(_))
     } yield expect(feedStreamLines == expectedOutput)
@@ -112,7 +112,7 @@ abstract class H2Source[I, O] extends SQLFeed[I, O] with TimeMeta with JavaTimeM
 
 object ExampleCsvFeed extends H2Source[ExampleCsvFeedRow, ExampleCsvFeedRow] with NoAggregate[ExampleCsvFeedRow] {
 
-  override def query(context: SourceContext[IO]): Query0[ExampleCsvFeedRow] =
+  override def query(context: FeedContext[IO]): Query0[ExampleCsvFeedRow] =
     (
       sql"""SELECT integer, bigint, float, double, decimal, numeric, bit, varchar, date, time, timestamp
            |FROM example""".stripMargin ++

Original file line number	Diff line number	Diff line change
`@@ -4,7 +4,6 @@ import java.io.StringReader`
`4`	`4`
`5`	`5`	`import cats.effect.{IO, Resource}`
`6`	`6`	`import cats.implicits.catsSyntaxApplicativeId`
`7`		`-import com.intenthq.action_processor.integrations.SourceContext`
`8`	`7`	`import de.siegmar.fastcsv.reader.CsvReader`
`9`	`8`	`import fs2.Stream`
`10`	`9`
`@@ -24,7 +23,6 @@ trait CSVFeed[O] extends Feed[Iterable[String], O] {`
`24`	`23`	`.pure[IO]`
`25`	`24`	`}`
`26`	`25`
`27`		`- override def inputStream(sourceContext: SourceContext[IO]): Stream[IO, Iterable[String]] =`
	`26`	`+ override def inputStream(feedContext: FeedContext[IO]): Stream[IO, Iterable[String]] =`
`28`	`27`	`rows.evalMap(csvParse)`
`29`		`-`
`30`	`28`	`}`