Non-effectful CSV serialization

Author: fernandomora

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

@@ -19,6 +19,7 @@ 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 csvIterable[T: Csv]: Csv[Iterable[T]] = (a: Iterable[T]) => a.foldLeft(Array[String](""))((acc, t) => acc ++ Csv[T].toCSV(t))
   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)

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

@@ -3,26 +3,25 @@ package com.intenthq.action_processor.integrations.serializations.csv
 import java.io._
 import java.nio.charset.StandardCharsets
 
-import cats.effect.IO
-//import de.siegmar.fastcsv.writer.CsvWriter
-import com.opencsv.CSVWriter
-
-import scala.collection.immutable.ArraySeq
+import de.siegmar.fastcsv.writer.CsvWriter
 
 object CsvSerialization {
 
   val lineDelimiter: String = "\n"
-  private val csvWriter = new CSVWriter(new StringWriter())
+  private val csvWriter = {
+    val writer = new CsvWriter()
+    writer.setLineDelimiter(lineDelimiter.toCharArray)
+    writer
+  }
 
-  def serialize[O](o: O)(implicit csv: Csv[O]): IO[Array[Byte]] =
-    IO.delay {
-      unsafeSerialise(ArraySeq.unsafeWrapArray(csv.toCSV(o)))
-    }
+  private def encode[O](o: O)(implicit csv: Csv[O]): Array[String] = csv.toCSV(o)
 
-  def unsafeSerialise(row: Seq[String]): Array[Byte] = {
+  def serialize[O](o: O)(implicit csv: Csv[O]): Array[Byte] = {
     val sw = new StringWriter()
+    val columns = encode(o)
     val appender = csvWriter.append(sw)
-    appender.appendLine(row: _*)
+    appender.appendLine(columns: _*)
+    // Make sure we flush internal appender FastBufferedWriter
     appender.close()
     sw.toString.getBytes(StandardCharsets.UTF_8)
   }

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

@@ -6,7 +6,6 @@ import cats.effect.{IO, Resource}
 import cats.implicits.catsSyntaxApplicativeId
 import de.siegmar.fastcsv.reader.CsvReader
 import fs2.Stream
-import sourcecode.Text.generate
 
 import scala.jdk.CollectionConverters._
 
@@ -23,7 +22,7 @@ abstract class CsvFeed[O] extends Feed[Iterable[String], O] {
   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))
+        .flatMap(parser => Option(parser.nextRow()).map(_.getFields.asScala))
         .getOrElse(collection.mutable.Buffer.empty[String])
         .pure[IO]
     }

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

@@ -6,7 +6,7 @@ import doobie.util.transactor.{Strategy, Transactor}
 
 import scala.util.Properties
 
-abstract class HiveFeed[I, O] extends SqlFeed[I, O]("org.apache.hive.jdbc.HiveDriver") with TimeMeta with JavaLegacyTimeMeta {
+abstract class HiveFeed[I, O] extends SQLFeed[I, O]("org.apache.hive.jdbc.HiveDriver") with TimeMeta with JavaLegacyTimeMeta {
   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")
 }

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

@@ -1,28 +1,25 @@
 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] {
+abstract class SQLFeed[I, O](driver: String) 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)
+  protected def query: 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
+
+  protected def createTransactor: Aux[IO, Unit] = Transactor.fromDriverManager[IO](driver, jdbcUrl)
+
+  override def inputStream: fs2.Stream[IO, I] =
+    query
+      .streamWithChunkSize(chunkSize)
+      .transact[IO](transactor)
 }

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

@@ -29,7 +29,7 @@ abstract class H2Source[O] extends SQLSource[O]("org.h2.Driver") with TimeMeta w
 
 object ExampleCsvFeed extends H2Source[ExampleCsvFeedRow] {
 
-  override def transform(context: SourceContext[IO])(e: ExampleCsvFeedRow): IO[Array[Byte]] = CsvSerialization.serialize[ExampleCsvFeedRow](e)
+  override def transform(context: SourceContext[IO])(e: ExampleCsvFeedRow): IO[Array[Byte]] = IO.pure(CsvSerialization.serialize[ExampleCsvFeedRow](e))
 
   override def query(context: SourceContext[IO]): Query0[ExampleCsvFeedRow] =
     (

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

@@ -24,9 +24,9 @@ object ExampleLocalFileCsvFeed extends LocalFileCsvFeed[AggregatedPerson] {
 
   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()
+  override def serialize(a: AggregatedPerson, counter: Long): Array[Byte] = CsvSerialization.serialize((a, counter))
 }
 
 object ExampleLocalFileCsvFeed2 extends LocalFileCsvFeed[Iterable[String]] with NoAggregate[Iterable[String]] {
-  override def serialize(o: Iterable[String], counter: Long): Array[Byte] = CsvSerialization.serialize(o).unsafeRunSync()
+  override def serialize(o: Iterable[String], counter: Long): Array[Byte] = CsvSerialization.serialize(o)
 }

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

@@ -1,19 +1,20 @@
 package com.intenthq.action_processor.integrations.serializations.csv
 
-import cats.effect.IO
-import cats.implicits._
+import java.nio.charset.StandardCharsets
 import java.time._
+
+import cats.implicits._
 import weaver.{Expectations, SimpleIOSuite}
 
 object CsvSerializationSpec extends SimpleIOSuite {
 
-  private def serialize[T: Csv](t: T): IO[String] = CsvSerialization.serialize(t).map(new String(_))
-  private def checkLine[T: Csv](toSer: T, csv: String): IO[Expectations] =
-    for {
-      result <- serialize(toSer)
-    } yield expect(result == csv + CsvSerialization.lineDelimiter)
+  private def serialize[T: Csv](t: T): String = new String(CsvSerialization.serialize(t), StandardCharsets.UTF_8)
+  private def checkLine[T: Csv](toSer: T, csv: String): Expectations = {
+    val result = serialize(toSer)
+    expect(result == csv + CsvSerialization.lineDelimiter)
+  }
 
-  simpleTest("Serialize a single field case class") {
+  pureTest("Serialize a single field case class") {
     case class Test(a: String)
     checkLine(Test("a"), "a")
   }