New approach (discussed with FM and MR)

Author: Merlin Rabens

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

@@ -1,18 +1,16 @@
-package com.intenthq.hybrid.integrationsV2
+package com.intenthq.action_processor.integrationsV2
 
 import cats.effect._
-import com.intenthq.hybrid.integrations.{JavaLegacyTimeMeta, TimeMeta}
+import com.intenthq.action_processor.integrations.{JavaLegacyTimeMeta, TimeMeta}
 import doobie.util.transactor.{Strategy, Transactor}
 
 import scala.util.Properties
 
-abstract class HiveSource[O] extends SQLSource[O] with Aggregations[O] with TimeMeta with JavaLegacyTimeMeta {
+abstract class HiveSource[I] extends SQLSource[I] with Sink 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]
 }

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

@@ -1,8 +1,12 @@
-package com.intenthq.hybrid.integrationsV2
+package com.intenthq.action_processor.integrationsV2
 
-import java.util.concurrent.ConcurrentMap
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
 
 import cats.effect.{IO, Resource}
+import com.intenthq.action_processor.integrations.SourceContext
+import com.intenthq.hybrid.integrationsV2.ProcessorContext
+import doobie.util.query
 import fs2.Stream
 
 import scala.jdk.CollectionConverters.IteratorHasAsScala
@@ -16,34 +20,66 @@ import scala.jdk.CollectionConverters.IteratorHasAsScala
  * 4. Stream[IO, I] => Stream[IO, O] => Stream[IO, (K,V)] ... Drain ... Stream[IO, (K1,V1)] => Stream[IO, Array[Byte]]
  */
 
+object Aggregate {
+  def apply[I, A, AC](processorContext: ProcessorContext, key: I => Option[A], counter: I => Long): fs2.Pipe[IO, I, A] = sourceStream => {
+    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(serializeAggregatedKV(i)))
+      }
+
+  }
+}
+trait Feed[I, A] {
+  def inputStream(processorContext: ProcessorContext): fs2.Stream[IO, I]
+  def transform(processorContext: ProcessorContext): fs2.Pipe[IO, I, (A, Long)] = Aggregate,apply(k = )
+  def serialize(a: (A, Long)): Array[Byte]
+
+  final def stream(processorContext: ProcessorContext) =
+    inputStream(processorContext)
+      .through(transform(processorContext))
+      .map(serialize)
+}
 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 Source[I] { self: Processor =>
+//  protected def sourceStream(processorContext: ProcessorContext): fs2.Stream[IO, I]
+//  protected def pipe: fs2.Pipe[IO, I, Array[Byte]] = _.map(serializeSource)
+//  def serializeSource(i:I): Array[Byte]
+//}
 
-trait Sink[O] { self: Processor =>
-  def serialize(o2: O): Array[Byte]
-}
+//trait Sink[I, O] { self: Processor =>
+//  def serialize(o2: I): O
+//}
 
-trait Aggregations[I] extends Source[I] with Sink[(String, Long)] { self: Processor =>
+trait Aggregations[I, NT] extends Processor {
 
-  val repository: ConcurrentMap[String, Long]
+  lazy val repository: ConcurrentMap[String, Long] = new ConcurrentHashMap[String, Long]()
 
   def key(a: I): String
   def value(a: I): Long
+  def serializeAggregatedKV(a: I):
+
 
-  override def serialize(o2: (String, Long)): Array[Byte]
+  def whatEveer(processorContext: ProcessorContext): Stream[IO, (String, Long)] = 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(serializeAggregatedKV(i)))
+    }
 
   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)))
+        }.drain ++ streamKeyValue.evalMap(i => IO.delay(serializeAggregatedKV(i)))
       }
 
   private def put(o: I): IO[Unit] =
@@ -63,3 +99,62 @@ trait Aggregations[I] extends Source[I] with Sink[(String, Long)] { self: Proces
       .map(e => (e.getKey, e.getValue))
 
 }
+
+//object Main {
+//  def main(args: Array[String]): Unit = {
+//    class A extends Processor with Aggregations[Int] {
+//      override protected def sourceStream(processorContext: ProcessorContext): Stream[IO, Int] = fs2.Stream(1, 2, 3, 4)
+////      override val repository: ConcurrentMap[String, Long] = new ConcurrentHashMap[String, Long]()
+//      override def key(a: Int): String = a.toString
+//      override def value(a: Int): Long = 1L
+//      override def serializeAggregatedKV(o2: (String, Long)): Array[Byte] = s"${o2._1},${o2._2}".getBytes(StandardCharsets.UTF_8)
+//    }
+//  }
+//}
+
+object Main2 {
+  def main(args: Array[String]): Unit = {
+    class A extends Processor {
+      override protected def stream(processorContext: ProcessorContext): Stream[IO, Array[Byte]] =
+        Stream(1, 2, 3, 4, 5).map(_.toString.getBytes)
+    }
+  }
+}
+
+object Main3 {
+  def main(args: Array[String]): Unit = {
+    class A extends Processor with HiveSource[Int] {
+      override protected val driver: String = "Mysql"
+      override protected val jdbcUrl: String = "jdbc://blah"
+
+      override protected def query(processorContext: ProcessorContext): query.Query0[Int] = ???
+
+      override def serializeRow(o2: Int): Array[Byte] = o2.toString.getBytes
+    }
+  }
+}
+
+S
+I
+O
+Source[S, I, O] {
+  1. sourceStream[I]
+  2. aggegation: noop MapDB[I, Long]
+  3. serialize(I, PartialRow)
+}
+
+object Main4 {
+  def main(args: Array[String]): Unit = {
+    class A extends Processor with SQLSource[Int] {
+      override protected val driver: String = "Mysql"
+      override protected val jdbcUrl: String = "jdbc://blah"
+
+      override protected def query(processorContext: ProcessorContext): query.Query0[Int] = ???
+
+      override def serializeRow(o2: Int): Array[Byte] = null
+
+      override protected def stream(processorContext: ProcessorContext): Stream[IO, Array[Byte]] =
+
+    }
+  }
+}

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

@@ -1,13 +1,14 @@
-package com.intenthq.hybrid.integrationsV2
+package com.intenthq.action_processor.integrationsV2
 
 import java.util.concurrent.ForkJoinPool
 
 import cats.effect.{ContextShift, IO}
+import com.intenthq.action_processor.integrations.DoobieImplicits
 import com.intenthq.hybrid.integrations.DoobieImplicits
+import com.intenthq.hybrid.integrationsV2.ProcessorContext
 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 {
 
@@ -17,6 +18,8 @@ trait SQLSource[I] extends Processor with Source[I] with DoobieImplicits {
 
   protected def query(processorContext: ProcessorContext): Query0[I]
 
+  def serializeRow(row: I): Array[Byte]
+
   implicit private val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
 
   protected def createTransactor: Aux[IO, Unit] = Transactor.fromDriverManager[IO](driver, jdbcUrl)
@@ -25,11 +28,11 @@ trait SQLSource[I] extends Processor with Source[I] with DoobieImplicits {
 
   protected val chunkSize: Int = doobie.util.query.DefaultChunkSize
 
-  override protected def sourceStream(processorContext: ProcessorContext): Stream[IO, I] =
+  override def stream(processorContext: ProcessorContext): fs2.Stream[IO, Array[Byte]] =
     query(processorContext)
       .streamWithChunkSize(chunkSize)
       .transact[IO](transactor)
-
+      .map(serializeRow)
 }
 
 object SQLSource {