Fixes CsvFeedSpec

Author: fernandomora

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

@@ -17,12 +17,13 @@ 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()).map(_.getFields.asScala))
+      Option(csvReader.parse(sr).nextRow())
+        .map(_.getFields.asScala)
         .getOrElse(collection.mutable.Buffer.empty[String])
         .pure[IO]
     }
 
   override def inputStream: Stream[IO, Iterable[String]] =
     rows.evalMap(csvParse)
+
 }

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/LocalFileCsvFeed.scala

@@ -18,5 +18,6 @@ trait LocalFileCsvFeed[O] extends CsvFeed[O] {
         .readAll[IO](Paths.get(localFilePath), blocker, 4096)
         .through(text.utf8Decode)
         .through(text.lines)
+        .dropLastIf(_.isEmpty)
     }
 }

src/main/scala/com/intenthq/action_processor/integrationsV2/SqlFeed.scala renamed to src/main/scala/com/intenthq/action_processor/integrationsV2/SQLFeed.scala

@@ -6,7 +6,7 @@ import doobie.util.query.Query0
 import doobie.util.transactor.Transactor
 import doobie.util.transactor.Transactor.Aux
 
-abstract class SqlFeed[I, O](driver: String) extends Feed[I, O] {
+abstract class SQLFeed[I, O](driver: String) extends Feed[I, O] {
 
   protected val jdbcUrl: String
   protected def query: Query0[I]

src/test/resources/persons.csv

@@ -1,4 +1,4 @@
 "Peter"|"Big Street 1"|"5"
 "Gabriela"|"Big Street 2"|"7"
 "Jolie"|"Big Street 3"|"4"
-"Sam"|"Big Street 4"|"6"
+"Peter"|"Big Street 1"|"6"

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

@@ -5,8 +5,9 @@ import weaver.SimpleIOSuite
 object CsvFeedSpec extends SimpleIOSuite {
 
   private val expectedResult: Set[String] = Set(
-    "447722222222,20150620000000,apple.com,1",
-    "447599999999,20150620000000,microsoft.co.uk,3"
+    "Peter,Big Street 1,11",
+    "Gabriela,Big Street 2,7",
+    "Jolie,Big Street 3,4"
   ).map(_ + '\n')
 
   simpleTest("should return a stream of parsed csv feed rows") {

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

@@ -4,7 +4,7 @@ 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, NoAggregate}
+import com.intenthq.action_processor.integrationsV2.{Aggregate, LocalFileCsvFeed}
 import fs2.Pipe
 
 case class AggregatedPerson(name: String, address: String)
@@ -25,8 +25,5 @@ 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))
-}
 
-object ExampleLocalFileCsvFeed2 extends LocalFileCsvFeed[Iterable[String]] with NoAggregate[Iterable[String]] {
-  override def serialize(o: Iterable[String], counter: Long): Array[Byte] = CsvSerialization.serialize(o)
 }