Add http gzip compression

Previously all the requests were delivered in plan bytes.
This adds ability to send compressed requests to the collector.

README.md

@@ -86,6 +86,7 @@ Alternatively you can write events directly to a S3 bucket:
 "output": {
   "type": "Http"
   "endpoint": "https://my.collector.endpoint.com"
+  "gzip": true
 }
 ```
 

sinks/src/main/scala/com.snowplowanalytics.snowplow.eventgen/Config.scala

@@ -71,7 +71,7 @@ object Config {
     case class File(path: URI) extends Output
     case class PubSub(subscription: String) extends Output
     case class Kafka(brokers: String, topic: String, producerConf: Map[String, String] = Map.empty) extends Output
-    case class Http(endpoint: org.http4s.Uri) extends Output
+    case class Http(endpoint: org.http4s.Uri, gzip: Option[Boolean]) extends Output
   }
 
   val configOpt   = Opts.option[Path]("config", "Path to the configuration HOCON").orNone

sinks/src/main/scala/com.snowplowanalytics.snowplow.eventgen/GZipMiddleware.scala

@@ -0,0 +1,65 @@
+/*
+ * Copyright (c) 2021-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.eventgen
+
+import cats.effect._
+import fs2.compression.Compression
+import fs2.compression.DeflateParams
+import fs2.compression.ZLibParams
+import org.http4s.Header
+import org.http4s.Request
+import org.http4s.headers.`Content-Encoding`
+import org.http4s.headers.`Content-Length`
+
+object GZipMiddleware {
+  val DefaultBufferSize = 32 * 1024
+
+  def apply[F[_]: Sync](
+    retainUserEncoding: Boolean,
+    bufferSize: Int = DefaultBufferSize,
+    level: DeflateParams.Level = DeflateParams.Level.DEFAULT
+  )(request: Request[F]): Request[F] = {
+    val updateContentTypeEncoding =
+      (retainUserEncoding, request.headers.get[`Content-Encoding`]) match {
+        case (true, Some(`Content-Encoding`(cc))) =>
+          Header.Raw(
+            `Content-Encoding`.headerInstance.name,
+            s"${cc.coding}, gzip"
+          )
+
+        case _ =>
+          Header.Raw(
+            `Content-Encoding`.headerInstance.name,
+            "gzip"
+          )
+      }
+    val compressPipe =
+      Compression
+        .forSync[F]
+        .gzip(
+          fileName = None,
+          modificationTime = None,
+          comment = None,
+          DeflateParams(
+            bufferSize = bufferSize,
+            level = level,
+            header = ZLibParams.Header.GZIP
+          )
+        )
+    request
+      .removeHeader[`Content-Length`]
+      .putHeaders(updateContentTypeEncoding)
+      .withBodyStream(request.body.through(compressPipe))
+  }
+
+}

sinks/src/main/scala/com.snowplowanalytics.snowplow.eventgen/Http.scala

@@ -35,7 +35,7 @@ object Http {
 
   def sink[F[_]: Async](properties: Config.Output.Http): Pipe[F, Main.GenOutput, Unit] = {
 
-    def buildRequesst(
+    def buildRequest(
       generatedRequest: HttpRequest
     ): Request[F] = {
 
@@ -97,7 +97,7 @@ object Http {
         case TrackerMethod.Head(_) => throw new NotImplementedError ( "HEAD requests not implemented" )
       }
 
-      return req
+      if (properties.gzip.getOrElse(false)) GZipMiddleware(false)(req) else req
     }
 
     val httpClient = EmberClientBuilder.default[F].build
@@ -106,7 +106,7 @@ object Http {
         Stream
           .resource(httpClient)
           .flatMap(client => 
-            st.map(_._3).map(buildRequesst).evalMap(req => client.status(req)).void)
+            st.map(_._3).map(buildRequest).evalMap(req => client.status(req)).void)
     }
   }
 }