Set maxBytes in the NsqSink

snowplow · Sep 27, 2023 · e8526a4 · e8526a4
1 parent c6982c8
commit e8526a4
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
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()))
 }