Next evolutionary step

Author: Merlin Rabens

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

@@ -4,18 +4,15 @@ import java.io._
 import java.nio.charset.StandardCharsets
 
 import cats.effect.IO
-import de.siegmar.fastcsv.writer.CsvWriter
+//import de.siegmar.fastcsv.writer.CsvWriter
+import com.opencsv.CSVWriter
 
 import scala.collection.immutable.ArraySeq
 
 object CsvSerialization {
 
   val lineDelimiter: String = "\n"
-  private val csvWriter = {
-    val writer = new CsvWriter()
-    writer.setLineDelimiter(lineDelimiter.toCharArray)
-    writer
-  }
+  private val csvWriter = new CSVWriter(new StringWriter())
 
   def serialize[O](o: O)(implicit csv: Csv[O]): IO[Array[Byte]] =
     IO.delay {

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

@@ -10,31 +10,32 @@ import sourcecode.Text.generate
 
 import scala.jdk.CollectionConverters._
 
-abstract class CsvFeed[I <: Product, O] extends Feed[I, O] {
+abstract class CsvFeed[O] extends Feed[Iterable[String], O] {
 
-  protected val csvResource: String
+//  protected val csvResource: String
 
-  private lazy val typeFactory = new ReflectionHelpers.CaseClassFactory[I]
+//  private lazy val typeFactory = new ReflectionHelpers.CaseClassFactory[I]
 
   protected lazy val csvReader: CsvReader = new CsvReader
 
-  protected def rows: Stream[IO, I]
+  protected def rows: Stream[IO, String]
 
   private 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])
+        .getOrElse(collection.mutable.Buffer.empty[String])
         .pure[IO]
     }
 
-  override def inputStream: Stream[IO, I] = rows
+  override def inputStream: Stream[IO, Iterable[String]] =
+    rows.evalMap(csvParse)
 
-  protected def fromString: fs2.Pipe[IO, String, I] =
-    sourceStream => {
-      sourceStream.evalMap { line =>
-        val params = csvParse(line).productIterator.toList
-        IO(typeFactory.buildWith(params))
-      }
-    }
+//  protected def fromString: fs2.Pipe[IO, Iterable[String], I] =
+//    sourceStream => {
+//      sourceStream.evalMap { line =>
+//        val params = .productIterator.toList
+//        IO(typeFactory.buildWith(params))
+//      }
+//    }
 }

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

@@ -11,7 +11,7 @@ trait Feed[I, O] {
   final def stream: fs2.Stream[IO, Array[Byte]] =
     inputStream
       .through(transform)
-      .map { case (a, counter) => serialize(a, counter) }
+      .map((serialize _).tupled)
 }
 
 trait NoAggregate[I] { self: Feed[I, I] =>

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

@@ -7,17 +7,16 @@ import fs2.text
 
 import scala.util.Properties
 
-abstract class LocalFileCsvFeed[I <: Product, O] extends CsvFeed[I, O] {
+trait LocalFileCsvFeed[O] extends CsvFeed[O] {
 
   implicit protected val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
-  override protected val csvResource: String = Properties.envOrElse("CSV_RESOURCE", "/data.csv")
+  protected val localFilePath: String = Properties.envOrElse("CSV_RESOURCE", "/data.csv")
 
-  override protected def rows: fs2.Stream[IO, I] =
+  override protected def rows: fs2.Stream[IO, String] =
     fs2.Stream.resource(Blocker[IO]).flatMap { blocker =>
       fs2.io.file
-        .readAll[IO](Paths.get(csvResource), blocker, 4096)
+        .readAll[IO](Paths.get(localFilePath), blocker, 4096)
         .through(text.utf8Decode)
         .through(text.lines)
-        .through(fromString)
     }
 }

src/test/scala/com/intenthq/action_processor/integrations/ExampleLocalFileCsvFeed.scala

@@ -4,23 +4,29 @@ import java.nio.file.Paths
 
 import cats.effect.IO
 import com.intenthq.action_processor.integrations.serializations.csv.CsvSerialization
-import com.intenthq.action_processor.integrationsV2.{Aggregate, LocalFileCsvFeed}
+import com.intenthq.action_processor.integrationsV2.{Aggregate, LocalFileCsvFeed, NoAggregate}
 import fs2.Pipe
 
-case class Person(name: String, address: String, score: Int) {
-  lazy val aggregateKey: AggregatedPerson = AggregatedPerson(name, address)
-}
-
 case class AggregatedPerson(name: String, address: String)
 
-object ExampleLocalFileCsvFeed extends LocalFileCsvFeed[Person, AggregatedPerson] {
+object ExampleLocalFileCsvFeed extends LocalFileCsvFeed[AggregatedPerson] {
 
-  override protected val csvResource: String = Paths.get(getClass.getResource("/persons.csv").toURI).toAbsolutePath.toString
+  override protected val localFilePath: String = Paths.get(getClass.getResource("/persons.csv").toURI).toAbsolutePath.toString
 
   csvReader.setFieldSeparator('|')
 
-  override def transform: Pipe[IO, Person, (AggregatedPerson, Long)] =
-    Aggregate.aggregateByKey[Person, AggregatedPerson](_.aggregateKey, _.score.toLong)
+  private def key(columns: Iterable[String]) = {
+    val v = columns.toVector
+    AggregatedPerson(v(0), v(1))
+  }
+
+  private def counter(columns: Iterable[String]) = columns.lastOption.flatMap(v => scala.util.Try(v.toLong).toOption).getOrElse(0L)
+
+  override def transform: Pipe[IO, Iterable[String], (AggregatedPerson, Long)] = Aggregate.aggregateByKey[Iterable[String], AggregatedPerson](key, counter)
 
   override def serialize(a: AggregatedPerson, counter: Long): Array[Byte] = CsvSerialization.serialize((a, counter)).unsafeRunSync()
 }
+
+object ExampleLocalFileCsvFeed2 extends LocalFileCsvFeed[Iterable[String]] with NoAggregate[Iterable[String]] {
+  override def serialize(o: Iterable[String], counter: Long): Array[Byte] = CsvSerialization.serialize(o).unsafeRunSync()
+}