Stream FS2: add (close #346)

.github/workflows/test.yml

@@ -78,8 +78,10 @@ jobs:
       run: sbt "project kinesis" docker:publish
     - name: Build and publish Stream Kafka Docker image
       run: sbt "project kafka" docker:publish
-    - name: Build and publish Stream NSQ Docker images
+    - name: Build and publish Stream NSQ Docker image
       run: sbt "project nsq" docker:publish
+    - name: Build and publish Stream NH Docker image
+      run: sbt "project fs2" docker:publish
 
   deploy_sce:
     needs: test

build.sbt

@@ -19,7 +19,7 @@
 lazy val root = project.in(file("."))
   .settings(name := "enrich")
   .settings(BuildSettings.basicSettings)
-  .aggregate(common, beam, stream, kinesis, kafka, nsq, stdin)
+  .aggregate(common, beam, stream, kinesis, kafka, nsq, stdin, fs2)
 
 lazy val common = project
   .in(file("modules/common"))
@@ -183,3 +183,53 @@ lazy val beam =
     .enablePlugins(JavaAppPackaging, DockerPlugin, BuildInfoPlugin)
 
 Global / onChangedBuildSource := ReloadOnSourceChanges
+
+lazy val fs2 = project
+  .in(file("modules/fs2"))
+  .dependsOn(common)
+  .settings(BuildSettings.basicSettings)
+  .settings(BuildSettings.formatting)
+  .settings(BuildSettings.scoverageSettings)
+  .settings(BuildSettings.addExampleConfToTestCp)
+  .settings(BuildSettings.sbtAssemblySettings)
+  .settings(
+    name := "fs2-enrich",
+    description := "High-performance streaming Snowplow Enrich job built on top of functional streams",
+    buildInfoKeys := Seq[BuildInfoKey](organization, name, version, description),
+    buildInfoPackage := "com.snowplowanalytics.snowplow.enrich.fs2.generated",
+    packageName in Docker := "snowplow/fs2-enrich",
+  )
+  .settings(parallelExecution in Test := false)
+  .settings(
+    libraryDependencies ++= Seq(
+      Dependencies.Libraries.decline,
+      Dependencies.Libraries.fs2PubSub,
+      Dependencies.Libraries.circeExtras,
+      Dependencies.Libraries.circeLiteral,
+      Dependencies.Libraries.circeConfig,
+      Dependencies.Libraries.catsEffect,
+      Dependencies.Libraries.fs2,
+      Dependencies.Libraries.fs2Io,
+      Dependencies.Libraries.slf4j,
+      Dependencies.Libraries.sentry,
+      Dependencies.Libraries.log4cats,
+      Dependencies.Libraries.catsRetry,
+      Dependencies.Libraries.http4sClient,
+      Dependencies.Libraries.fs2BlobS3,
+      Dependencies.Libraries.fs2BlobGcs,
+      Dependencies.Libraries.metrics,
+      Dependencies.Libraries.pureconfig.withRevision(Dependencies.V.pureconfig013),
+      Dependencies.Libraries.pureconfigCats.withRevision(Dependencies.V.pureconfig013),
+      Dependencies.Libraries.pureconfigCirce.withRevision(Dependencies.V.pureconfig013),
+      Dependencies.Libraries.specs2,
+      Dependencies.Libraries.specs2CE,
+      Dependencies.Libraries.scalacheck,
+      Dependencies.Libraries.specs2Scalacheck,
+      Dependencies.Libraries.http4sDsl,
+      Dependencies.Libraries.http4sServer
+    ),
+    addCompilerPlugin("com.olegpy" %% "better-monadic-for" % "0.3.1")
+  )
+  .enablePlugins(BuildInfoPlugin)
+  .settings(BuildSettings.dockerSettings)
+  .enablePlugins(BuildInfoPlugin, JavaAppPackaging, DockerPlugin)

config/config.fs2.hocon.sample

@@ -0,0 +1,24 @@
+auth = {
+  type = "Gcp"
+  projectId = "test-project"
+}
+input = {
+  type = "PubSub"
+  subscriptionId = "inputSub"
+}
+good = {
+  type = "PubSub"
+  topic = "good-topic"
+}
+bad = {
+  type = "PubSub"
+  topic = "bad-topic"
+}
+
+assetsUpdatePeriod = "7 days"
+
+sentry = {
+  dsn = "http://sentry.acme.com"
+}
+
+metricsReportPeriod = "1 second"

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/adapters/registry/Adapter.scala

@@ -13,12 +13,9 @@
 package com.snowplowanalytics.snowplow.enrich.common.adapters.registry
 
 import cats.Monad
-import cats.data.{NonEmptyList, ValidatedNel}
+import cats.data.{NonEmptyList, Validated, ValidatedNel}
 import cats.data.Validated._
-import cats.syntax.either._
-import cats.syntax.eq._
-import cats.syntax.option._
-import cats.syntax.validated._
+import cats.implicits._
 
 import cats.effect.Clock
 
@@ -265,17 +262,11 @@ trait Adapter {
    * or Failures
    */
   protected[registry] def rawEventsListProcessor(
-    rawEventsList: List[ValidatedNel[FailureDetails.AdapterFailure, RawEvent]]
+    rawEventsList: List[Validated[NonEmptyList[FailureDetails.AdapterFailure], RawEvent]]
   ): ValidatedNel[FailureDetails.AdapterFailure, NonEmptyList[RawEvent]] = {
-    val successes: List[RawEvent] =
-      for {
-        Valid(s) <- rawEventsList
-      } yield s
-
-    val failures: List[FailureDetails.AdapterFailure] =
-      (for {
-        Invalid(NonEmptyList(h, t)) <- rawEventsList
-      } yield h :: t).flatten
+    val (failures, successes) = rawEventsList.separate match {
+      case (nel, list) => (nel.flatMap(_.toList), list)
+    }
 
     (successes, failures) match {
       // No Failures collected.

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/registry/apirequest/ApiRequestEnrichment.scala

@@ -18,6 +18,8 @@ import cats.{Id, Monad}
 import cats.data.{EitherT, NonEmptyList, ValidatedNel}
 import cats.implicits._
 
+import cats.effect.Sync
+
 import io.circe._
 import io.circe.generic.auto._
 
@@ -26,7 +28,6 @@ import com.snowplowanalytics.iglu.core.circe.implicits._
 
 import com.snowplowanalytics.lrumap._
 import com.snowplowanalytics.snowplow.badrows.FailureDetails
-
 import com.snowplowanalytics.snowplow.enrich.common.enrichments.registry.{Enrichment, ParseableEnrichment}
 import com.snowplowanalytics.snowplow.enrich.common.enrichments.registry.EnrichmentConf.ApiRequestConf
 import com.snowplowanalytics.snowplow.enrich.common.outputs.EnrichedEvent
@@ -231,4 +232,24 @@ object CreateApiRequestEnrichment {
             )
           )
     }
+
+  implicit def syncCreateApiRequestEnrichment[F[_]: Sync](
+    implicit CLM: CreateLruMap[F, String, (Either[Throwable, Json], Long)],
+    HTTP: HttpClient[F]
+  ): CreateApiRequestEnrichment[F] =
+    new CreateApiRequestEnrichment[F] {
+      def create(conf: ApiRequestConf): F[ApiRequestEnrichment[F]] =
+        CLM
+          .create(conf.cache.size)
+          .map(c =>
+            ApiRequestEnrichment(
+              conf.schemaKey,
+              conf.inputs,
+              conf.api,
+              conf.outputs,
+              conf.cache.ttl,
+              c
+            )
+          )
+    }
 }