Cleans tests

Author: fernandomora

build.sbt

@@ -8,6 +8,8 @@ ThisBuild / licenses := Seq(("MIT", url("http://opensource.org/licenses/MIT")))
 
 ThisBuild / scalaVersion := "2.13.3"
 
+addCompilerPlugin("org.typelevel" % "kind-projector" % "0.11.0" cross CrossVersion.full)
+
 lazy val root = (project in file("."))
   .settings(
     name := "action-processor-integrations",

src/main/scala/com/intenthq/action_processor/integrationsV2/Feed.scala

@@ -1,10 +1,10 @@
-package com.intenthq.action_processor.integrationsV2
+package com.intenthq.action_processor.integrationsV2.aggregations
 
 import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
 
 import cats.effect.{IO, Resource}
 
-import scala.jdk.CollectionConverters.IteratorHasAsScala
+import scala.jdk.CollectionConverters._
 
 object Aggregate {
 

src/main/scala/com/intenthq/action_processor/integrationsV2/Aggregate.scala renamed to src/main/scala/com/intenthq/action_processor/integrationsV2/aggregations/Aggregate.scala

@@ -0,0 +1,9 @@
+package com.intenthq.action_processor.integrationsV2.aggregations
+
+import cats.effect.IO
+import com.intenthq.action_processor.integrationsV2.feeds.Feed
+
+trait NoAggregate[I] {
+  self: Feed[I, I] =>
+  override def transform: fs2.Pipe[IO, I, (I, Long)] = Aggregate.noop
+}

src/main/scala/com/intenthq/action_processor/integrationsV2/aggregations/NoAggregate.scala

@@ -1,15 +1,16 @@
-package com.intenthq.action_processor.integrationsV2
+package com.intenthq.action_processor.integrationsV2.feeds
 
 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
 
 import scala.jdk.CollectionConverters._
 
-abstract class CsvFeed[O] extends Feed[Iterable[String], O] {
+trait CSVFeed[O] extends Feed[Iterable[String], O] {
 
   protected lazy val csvReader: CsvReader = new CsvReader
 
@@ -23,7 +24,7 @@ abstract class CsvFeed[O] extends Feed[Iterable[String], O] {
         .pure[IO]
     }
 
-  override def inputStream: Stream[IO, Iterable[String]] =
+  override def inputStream(sourceContext: SourceContext[IO]): Stream[IO, Iterable[String]] =
     rows.evalMap(csvParse)
 
 }

src/main/scala/com/intenthq/action_processor/integrationsV2/CsvFeed.scala renamed to src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/CSVFeed.scala

@@ -0,0 +1,16 @@
+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 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)
+      .through(transform)
+      .map((serialize _).tupled)
+}

src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/Feed.scala

@@ -1,4 +1,4 @@
-package com.intenthq.action_processor.integrationsV2
+package com.intenthq.action_processor.integrationsV2.feeds
 
 import cats.effect.IO
 import com.intenthq.action_processor.integrations.{JavaLegacyTimeMeta, TimeMeta}

src/main/scala/com/intenthq/action_processor/integrationsV2/HiveFeed.scala renamed to src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/HiveFeed.scala

@@ -1,4 +1,4 @@
-package com.intenthq.action_processor.integrationsV2
+package com.intenthq.action_processor.integrationsV2.feeds
 
 import java.nio.file.Paths
 
@@ -7,7 +7,7 @@ import fs2.text
 
 import scala.util.Properties
 
-trait LocalFileCsvFeed[O] extends CsvFeed[O] {
+trait LocalFileCSVFeed[O] extends CSVFeed[O] {
 
   implicit protected val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
   protected val localFilePath: String = Properties.envOrElse("CSV_RESOURCE", "/data.csv")

src/main/scala/com/intenthq/action_processor/integrationsV2/LocalFileCsvFeed.scala renamed to src/main/scala/com/intenthq/action_processor/integrationsV2/feeds/LocalFileCSVFeed.scala

@@ -1,15 +1,15 @@
-package com.intenthq.action_processor.integrationsV2
+package com.intenthq.action_processor.integrationsV2.feeds
 
 import cats.effect.{ContextShift, IO}
-import doobie.implicits.toDoobieStreamOps
+import com.intenthq.action_processor.integrations.{DoobieImplicits, SourceContext}
 import doobie.util.query.Query0
 import doobie.util.transactor.Transactor
 import doobie.util.transactor.Transactor.Aux
 
-abstract class SQLFeed[I, O](driver: String) extends Feed[I, O] {
+abstract class SQLFeed[I, O](driver: String) extends Feed[I, O] with DoobieImplicits {
 
   protected val jdbcUrl: String
-  protected def query: Query0[I]
+  protected def query(sourceContext: SourceContext[IO]): Query0[I]
 
   implicit private val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
 
@@ -18,8 +18,8 @@ abstract class SQLFeed[I, O](driver: String) extends Feed[I, O] {
 
   protected def createTransactor: Aux[IO, Unit] = Transactor.fromDriverManager[IO](driver, jdbcUrl)
 
-  override def inputStream: fs2.Stream[IO, I] =
-    query
+  override def inputStream(sourceContext: SourceContext[IO]): fs2.Stream[IO, I] =
+    query(sourceContext)
       .streamWithChunkSize(chunkSize)
       .transact[IO](transactor)
 }