IC-445 Refactor action processor feeds

fernandomora · fernandomora · commit 9cdd5ac9ffdf · 2021-11-22T23:05:31.000+01:00
diff --git a/build.sbt b/build.sbt
@@ -27,6 +27,7 @@ lazy val root = (project in file("."))
       "org.mapdb" % "mapdb" % "3.0.8",
       "org.tpolecat" %% "doobie-core" % "1.0.0-RC1",
       "org.tpolecat" %% "doobie-hikari" % "1.0.0-RC1",
+      "org.tpolecat" %% "doobie-postgres" % "1.0.0-RC1",
       "com.disneystreaming" %% "weaver-cats" % "0.7.7" % Test,
       "com.disneystreaming" %% "weaver-core" % "0.7.7" % Test,
       "org.tpolecat" %% "doobie-h2" % "1.0.0-RC1" % Test
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/DoobieImplicits.scala b/src/main/scala/com/intenthq/action_processor/integrations/DoobieImplicits.scala
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/feeds/CSVFeed.scala b/src/main/scala/com/intenthq/action_processor/integrations/feeds/CSVFeed.scala
@@ -9,7 +9,7 @@ import fs2.Stream
 
 import scala.jdk.CollectionConverters._
 
-trait CSVFeed[O] extends Feed[Iterable[String], O] {
+trait ParseCSVInput[O] extends Feed[Iterable[String], O] {
 
   protected lazy val csvReader: CsvReader = new CsvReader
 
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/feeds/HiveFeed.scala b/src/main/scala/com/intenthq/action_processor/integrations/feeds/HiveFeed.scala
@@ -1,12 +1,12 @@
 package com.intenthq.action_processor.integrations.feeds
 
 import cats.effect.IO
-import com.intenthq.action_processor.integrations.{JavaLegacyTimeMeta, TimeMeta}
+import com.intenthq.action_processor.integrations.implicits.DoobieImplicits
 import doobie.util.transactor.{Strategy, Transactor}
 
 import scala.util.Properties
 
-trait HiveFeed[I, O] extends SQLFeed[I, O] with TimeMeta with JavaLegacyTimeMeta {
+trait HiveFeed[I, O] extends SQLFeed[I, O] with DoobieImplicits.javatime.legacy {
   override protected val driver: String = "org.apache.hive.jdbc.HiveDriver"
   override protected lazy val transactor: Transactor[IO] = Transactor.strategy.set(createTransactor, Strategy.void)
   override protected val jdbcUrl: String = Properties.envOrElse("HIVE_JDBC_URL", "jdbc:hive2://localhost:10000")
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/feeds/LocalFileCSVFeed.scala b/src/main/scala/com/intenthq/action_processor/integrations/feeds/LocalFileCSVFeed.scala
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/feeds/LocalFileFeed.scala b/src/main/scala/com/intenthq/action_processor/integrations/feeds/LocalFileFeed.scala
@@ -0,0 +1,20 @@
+package com.intenthq.action_processor.integrations.feeds
+
+import cats.effect.IO
+import fs2.Pipe
+import fs2.io.file.{Files, Flags, Path}
+
+import java.nio.file.Paths
+import scala.annotation.unused
+import scala.util.Properties
+
+trait LocalFileFeed[I, O] extends Feed[I, O] {
+
+  protected val localFilePath: String = Properties.envOrElse("CSV_RESOURCE", "/data.csv")
+  protected val parseInput: Pipe[IO, Byte, I]
+
+  override def inputStream(@unused feedContext: FeedContext[IO]): fs2.Stream[IO, I] =
+    Files[IO]
+      .readAll(Path.fromNioPath(Paths.get(localFilePath)), 64 * 1024, Flags.Read)
+      .through(parseInput)
+}
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/feeds/PostgresFeed.scala b/src/main/scala/com/intenthq/action_processor/integrations/feeds/PostgresFeed.scala
@@ -0,0 +1,10 @@
+package com.intenthq.action_processor.integrations.feeds
+
+import com.intenthq.action_processor.integrations.implicits.PgDoobieImplicits
+
+import scala.util.Properties
+
+trait PostgresFeed[I, O] extends SQLFeed[I, O] with PgDoobieImplicits {
+  override val driver: String = "org.postgresql.Driver"
+  override protected val jdbcUrl: String = Properties.envOrElse("POSTGRESQL_JDBC_URL", "jdbc:postgresql://localhost:5432/database")
+}
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/feeds/SQLFeed.scala b/src/main/scala/com/intenthq/action_processor/integrations/feeds/SQLFeed.scala
@@ -1,7 +1,7 @@
 package com.intenthq.action_processor.integrations.feeds
 
 import cats.effect.IO
-import com.intenthq.action_processor.integrations.DoobieImplicits
+import com.intenthq.action_processor.integrations.implicits.DoobieImplicits
 import doobie.util.query.Query0
 import doobie.util.transactor.Transactor
 import doobie.util.transactor.Transactor.Aux
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/implicits/DoobieImplicits.scala b/src/main/scala/com/intenthq/action_processor/integrations/implicits/DoobieImplicits.scala
@@ -0,0 +1,39 @@
+package com.intenthq.action_processor.integrations.implicits
+
+import doobie.Meta
+
+trait DoobieImplicits
+// https://github.com/tpolecat/doobie/blob/series/0.13.x/modules/core/src/main/scala/doobie/package.scala#L21
+    extends doobie.free.Instances
+    with doobie.syntax.AllSyntax
+    with doobie.util.meta.MetaConstructors
+    with doobie.util.meta.SqlMetaInstances
+    with CatsImplicits {
+  val fragments: doobie.util.fragments.type = doobie.util.fragments
+}
+
+object DoobieImplicits {
+  object javatime {
+    trait legacy
+        extends doobie.util.meta.LegacyInstantMetaInstance
+        with doobie.util.meta.LegacyLocalDateMetaInstance
+        with LegacyLocalTimeMetaInstance
+        with LegacyLocalDateTimeMetaInstance
+
+    trait drivernative extends doobie.util.meta.MetaConstructors with doobie.util.meta.TimeMetaInstances
+  }
+}
+
+trait LegacyLocalTimeMetaInstance {
+  implicit val JavaLocalTimeMeta: Meta[java.time.LocalTime] =
+    doobie.implicits.javasql.TimeMeta.imap(_.toLocalTime)(java.sql.Time.valueOf)
+}
+
+trait LegacyLocalDateTimeMetaInstance {
+  implicit val JavaTimeLocalDateTimeMeta: Meta[java.time.LocalDateTime] =
+    doobie.implicits.javasql.TimestampMeta.imap(_.toLocalDateTime)(java.sql.Timestamp.valueOf)
+}
+trait CatsImplicits
+// https://github.com/typelevel/cats/blob/main/core/src/main/scala/cats/implicits.scala#L3
+    extends cats.syntax.AllSyntax
+    with cats.instances.AllInstances
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/implicits/PgDoobieImplicits.scala b/src/main/scala/com/intenthq/action_processor/integrations/implicits/PgDoobieImplicits.scala
@@ -0,0 +1,10 @@
+package com.intenthq.action_processor.integrations.implicits
+
+trait PgDoobieImplicits
+// https://github.com/tpolecat/doobie/blob/series/0.13.x/modules/postgres/src/main/scala/doobie/postgres/package.scala
+    extends doobie.postgres.Instances
+    with doobie.postgres.free.Instances
+    with doobie.postgres.JavaTimeInstances
+    with doobie.postgres.syntax.ToPostgresMonadErrorOps
+    with doobie.postgres.syntax.ToFragmentOps
+    with doobie.postgres.syntax.ToPostgresExplainOps
diff --git a/src/main/scala/com/intenthq/action_processor/integrations/serializations/csv/ParseCSVInput.scala b/src/main/scala/com/intenthq/action_processor/integrations/serializations/csv/ParseCSVInput.scala
@@ -0,0 +1,33 @@
+package com.intenthq.action_processor.integrations.serializations.csv
+
+import cats.effect.{Resource, Sync}
+import cats.implicits.catsSyntaxApplicativeId
+import de.siegmar.fastcsv.reader.CsvReader
+import fs2.Pipe
+
+import java.io.StringReader
+import scala.jdk.CollectionConverters._
+
+object ParseCSVInput {
+  private def csvParse[F[_]: Sync](csvReader: CsvReader)(line: String): F[Iterable[String]] =
+    Resource.fromAutoCloseable(Sync[F].delay(new StringReader(line))).use { sr =>
+      Option(csvReader.parse(sr).nextRow())
+        .map(_.getFields.asScala)
+        .getOrElse(Iterable.empty[String])
+        .pure[F]
+    }
+
+  def parseInput[F[_]: Sync](fieldSeparator: Char = ','): Pipe[F, Byte, Iterable[String]] = {
+    val csvReader: CsvReader = {
+      val csvReader = new CsvReader
+      csvReader.setFieldSeparator(fieldSeparator)
+      csvReader
+    }
+
+    _.through(fs2.text.utf8.decode)
+      .through(fs2.text.lines)
+      .dropLastIf(_.isEmpty)
+      .evalMap(csvParse[F](csvReader))
+  }
+
+}
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/CSVFeedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/CSVFeedSpec.scala
@@ -2,8 +2,9 @@ package com.intenthq.action_processor.integrations
 
 import cats.effect.{IO, Resource}
 import com.intenthq.action_processor.integrations.aggregations.NoAggregate
-import com.intenthq.action_processor.integrations.feeds.LocalFileCSVFeed
-import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
+import com.intenthq.action_processor.integrations.feeds.LocalFileFeed
+import com.intenthq.action_processor.integrations.serializations.csv.{CsvSerialization, ParseCSVInput}
+import fs2.Pipe
 import weaver.IOSuite
 
 import java.io.{File, FileWriter}
@@ -36,7 +37,7 @@ object CSVFeedSpec extends IOSuite with CSVFeedSpecResources {
 
 trait CSVFeedSpecResources { self: IOSuite =>
 
-  case class Resources(csvFeed: ExampleLocalFileCSVFeed)
+  case class Resources(csvFeed: ExampleLocalFileSource)
   override type Res = Resources
 
   protected val csvFeedContent: String
@@ -47,7 +48,7 @@ trait CSVFeedSpecResources { self: IOSuite =>
       csvFeed <- csvFeed(csvFeedContent)
     } yield Resources(csvFeed)
 
-  def csvFeed(content: String): Resource[IO, ExampleLocalFileCSVFeed] = {
+  def csvFeed(content: String): Resource[IO, ExampleLocalFileSource] = {
     def createFileWriter(file: File) = Resource.fromAutoCloseable(IO.delay(new FileWriter(file)))
     def createTmpFile = {
       def createTmpFile = IO.delay(Files.createTempFile(null, null).toFile).flatTap(f => IO.delay(f.deleteOnExit()))
@@ -57,13 +58,13 @@ trait CSVFeedSpecResources { self: IOSuite =>
     createTmpFile
       .evalTap(createFileWriter(_).use(fw => IO.delay(fw.write(content))))
       .map(_.getAbsolutePath)
-      .map(new ExampleLocalFileCSVFeed(_))
+      .map(new ExampleLocalFileSource(_))
   }
 }
 
-class ExampleLocalFileCSVFeed(override val localFilePath: String) extends LocalFileCSVFeed[Iterable[String]] with NoAggregate[Iterable[String]] {
+class ExampleLocalFileSource(override val localFilePath: String) extends LocalFileFeed[Iterable[String], Iterable[String]] with NoAggregate[Iterable[String]] {
 
-  csvReader.setFieldSeparator('|')
+  protected val parseInput: Pipe[IO, Byte, Iterable[String]] = ParseCSVInput.parseInput[IO]('|')
 
   override def serialize(o: Iterable[String], counter: Long): Array[Byte] = CsvSerialization.columnsAsCsv(o)
 }
diff --git a/src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala b/src/test/scala/com/intenthq/action_processor/integrations/SQLFeedSpec.scala
@@ -1,20 +1,20 @@
 package com.intenthq.action_processor.integrations
 
-import java.time.temporal.ChronoUnit
-import java.time.{Instant, LocalDate, LocalTime}
 import cats.effect.{Async, IO, MonadCancel, 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.implicits.DoobieImplicits
 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.javatimedrivernative._
 import weaver.IOSuite
 
+import java.time.temporal.ChronoUnit
+import java.time.{Instant, LocalDate, LocalTime}
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor}
 
 object SQLFeedSpec extends IOSuite with SQLFeedSpecResources {
@@ -72,6 +72,10 @@ trait SQLFeedSpecResources { self: IOSuite =>
            |  time TIME NOT NULL,
            |  timestamp TIMESTAMP NOT NULL
            |)""".stripMargin.update.run.void
+
+    import doobie.implicits.javatimedrivernative._
+    locally(JavaTimeLocalDateMeta)
+
     val insertRows: Update[ExampleCsvFeedRow] = Update[ExampleCsvFeedRow](
       """INSERT INTO example(integer, bigint, float, double, decimal, numeric, bit, varchar, date, time, timestamp)
         |VALUES             (?,       ?,      ?,     ?,      ?,       ?,       ?,   ?,       ?,    ?,    ?        )""".stripMargin
@@ -99,12 +103,12 @@ case class ExampleCsvFeedRow(integer: Int,
                              timestamp: Instant
 )
 
-abstract class H2Source[I, O] extends SQLFeed[I, O] with TimeMeta with JavaTimeMeta {
+abstract class H2Feed[I, O] extends SQLFeed[I, O] with DoobieImplicits.javatime.drivernative {
   override protected val driver: String = "org.h2.Driver"
   override protected val jdbcUrl: String = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1;USER=sa;PASSWORD="
 }
 
-object ExampleCsvFeed extends H2Source[ExampleCsvFeedRow, ExampleCsvFeedRow] with NoAggregate[ExampleCsvFeedRow] {
+object ExampleCsvFeed extends H2Feed[ExampleCsvFeedRow, ExampleCsvFeedRow] with NoAggregate[ExampleCsvFeedRow] {
 
   override def query(context: FeedContext[IO]): Query0[ExampleCsvFeedRow] =
     (
