Set maxBytes in the NsqSink

snowplow · Sep 27, 2023 · 1d466cf · 1d466cf
1 parent b108007
commit 1d466cf
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 6 deletions.
diff --git a/http4s/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Run.scala b/http4s/src/main/scala/com.snowplowanalytics.snowplow.collector.core/Run.scala
@@ -44,7 +44,7 @@ object Run {
   ): F[ExitCode] = {
     val eitherT = for {
       config <- ConfigParser.fromPath[F, SinkConfig](path)
-      _      <- EitherT.right[ExitCode](fromConfig(appInfo, mkSinks, config, telemetryInfo))
+      _      <- EitherT.right[ExitCode](fromConfig(appInfo, mkSinks, telemetryInfo, config))
     } yield ExitCode.Success
 
     eitherT.merge.handleErrorWith { e =>
@@ -56,8 +56,8 @@ object Run {
   private def fromConfig[F[_]: Async: Tracking, SinkConfig](
     appInfo: AppInfo,
     mkSinks: Config.Streams[SinkConfig] => Resource[F, Sinks[F]],
-    config: Config[SinkConfig],
-    telemetryInfo: Config[SinkConfig] => Telemetry.TelemetryInfo
+    telemetryInfo: Config[SinkConfig] => Telemetry.TelemetryInfo,
+    config: Config[SinkConfig]
   ): F[ExitCode] = {
     val resources = for {
       sinks <- mkSinks(config.streams)

diff --git a/nsq/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/NsqCollector.scala b/nsq/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/NsqCollector.scala
@@ -23,10 +23,12 @@ object NsqCollector extends App[NsqSinkConfig](BuildInfo) {
   override def mkSinks(config: Config.Streams[NsqSinkConfig]): Resource[IO, Sinks[IO]] =
     for {
       good <- NsqSink.create[IO](
+        config.sink.maxBytes,
         config.sink,
         config.good
       )
       bad <- NsqSink.create[IO](
+        config.sink.maxBytes,
         config.sink,
         config.bad
       )

diff --git a/nsq/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/NsqSink.scala b/nsq/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/sinks/NsqSink.scala
@@ -65,14 +65,13 @@ class NsqSink[F[_]: Sync] private (
 object NsqSink {
 
   def create[F[_]: Sync](
+    maxBytes: Int,
     nsqConfig: NsqSinkConfig,
     topicName: String
   ): Resource[F, NsqSink[F]] =
     Resource.make(
       Sync[F].delay(
-        // MaxBytes is never used but is required by the sink interface definition,
-        // So just pass any int val in.
-        new NsqSink(0, nsqConfig, topicName)
+        new NsqSink(maxBytes, nsqConfig, topicName)
       )
     )(sink => Sync[F].delay(sink.shutdown()))
 }