WIP

Author: marc ramirez

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

@@ -0,0 +1,21 @@
+package com.intenthq.hybrid.integrationsV2
+
+import java.nio.charset.StandardCharsets
+
+import cats.effect._
+import com.intenthq.hybrid.integrations.{JavaLegacyTimeMeta, TimeMeta}
+import doobie.util.transactor.{Strategy, Transactor}
+
+import scala.util.Properties
+
+abstract class HiveSource[O] extends Processor with SQLSource[O] with Aggregations[O] with TimeMeta with JavaLegacyTimeMeta {
+
+  override protected val driver: String = "org.apache.hive.jdbc.HiveDriver"
+
+  override val jdbcUrl: String = Properties.envOrElse("HIVE_JDBC_URL", "jdbc:hive2://localhost:10000")
+
+  override protected lazy val transactor: Transactor[IO] = Transactor.strategy.set(createTransactor, Strategy.void)
+
+  override def serialize(o2: (String, Long)): Array[Byte] = s"${o2._1},${o2._2}\n".getBytes(StandardCharsets.UTF_8)
+
+}

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

@@ -0,0 +1,65 @@
+package com.intenthq.hybrid.integrationsV2
+
+import java.util.concurrent.ConcurrentMap
+
+import cats.effect.{IO, Resource}
+import fs2.Stream
+
+import scala.jdk.CollectionConverters.IteratorHasAsScala
+
+/**
+ * Scenarios with Processor
+ *
+ * 1. () => Stream[IO, Array[Byte]]
+ * 2. Stream[IO, I] => Stream[IO, Array[Byte]]
+ * 3. Stream[IO, I] => Stream[IO, O] => Stream[IO, Array[Byte]]
+ * 4. Stream[IO, I] => Stream[IO, O] => Stream[IO, (K,V)] ... Drain ... Stream[IO, (K1,V1)] => Stream[IO, Array[Byte]]
+ */
+
+trait Processor {
+  def stream(processorContext: ProcessorContext): fs2.Stream[IO, Array[Byte]]
+}
+
+trait Source[I] { self: Processor =>
+  protected def sourceStream(processorContext: ProcessorContext): fs2.Stream[IO, I]
+}
+
+trait Sink[O] { self: Processor =>
+  def serialize(o2: O): Array[Byte]
+}
+
+trait Aggregations[I] extends Source[I] with Sink[(String, Long)] { self: Processor =>
+
+  val repository: ConcurrentMap[String, Long]
+
+  def key(a: I): String
+  def value(a: I): Long
+
+  override def serialize(o2: (String, Long)): Array[Byte]
+
+  override def stream(processorContext: ProcessorContext): fs2.Stream[IO, Array[Byte]] =
+    Stream
+      .resource[IO, ConcurrentMap[String, Long]](Resource.liftF(IO(repository)))
+      .flatMap { _ =>
+        sourceStream(processorContext).evalMap { o =>
+          if (key(o).nonEmpty) put(o) else IO.unit
+        }.drain ++ streamKeyValue.evalMap(i => IO.delay(serialize(i)))
+      }
+
+  private def put(o: I): IO[Unit] =
+    IO.delay {
+      val previousCounter = repository.getOrDefault(key(o), 0)
+      repository.put(key(o), previousCounter + value(o))
+    }.void
+
+  private def streamKeyValue: Stream[IO, (String, Long)] =
+    fs2.Stream
+      .fromIterator[IO](
+        repository
+          .entrySet()
+          .iterator()
+          .asScala
+      )
+      .map(e => (e.getKey, e.getValue))
+
+}

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

@@ -0,0 +1,4 @@
+package com.intenthq.hybrid.integrationsV2
+
+case class ProcessorContext()
+object ProcessorContext {}

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

@@ -0,0 +1,37 @@
+package com.intenthq.hybrid.integrationsV2
+
+import java.util.concurrent.ForkJoinPool
+
+import cats.effect.{ContextShift, IO}
+import com.intenthq.hybrid.integrations.DoobieImplicits
+import doobie.util.query.Query0
+import doobie.util.transactor.Transactor
+import doobie.util.transactor.Transactor.Aux
+import fs2._
+
+trait SQLSource[I] extends Processor with Source[I] with DoobieImplicits {
+
+  protected val driver: String
+
+  protected val jdbcUrl: String
+
+  protected def query(processorContext: ProcessorContext): Query0[I]
+
+  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
+
+  override protected def sourceStream(processorContext: ProcessorContext): Stream[IO, I] =
+    query(processorContext)
+      .streamWithChunkSize(chunkSize)
+      .transact[IO](transactor)
+
+}
+
+object SQLSource {
+  val DefaultParallelism: Int = scala.concurrent.ExecutionContext.global.asInstanceOf[ForkJoinPool].getParallelism
+}