Working outline of a cats-effect wrapped nsq implementation

build.sbt

@@ -261,27 +261,30 @@ lazy val kafkaDistroless = project
   .dependsOn(core % "test->test;compile->compile")
 
 lazy val nsqSettings =
-  allSettings ++ buildInfoSettings ++ Seq(
+  allSettings ++ buildInfoSettings ++ http4sBuildInfoSettings ++ Seq(
     moduleName := "snowplow-stream-collector-nsq",
     Docker / packageName := "scala-stream-collector-nsq",
+    buildInfoPackage := s"com.snowplowanalytics.snowplow.collectors.scalastream",
     libraryDependencies ++= Seq(
+      Dependencies.Libraries.catsRetry,
       Dependencies.Libraries.nsqClient,
       Dependencies.Libraries.jackson,
+      Dependencies.Libraries.nettyAll,
       Dependencies.Libraries.log4j
     )
   )
 
 lazy val nsq = project
   .settings(nsqSettings)
   .enablePlugins(JavaAppPackaging, SnowplowDockerPlugin, BuildInfoPlugin)
-  .dependsOn(core % "test->test;compile->compile")
+  .dependsOn(http4s % "test->test;compile->compile")
 
 lazy val nsqDistroless = project
   .in(file("distroless/nsq"))
   .settings(sourceDirectory := (nsq / sourceDirectory).value)
   .settings(nsqSettings)
   .enablePlugins(JavaAppPackaging, SnowplowDistrolessDockerPlugin, BuildInfoPlugin)
-  .dependsOn(core % "test->test;compile->compile")
+  .dependsOn(http4s % "test->test;compile->compile")
 
 lazy val stdoutSettings =
   allSettings ++ buildInfoSettings ++ http4sBuildInfoSettings ++ Seq(

nsq/src/main/resources/application.conf

@@ -14,26 +14,3 @@ collector {
     }
   }
 }
-
-
-akka {
-  loglevel = WARNING
-  loggers = ["akka.event.slf4j.Slf4jLogger"]
-
-  http.server {
-    remote-address-header = on
-    raw-request-uri-header = on
-
-    parsing {
-      max-uri-length = 32768
-      uri-parsing-mode = relaxed
-      illegal-header-warnings = off
-    }
-
-    max-connections = 2048
-  }
-
-  coordinated-shutdown {
-    run-by-jvm-shutdown-hook = off
-  }
-}

nsq/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/NsqCollector.scala

@@ -14,28 +14,31 @@
  */
 package com.snowplowanalytics.snowplow.collectors.scalastream
 
-import com.snowplowanalytics.snowplow.collectors.scalastream.generated.BuildInfo
-import com.snowplowanalytics.snowplow.collectors.scalastream.model._
-import com.snowplowanalytics.snowplow.collectors.scalastream.sinks.NsqSink
-import com.snowplowanalytics.snowplow.collectors.scalastream.telemetry.TelemetryAkkaService
+import java.util.concurrent.ScheduledThreadPoolExecutor
 
-object NsqCollector extends Collector {
-  def appName      = BuildInfo.shortName
-  def appVersion   = BuildInfo.version
-  def scalaVersion = BuildInfo.scalaVersion
+import cats.effect.{IO, Resource}
+import com.snowplowanalytics.snowplow.collector.core.model.Sinks
+import com.snowplowanalytics.snowplow.collector.core.{App, Config}
+import com.snowplowanalytics.snowplow.collectors.scalastream.sinks._
 
-  def main(args: Array[String]): Unit = {
-    val (collectorConf, akkaConf) = parseConfig(args)
-    val telemetry                 = TelemetryAkkaService.initWithCollector(collectorConf, BuildInfo.moduleName, appVersion)
-    val sinks = {
-      val goodStream = collectorConf.streams.good
-      val badStream  = collectorConf.streams.bad
-      val (good, bad) = collectorConf.streams.sink match {
-        case nc: Nsq => (new NsqSink(nc.maxBytes, nc, goodStream), new NsqSink(nc.maxBytes, nc, badStream))
-        case _       => throw new IllegalArgumentException("Configured sink is not NSQ")
-      }
-      CollectorSinks(good, bad)
-    }
-    run(collectorConf, akkaConf, sinks, telemetry)
+object NsqCollector extends App[NsqSinkConfig](BuildInfo) {
+  override def mkSinks(config: Config.Streams[NsqSinkConfig]): Resource[IO, Sinks[IO]] = {
+    val threadPoolExecutor = new ScheduledThreadPoolExecutor(config.sink.threadPoolSize)
+    for {
+      good <- NsqSink.create[IO](
+        config.sink.maxBytes,
+        config.sink,
+        // config.buffer,
+        config.good,
+        threadPoolExecutor
+      )
+      bad <- NsqSink.create[IO](
+        config.sink.maxBytes,
+        config.sink,
+        // config.buffer,
+        config.bad,
+        threadPoolExecutor
+      )
+    } yield Sinks(good, bad)
   }
 }

nsq/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/NsqSink.scala

@@ -21,15 +21,38 @@ package sinks
 
 import scala.collection.JavaConverters._
 
+import java.util.concurrent.ScheduledExecutorService
+
+import scala.concurrent.ExecutionContextExecutorService
+import scala.concurrent.duration.MILLISECONDS
+import scala.collection.JavaConverters._
+
+import cats.effect.{Resource, Sync}
+// import cats.implicits.catsSyntaxMonadErrorRethrow
+
+// import org.slf4j.LoggerFactory
+
 import com.snowplowanalytics.client.nsq.NSQProducer
-import com.snowplowanalytics.snowplow.collectors.scalastream.model._
+import com.snowplowanalytics.snowplow.collector.core.{Sink}
 
 /**
   * NSQ Sink for the Scala Stream Collector
   * @param nsqConfig Configuration for Nsq
   * @param topicName Nsq topic name
   */
-class NsqSink(val maxBytes: Int, nsqConfig: Nsq, topicName: String) extends Sink {
+class NsqSink[F[_]: Sync] private (
+  val maxBytes: Int,
+  nsqConfig: NsqSinkConfig,
+  topicName: String,
+  executorService: ScheduledExecutorService
+) extends Sink[F] {
+
+  // private lazy val log = LoggerFactory.getLogger(getClass())
+
+  implicit lazy val ec: ExecutionContextExecutorService =
+    concurrent.ExecutionContext.fromExecutorService(executorService)
+
+  override def isHealthy: F[Boolean] = Sync[F].pure(true)
 
   private val producer = new NSQProducer().addAddress(nsqConfig.host, nsqConfig.port).start()
 
@@ -38,9 +61,32 @@ class NsqSink(val maxBytes: Int, nsqConfig: Nsq, topicName: String) extends Sink
     * @param events The list of events to send
     * @param key The partition key (unused)
     */
-  override def storeRawEvents(events: List[Array[Byte]], key: String): Unit =
-    producer.produceMulti(topicName, events.asJava)
+  override def storeRawEvents(events: List[Array[Byte]], key: String): F[Unit] =
+    Sync[F].delay(producer.produceMulti(topicName, events.asJava))
 
-  override def shutdown(): Unit =
+  def shutdown(): Unit =
     producer.shutdown()
+  executorService.shutdown()
+  executorService.awaitTermination(10000, MILLISECONDS)
+  ()
+}
+
+object NsqSink {
+
+  def create[F[_]: Sync](
+    maxBytes: Int,
+    nsqConfig: NsqSinkConfig,
+    // bufferConfig: Config.Buffer,
+    topicName: String,
+    executorService: ScheduledExecutorService
+  ): Resource[F, NsqSink[F]] = {
+    val acquire = Sync[F].delay(
+      new NsqSink(maxBytes, nsqConfig, topicName, executorService)
+    )
+    //.rethrow
+
+    val release = (sink: NsqSink[F]) => (Sync[F].delay(sink.shutdown()))
+
+    Resource.make(acquire)(release)
+  }
 }

nsq/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/NsqSinkConfig.scala

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2013-2022 Snowplow Analytics Ltd.
+ * All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache
+ * License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at
+ * http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied.
+ *
+ * See the Apache License Version 2.0 for the specific language
+ * governing permissions and limitations there under.
+ */
+
+package com.snowplowanalytics.snowplow.collectors.scalastream.sinks
+
+import io.circe.Decoder
+import io.circe.generic.semiauto._
+
+import com.snowplowanalytics.snowplow.collector.core.Config
+
+final case class NsqSinkConfig(
+  maxBytes: Int,
+  // backoffPolicy: NsqSinkConfig.BackoffPolicyConfig, // TODO: Figure out if this is used/needed/should be included as optional
+  threadPoolSize: Int,
+  host: String,
+  port: Int
+) extends Config.Sink
+
+object NsqSinkConfig {
+
+  final case class BackoffPolicyConfig(minBackoff: Long, maxBackoff: Long, maxRetries: Int)
+
+  implicit val configDecoder: Decoder[NsqSinkConfig]              = deriveDecoder[NsqSinkConfig]
+  implicit val backoffPolicyDecoder: Decoder[BackoffPolicyConfig] = deriveDecoder[BackoffPolicyConfig]
+}

nsq/src/test/scala/com.snowplowanalytics.snowplow.collectors.scalastream/NsqConfigSpec.scala

@@ -16,10 +16,16 @@
   * See the Apache License Version 2.0 for the specific language
   * governing permissions and limitations there under.
   */
+
+/*
 package com.snowplowanalytics.snowplow.collectors.scalastream
 
 import com.snowplowanalytics.snowplow.collectors.scalastream.config.ConfigSpec
 
 class NsqConfigSpec extends ConfigSpec {
   makeConfigTest("nsq", "", "")
 }
+
+ */
+
+// TODO: Add back in and fix test

project/Dependencies.scala

@@ -56,6 +56,7 @@ object Dependencies {
     val circeConfig      = "0.10.0"
     val fs2PubSub        = "0.22.0"
     val catsRetry        = "3.1.0"
+    val nettyAll         = "4.1.95.Final" // to fix nsq dependency
 
     // Scala (test only)
     val specs2         = "4.11.0"
@@ -72,6 +73,7 @@ object Dependencies {
   object Libraries {
     // Java
     val jackson        = "com.fasterxml.jackson.core"       % "jackson-databind"        % V.jackson // nsq only
+    val nettyAll       = "io.netty"                         % "netty-all"               % V.nettyAll //nsq only
     val thrift         = "org.apache.thrift"                % "libthrift"               % V.thrift
     val kinesis        = "com.amazonaws"                    % "aws-java-sdk-kinesis"    % V.awsSdk
     val sqs            = "com.amazonaws"                    % "aws-java-sdk-sqs"        % V.awsSdk