#2309 Add csv local file source

Signed-off-by: Merlin Rabens <merlin.rabens@intenthq.com>

Author: Merlin Rabens

build.sbt

@@ -17,11 +17,13 @@ lazy val root = (project in file("."))
     scalafmtOnCompile := true,
     libraryDependencies ++= Seq(
       "co.fs2" %% "fs2-core" % "2.4.4",
+      "co.fs2" %% "fs2-io" % "2.4.4",
       "com.propensive" %% "magnolia" % "0.17.0",
       "de.siegmar" % "fastcsv" % "1.0.3",
       "org.mapdb" % "mapdb" % "3.0.8",
       "org.tpolecat" %% "doobie-core" % "0.9.0",
       "org.tpolecat" %% "doobie-hikari" % "0.9.0",
+      "com.google.guava" % "guava" % "30.0-jre",
       "com.disneystreaming" %% "weaver-framework" % "0.5.0" % "test",
       "com.disneystreaming" %% "weaver-scalacheck" % "0.4.3" % "test",
       "org.tpolecat" %% "doobie-h2" % "0.9.0" % "test"

src/main/scala/com/intenthq/hybrid/integrations/CsvSource.scala

@@ -1,3 +1,46 @@
 package com.intenthq.hybrid.integrations
 
-class CsvSource {}
+import java.io.StringReader
+import java.nio.charset.StandardCharsets
+
+import cats.effect.{IO, Resource}
+import cats.implicits.catsSyntaxApplicativeId
+import de.siegmar.fastcsv.reader.CsvReader
+import fs2.Stream
+
+import scala.jdk.CollectionConverters._
+
+abstract class CsvSource extends Source[String] {
+  protected val csvResource: String
+  protected def lines(): Stream[IO, String]
+  override def stream(context: SourceContext[IO]): Stream[IO, Array[Byte]] = lines().map(_.getBytes)
+
+  protected lazy val csvReader: CsvReader = new CsvReader
+
+  protected 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])
+        .pure[IO]
+    }
+
+}
+
+abstract class CsvSourceAgg extends CsvSource {
+
+  protected def aggregate(context: SourceContext[IO])(line: String): IO[Unit]
+
+  override def stream(context: SourceContext[IO]): Stream[IO, Array[Byte]] = {
+    lazy val readMapAndEmit: Stream[IO, Array[Byte]] =
+      Stream
+        .fromIterator[IO](
+          context.map
+            .entrySet()
+            .iterator()
+            .asScala
+        )
+        .map(e => (e.getKey + ',' + e.getValue + '\n').getBytes(StandardCharsets.UTF_8))
+    super.stream(context).filter(_.nonEmpty).evalMap(line => aggregate(context)(new String(line))).drain ++ readMapAndEmit
+  }
+}

src/main/scala/com/intenthq/hybrid/integrations/DomainStemmer.scala

@@ -0,0 +1,10 @@
+package com.intenthq.hybrid.integrations
+
+import com.google.common.net.InternetDomainName
+
+import scala.util.Try
+
+object DomainStemmer {
+  // using publicsuffix.org
+  def apply(url: String): String = Try(InternetDomainName.from(url).topPrivateDomain().toString).getOrElse("")
+}

src/main/scala/com/intenthq/hybrid/integrations/LocalFileCsvSource.scala

@@ -0,0 +1,20 @@
+package com.intenthq.hybrid.integrations
+import java.nio.file.Paths
+
+import cats.effect.{Blocker, ContextShift, IO}
+import fs2.{text, Stream}
+
+import scala.util.Properties
+
+abstract class LocalFileCsvSource extends CsvSourceAgg {
+  implicit protected val contextShift: ContextShift[IO] = IO.contextShift(scala.concurrent.ExecutionContext.global)
+  override protected val csvResource: String = Properties.envOrElse("CSV_RESOURCE", "/data.csv")
+  override protected def lines(): Stream[IO, String] =
+    Stream.resource(Blocker[IO]).flatMap { blocker =>
+      fs2.io.file
+        .readAll[IO](Paths.get(csvResource), blocker, 4096)
+        .through(text.utf8Decode)
+        .through(text.lines)
+        .dropLastIf(_.isEmpty)
+    }
+}

src/test/resources/example.csv

@@ -0,0 +1,4 @@
+"447599999999"|"20150620000000"|"173806"|"www2.microsoft.co.uk"
+"447722222222"|"20150620000000"|"155322"|"apple.com"
+"447599999999"|"20150620000000"|"173806"|"m.microsoft.co.uk"
+"447599999999"|"20150620000000"|"173806"|"www.microsoft.co.uk"

src/test/resources/rows.csv

@@ -1,3 +1,18 @@
 package com.intenthq.hybrid.integrations
 
-object CsvFeedSpec {}
+import weaver.SimpleIOSuite
+
+object CsvFeedSpec extends SimpleIOSuite {
+
+  private val expectedResult: Set[String] = Set(
+    "447722222222,20150620000000,apple.com,1",
+    "447599999999,20150620000000,microsoft.co.uk,3"
+  ).map(_ + '\n')
+
+  simpleTest("should return a stream of parsed O2 weblog feed rows") {
+    for {
+      rowsBytes <- ExampleLocalFileCsvFeed.stream(SourceContext.empty).compile.toList
+      rows = rowsBytes.map(new String(_)).toSet
+    } yield expect(rows == expectedResult)
+  }
+}

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

@@ -0,0 +1,26 @@
+package com.intenthq.hybrid.integrations
+
+import java.nio.file.Paths
+
+import cats.effect.IO
+
+object ExampleLocalFileCsvFeed extends LocalFileCsvSource {
+
+  override protected val csvResource: String = Paths.get(getClass.getResource("/example.csv").toURI).toAbsolutePath.toString
+
+  csvReader.setFieldSeparator('|')
+
+  override def aggregate(context: SourceContext[IO])(line: String): IO[Unit] = {
+    def addToContextMap(csvTokens: Iterable[String]) =
+      csvTokens.lastOption.map(DomainStemmer.apply).fold(IO.unit) { domain =>
+        val key = (csvTokens.dropRight(2) ++ Seq(domain)).mkString(",")
+        val previousCounter = context.map.getOrDefault(key, 0)
+        IO.delay(context.map.put(key, previousCounter + 1)).void
+      }
+
+    for {
+      csvTokens <- csvParse(line)
+      _ <- addToContextMap(csvTokens)
+    } yield ()
+  }
+}