Common: Change http4s client backend to async-http-client (close #903)

Author: istreeter

modules/loader/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/dsl/Environment.scala

@@ -22,7 +22,7 @@ import cats.effect.{ContextShift, Blocker, Clock, Resource, Timer, ConcurrentEff
 
 import doobie.ConnectionIO
 
-import org.http4s.client.blaze.BlazeClientBuilder
+import org.http4s.client.asynchttpclient.AsyncHttpClient
 
 import io.sentry.{SentryClient, Sentry, SentryOptions}
 
@@ -76,7 +76,7 @@ object Environment {
 
     for {
       blocker <- Blocker[F]
-      httpClient <- BlazeClientBuilder[F](blocker.blockingContext).resource
+      httpClient <- AsyncHttpClient.resource[F]()
       iglu <- Iglu.igluInterpreter(httpClient, cli.resolverConfig)
       implicit0(logging: Logging[F]) = Logging.loggingInterpreter[F](List(cli.config.storage.password.getUnencrypted, cli.config.storage.username))
       tracker <- Monitoring.initializeTracking[F](cli.config.monitoring, httpClient)

modules/transformer-kinesis/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/kinesis/Main.scala

@@ -35,8 +35,7 @@ object Main extends IOApp {
         case Right(cliConfig) =>
           Resources.mk[IO](
             cliConfig.igluConfig,
-            cliConfig.config,
-            executionContext
+            cliConfig.config
           ).use { resources =>
             logger[IO].info(s"Starting RDB Shredder with ${cliConfig.config} config") *>
               Processing.run[IO](resources, cliConfig.config)

modules/transformer-kinesis/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/kinesis/Resources.scala

@@ -16,7 +16,6 @@ import java.util.UUID
 import java.net.URI
 
 import scala.concurrent.duration.DurationInt
-import scala.concurrent.ExecutionContext
 
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
@@ -68,19 +67,17 @@ object Resources {
 
   def mk[F[_]: ConcurrentEffect : ContextShift: Clock: InitSchemaCache: InitListCache: Timer](
     igluConfig: Json,
-    config: TransformerConfig.Stream,
-    executionContext: ExecutionContext
+    config: TransformerConfig.Stream
   ): Resource[F, Resources[F]] =
     for {
       awsQueue <- mkQueueFromConfig(config.queue)
-      resources <- mk(igluConfig, config, awsQueue, executionContext)
+      resources <- mk(igluConfig, config, awsQueue)
     } yield resources
 
   def mk[F[_]: ConcurrentEffect : ContextShift: Clock: InitSchemaCache: InitListCache: Timer](
     igluConfig: Json,
     config: TransformerConfig.Stream,
-    awsQueue: AWSQueue[F],
-    executionContext: ExecutionContext
+    awsQueue: AWSQueue[F]
   ): Resource[F, Resources[F]] =
     for {
       client        <- mkClient(igluConfig)
@@ -92,7 +89,7 @@ object Resources {
       global        <- Resource.eval(Ref.of(0L))
       store         <- Resource.eval(mkStore[F](blocker, config.output.path))
       metrics       <- Resource.eval(Metrics.build[F](blocker, config.monitoring.metrics))
-      telemetry     <- Telemetry.build[F](config, BuildInfo.name, BuildInfo.version, executionContext)
+      telemetry     <- Telemetry.build[F](config, BuildInfo.name, BuildInfo.version)
     } yield
       Resources(
         client,

modules/transformer-kinesis/src/main/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/telemetry/Telemetry.scala

@@ -14,8 +14,6 @@
  */
 package com.snowplowanalytics.snowplow.rdbloader.transformer.telemetry
 
-import scala.concurrent.ExecutionContext
-
 import org.typelevel.log4cats.Logger
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
@@ -27,7 +25,7 @@ import cats.effect.{ConcurrentEffect, Resource, Sync, Timer}
 import fs2.Stream
 
 import org.http4s.client.{Client => HttpClient}
-import org.http4s.client.blaze.BlazeClientBuilder
+import org.http4s.client.asynchttpclient.AsyncHttpClient
 
 import io.circe.Json
 import io.circe.Encoder
@@ -56,11 +54,10 @@ object Telemetry {
   def build[F[_]: ConcurrentEffect: Timer](
     config: TransformerConfig.Stream,
     appName: String,
-    appVersion: String,
-    executionContext: ExecutionContext
+    appVersion: String
   ): Resource[F, Telemetry[F]] =
     for {
-      httpClient <- BlazeClientBuilder[F](executionContext).resource
+      httpClient <- AsyncHttpClient.resource[F]()
       tracker <- initTracker(config.telemetry, appName, httpClient)
     } yield new Telemetry[F] {
       def report: Stream[F, Unit] =

modules/transformer-kinesis/src/test/scala/com/snowplowanalytics/snowplow/rdbloader/transformer/kinesis/processing/TestApplication.scala

@@ -39,8 +39,7 @@ object TestApplication {
           Resources.mk[IO](
             cliConfig.igluConfig,
             cliConfig.config,
-            queueFromDeferred(forCompletionMessage),
-            concurrent.ExecutionContext.global
+            queueFromDeferred(forCompletionMessage)
           )
           .use { resources =>
             logger[IO].info(s"Starting RDB Shredder with ${cliConfig.config} config") *>

project/Dependencies.scala

@@ -101,7 +101,7 @@ object Dependencies {
   val monocleMacro      = "com.github.julien-truffaut" %% "monocle-macro"                     % V.monocle
   val catsRetry         = "com.github.cb372"           %% "cats-retry"                        % V.catsRetry
   val log4cats          = "org.typelevel"              %% "log4cats-slf4j"                    % V.log4cats
-  val http4sClient      = "org.http4s"                 %% "http4s-blaze-client"               % V.http4s
+  val http4sClient      = "org.http4s"                 %% "http4s-async-http-client"          % V.http4s
   val scalaTracker      = "com.snowplowanalytics"      %% "snowplow-scala-tracker-core"       % V.scalaTracker
   val scalaTrackerEmit  = "com.snowplowanalytics"      %% "snowplow-scala-tracker-emitter-http4s" % V.scalaTracker