Load config

build.sbt

@@ -134,6 +134,9 @@ lazy val http4s = project
       Dependencies.Libraries.badRows,
       Dependencies.Libraries.collectorPayload,
       Dependencies.Libraries.slf4j,
+      Dependencies.Libraries.decline,
+      Dependencies.Libraries.circeGeneric,
+      Dependencies.Libraries.circeConfig,
       Dependencies.Libraries.specs2
     )
   )

http4s/src/main/resources/reference.conf

@@ -0,0 +1,82 @@
+collector {
+  paths {}
+
+  p3p {
+    policyRef = "/w3c/p3p.xml"
+    CP = "NOI DSP COR NID PSA OUR IND COM NAV STA"
+  }
+
+  crossDomain {
+    enabled = false
+    domains = [ "*" ]
+    secure = true
+  }
+
+  cookie {
+    enabled = true
+    expiration = 365 days
+    name = sp
+    secure = true
+    httpOnly = true
+    sameSite = "None"
+  }
+
+  doNotTrackCookie {
+    enabled = false
+    name = ""
+    value = ""
+  }
+
+  cookieBounce {
+    enabled = false
+    name = "n3pc"
+    fallbackNetworkUserId = "00000000-0000-4000-A000-000000000000"
+  }
+
+  redirectMacro {
+    enabled = false
+  }
+
+  rootResponse {
+    enabled = false
+    statusCode = 302
+    headers = {}
+    body = ""
+  }
+
+  cors {
+    accessControlMaxAge = 60 minutes
+  }
+
+  streams {
+    useIpAddressAsPartitionKey = false
+
+    buffer {
+      byteLimit = 3145728
+      recordLimit = 500
+      timeLimit = 5000
+    }
+  }
+
+  monitoring {
+    metrics {
+      statsd {
+        enabled = false
+        hostname = localhost
+        port = 8125
+        period = 10 seconds
+        prefix = snowplow.collector
+      }
+    }
+  }
+
+  ssl {
+    enable = false
+    redirect = false
+    port = 443
+  }
+
+  enableDefaultRedirect = false
+
+  redirectDomains = []
+}

http4s/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/CollectorApp.scala

@@ -1,20 +1,26 @@
 package com.snowplowanalytics.snowplow.collectors.scalastream
 
+import java.net.InetSocketAddress
+
+import org.typelevel.log4cats.Logger
+import org.typelevel.log4cats.slf4j.Slf4jLogger
+
+import scala.concurrent.duration.{DurationLong, FiniteDuration}
+
 import cats.implicits._
+
 import cats.effect.{Async, ExitCode, Sync}
 import cats.effect.kernel.Resource
+
 import fs2.io.net.Network
+
 import com.comcast.ip4s.IpLiteralSyntax
+
 import org.http4s.HttpApp
 import org.http4s.server.Server
 import org.http4s.ember.server.EmberServerBuilder
 import org.http4s.blaze.server.BlazeServerBuilder
 import org.http4s.netty.server.NettyServerBuilder
-import org.typelevel.log4cats.Logger
-import org.typelevel.log4cats.slf4j.Slf4jLogger
-
-import java.net.InetSocketAddress
-import scala.concurrent.duration.{DurationLong, FiniteDuration}
 
 import com.snowplowanalytics.snowplow.collectors.scalastream.model._
 
@@ -24,18 +30,21 @@ object CollectorApp {
     Slf4jLogger.getLogger[F]
 
   def run[F[_]: Async](
-    mkGood: Resource[F, Sink[F]],
-    mkBad: Resource[F, Sink[F]],
-    config: CollectorConfig,
+    args: List[String],
     appName: String,
-    appVersion: String
+    appVersion: String,
+    mkSinks: Config.Streams => Resource[F, Sinks[F]]
   ): F[ExitCode] = {
     val resources = for {
-      bad  <- mkBad
-      good <- mkGood
+      config <- Resource.eval(Config.parse(args))
+      sinks  <- mkSinks(config.streams)
       _ <- withGracefulShutdown(610.seconds) {
-        val sinks                                 = CollectorSinks(good, bad)
-        val collectorService: CollectorService[F] = new CollectorService[F](config, sinks, appName, appVersion)
+        val collectorService: CollectorService[F] = new CollectorService[F](
+          config,
+          Sinks(sinks.good, sinks.bad),
+          appName,
+          appVersion
+        )
         buildHttpServer[F](new CollectorRoutes[F](collectorService).value)
       }
     } yield ()

http4s/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/CollectorService.scala

@@ -36,8 +36,8 @@ trait Service[F[_]] {
 }
 
 class CollectorService[F[_]: Sync](
-  config: CollectorConfig,
-  sinks: CollectorSinks[F],
+  config: Config,
+  sinks: Sinks[F],
   appName: String,
   appVersion: String
 ) extends Service[F] {
@@ -65,8 +65,7 @@ class CollectorService[F[_]: Sync](
     for {
       body     <- body
       hostname <- hostname
-      // TODO: Get ipAsPartitionKey from config
-      (ipAddress, partitionKey) = ipAndPartitionKey(ip, ipAsPartitionKey = false)
+      (ipAddress, partitionKey) = ipAndPartitionKey(ip, config.streams.useIpAddressAsPartitionKey)
       // TODO: nuid should be set properly
       nuid = UUID.randomUUID().toString
       event = buildEvent(