Next iteration

Author: Merlin Rabens

src/main/scala/com/intenthq/action_processor/integrations/serializations/csv/Csv.scala

@@ -19,7 +19,6 @@ object Csv {
   def dispatch[A](ctx: SealedTrait[Csv, A]): Csv[A] = (a: A) => ctx.dispatch(a)(sub => sub.typeclass.toCSV(sub.cast(a)))
 
   implicit def csvOpt[T: Csv]: Csv[Option[T]] = (a: Option[T]) => a.fold(Array[String](""))(Csv[T].toCSV)
-  implicit def deriveCsv[A]: Csv[A] = macro Magnolia.gen[A]
   implicit val csvStr: Csv[String] = (a: String) => Array(a)
   implicit val csvInt: Csv[Int] = (a: Int) => Array(a.toString)
   implicit val csvLong: Csv[Long] = (a: Long) => Array(a.toString)
@@ -31,4 +30,6 @@ object Csv {
   implicit val csvLocalTime: Csv[LocalTime] = (a: LocalTime) => Array(a.toString)
   implicit val csvInstant: Csv[Instant] = (a: Instant) => Array(a.toString)
   implicit val csvLocalDateTime: Csv[LocalDateTime] = (a: LocalDateTime) => Array(a.toString)
+
+  implicit def deriveCsv[A]: Csv[A] = macro Magnolia.gen[A]
 }

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

@@ -0,0 +1,31 @@
+package com.intenthq.action_processor.integrationsV2
+
+import java.io.StringReader
+
+import cats.effect.{IO, Resource}
+import cats.implicits.catsSyntaxApplicativeId
+import com.intenthq.action_processor.integrations.SourceContext
+
+import scala.jdk.CollectionConverters._
+import de.siegmar.fastcsv.reader.CsvReader
+import fs2.Stream
+
+abstract class CsvFeed extends Feed[String, String] {
+  protected val csvResource: String
+
+  protected lazy val csvReader: CsvReader = new CsvReader
+
+  protected def csvParse(line: String): IO[Iterable[String]] =
+    Resource.fromAutoCloseable(IO.delay(new StringReader(line))).use { sr =>
+      Option(csvReader.parse(sr))
+        .flatMap(parser => Option(parser.nextRow().getFields.asScala))
+        .getOrElse(Iterable.empty[String])
+        .pure[IO]
+    }
+
+  override def inputStream(feedContext: FeedContext): Stream[IO, String]
+}
+
+new LocalFileCsvFeed(){
+override csvResource = "file.csv"
+}

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

@@ -4,6 +4,7 @@ import java.nio.charset.StandardCharsets
 import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
 
 import cats.effect.{ContextShift, IO, Resource}
+import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
 import doobie.implicits.{toDoobieStreamOps, toSqlInterpolator}
 import doobie.util.query.Query0
 import doobie.util.transactor.Transactor
@@ -23,27 +24,31 @@ import scala.jdk.CollectionConverters.IteratorHasAsScala
 
 object Aggregate {
 
-  def apply[I, K]( /*feedContext: FeedContext,*/ key: I => K, counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
+  def noop[I]: fs2.Pipe[IO, I, (I, Long)] = _.map(_ -> 1L)
+
+  private def loadAggRepository[K]: Resource[IO, ConcurrentMap[K, Long]] =
+    Resource.pure(new ConcurrentHashMap[K, Long]())
+
+  def aggregateByKey[I, K]( /*feedContext: FeedContext,*/ key: I => K, counter: I => Long): fs2.Pipe[IO, I, (K, Long)] =
     sourceStream => {
-      val repository: ConcurrentMap[K, Long] = new ConcurrentHashMap[K, Long]()
 
-      def put(o: I): IO[Unit] =
+      def put(aggRepository: ConcurrentMap[K, Long], o: I): IO[Unit] =
         IO.delay {
-          val previousCounter = repository.getOrDefault(key(o), 0L)
-          repository.put(key(o), counter(o) + previousCounter)
+          val previousCounter = aggRepository.getOrDefault(key(o), 0L)
+          aggRepository.put(key(o), counter(o) + previousCounter)
         }.void
 
-      def streamKeyValue: fs2.Stream[IO, (K, Long)] =
+      def streamKeyValue(aggRepository: ConcurrentMap[K, Long]): fs2.Stream[IO, (K, Long)] =
         fs2.Stream
           .fromIterator[IO](
-            repository
+            aggRepository
               .entrySet()
               .iterator()
               .asScala
           )
           .map(e => (e.getKey, e.getValue))
 
-      fs2.Stream.resource[IO, ConcurrentMap[K, Long]](Resource.liftF(IO.delay(repository))).flatMap { _ =>
+      fs2.Stream.resource[IO, ConcurrentMap[K, Long]](loadAggRepository).flatMap { aggRepository =>
         sourceStream.evalMap { i =>
           put(i)
         }.drain ++ streamKeyValue
@@ -53,65 +58,42 @@ object Aggregate {
 
 trait Feed[I, A] {
   def inputStream(feedContext: FeedContext): fs2.Stream[IO, I]
-  def transform(feedContext: FeedContext): fs2.Pipe[IO, I, A]
-  def serialize(a: A): Array[Byte]
+  def transform(feedContext: FeedContext): fs2.Pipe[IO, I, (A, Long)]
+  def serialize(a: A, counter: Long): Array[Byte]
 
   final def stream(processorContext: FeedContext): fs2.Stream[IO, Array[Byte]] =
     inputStream(processorContext)
       .through(transform(processorContext))
-      .map(serialize)
-}
-
-abstract class SQLFeed[I, O] extends Feed[I, O] {
-  protected val jdbcUrl: String
-
-  protected val driver: String
-
-  protected def query(feedContext: FeedContext): Query0[I]
-
-  override def inputStream(feedContext: FeedContext): fs2.Stream[IO, I] =
-    query(feedContext)
-      .streamWithChunkSize(chunkSize)
-      .transact[IO](transactor)
-
-  implicit private val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
-
-  protected def createTransactor: Aux[IO, Unit] = Transactor.fromDriverManager[IO](driver, jdbcUrl)
-
-  protected lazy val transactor: Transactor[IO] = createTransactor
-
-  protected val chunkSize: Int = doobie.util.query.DefaultChunkSize
+      .map { case (a, counter) => serialize(a, counter) }
 }
 
-abstract class Hive[I, O] extends SQLFeed[I, O] {
-
-  override protected val jdbcUrl: String = ""
-
-  override protected val driver: String = ""
-
+trait NoAggregate[I] { self: Feed[I, I] =>
+  override def transform(feedContext: FeedContext): fs2.Pipe[IO, I, (I, Long)] = Aggregate.noop
 }
 
 object Main {
   def main(args: Array[String]): Unit = {
 
-    class NoAggCase extends Hive[Int, String] {
+    case class Person(name: String, address: String, score: Int) {
+      lazy val aggregateKey = new AggregatedPerson(name, address)
+    }
+    case class AggregatedPerson(name: String, address: String)
 
-      override protected def query(feedContext: FeedContext): Query0[Int] = sql"1".query[Int]
+    class PersonFeed extends HiveFeed[Person, Person] with NoAggregate[Person] {
 
-      override def transform(feedContext: FeedContext): Pipe[IO, Int, String] = s => s.map(_.toString)
+      override protected def query(feedContext: FeedContext): Query0[Person] = sql"SELECT 'Nic Cage', 9000".query[Person]
 
-      override def serialize(a: String): Array[Byte] = a.getBytes(StandardCharsets.UTF_8)
+      override def serialize(a: Person, counter: Long): Array[Byte] = CsvSerialization.serialize(a).unsafeRunSync()
     }
 
-    class AggCase extends Hive[Int, (String, Long)] {
+    class PersonsAggregatedByScoreFeed extends HiveFeed[Person, AggregatedPerson] {
 
-      override protected def query(feedContext: FeedContext): Query0[Int] = sql"1".query[Int]
+      override protected def query(feedContext: FeedContext): Query0[Person] = sql"SELECT 'Nic Cage', 9000".query[Person]
 
-      override def transform(feedContext: FeedContext): Pipe[IO, Int, (String, Long)] = Aggregate.apply(_.toString, _ => 1L)
+      override def transform(feedContext: FeedContext): Pipe[IO, Person, (AggregatedPerson, Long)] =
+        Aggregate.aggregateByKey[Person, AggregatedPerson](_.aggregateKey, _.score)
 
-      override def serialize(a: (String, Long)): Array[Byte] = a._1.getBytes(StandardCharsets.UTF_8)
+      override def serialize(a: AggregatedPerson, counter: Long): Array[Byte] = CsvSerialization.serialize((a, counter)).unsafeRunSync()
     }
-
-    new AggCase().stream(new FeedContext()).compile.drain.unsafeRunSync()
   }
 }

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

@@ -0,0 +1,9 @@
+package com.intenthq.action_processor.integrationsV2
+
+import com.intenthq.action_processor.integrations.{JavaLegacyTimeMeta, TimeMeta}
+
+import scala.util.Properties
+
+abstract class HiveFeed[I, O] extends SQLFeed[I, O]("org.apache.hive.jdbc.HiveDriver") with TimeMeta with JavaLegacyTimeMeta {
+  override protected val jdbcUrl: String = Properties.envOrElse("HIVE_JDBC_URL", "jdbc:hive2://localhost:10000")
+}

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

@@ -0,0 +1,28 @@
+package com.intenthq.action_processor.integrationsV2
+
+import cats.effect.{ContextShift, IO}
+import com.intenthq.action_processor.integrations.SQLSource
+import doobie.implicits.toDoobieStreamOps
+import doobie.util.query.Query0
+import doobie.util.transactor.Transactor
+import doobie.util.transactor.Transactor.Aux
+
+abstract class SQLFeed[I, O](driver: String, parallelism: Int = SQLSource.DefaultParallelism) extends Feed[I, O] {
+
+  protected val jdbcUrl: String
+
+  protected def query(feedContext: FeedContext): Query0[I]
+
+  override def inputStream(feedContext: FeedContext): fs2.Stream[IO, I] =
+    query(feedContext)
+      .streamWithChunkSize(chunkSize)
+      .transact[IO](transactor)
+
+  implicit private val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
+
+  protected def createTransactor: Aux[IO, Unit] = Transactor.fromDriverManager[IO](driver, jdbcUrl)
+
+  protected lazy val transactor: Transactor[IO] = createTransactor
+
+  protected val chunkSize: Int = doobie.util.query.DefaultChunkSize
+}