Merge pull request #2 from slve/refactor-and-add-tests

Refactor, add tests, upgrade http4s

build.sbt

@@ -1,18 +1,22 @@
 name := "scala"
+parallelExecution := false
 scalaVersion := "2.13.3"
 scalacOptions ++= Seq("-Wconf:any:warning-verbose")
 
 libraryDependencies ++= Seq(
   "co.fs2"                        %% "fs2-core"               % "2.5.0",
   "com.softwaremill.sttp.client3" %% "core"                   % "3.1.1",
   "org.http4s"                    %% "http4s-jdk-http-client" % "0.3.5",
-  "org.scalatest"                 %% "scalatest"              % "3.0.8",
+  "org.scalatest"                 %% "scalatest"              % "3.2.2",
   "org.typelevel"                 %% "cats-core"              % "2.4.1",
   "org.typelevel"                 %% "cats-effect"            % "2.3.1"
 ) ++ Seq(
   "org.http4s" %% "http4s-blaze-client",
   "org.http4s" %% "http4s-blaze-client",
   "org.http4s" %% "http4s-blaze-server",
   "org.http4s" %% "http4s-dsl",
-  "org.http4s" %% "http4s-ember-client"
-).map(_ % "0.21.18")
+  "org.http4s" %% "http4s-async-http-client",
+  "org.http4s" %% "http4s-ember-client",
+  "org.http4s" %% "http4s-jetty-client",
+  "org.http4s" %% "http4s-okhttp-client"
+).map(_ % "0.21.19")

src/main/scala/AsyncHttpClientTest.scala

@@ -0,0 +1,31 @@
+import cats.effect.{ExitCode, IO, IOApp, Resource}
+import fs2.Stream
+import helpers.requestStream
+import org.http4s._
+import org.http4s.client.Client
+import org.http4s.client.asynchttpclient.AsyncHttpClient
+import org.http4s.dsl.io._
+import org.http4s.implicits._
+
+import scala.concurrent.duration._
+
+class AsyncHttpClientTest(appTime: FiniteDuration, requestPayloadSize: Int, responsePayloadSize: Int) extends IOApp {
+
+  val uri = uri"http://localhost:8099"
+  val body = "x" * requestPayloadSize
+  val req = Request[IO](POST, uri).withEntity(body)
+  val response = "x" * responsePayloadSize
+
+  override def run(args: List[String]): IO[ExitCode] = {
+    def request(client: Client[IO]): Stream[IO, String] = client.stream(req).flatMap(_.bodyText)
+
+    val simpleClient: Resource[IO, Client[IO]] = {
+      AsyncHttpClient.resource[IO]()
+    }
+
+    simpleClient.use { client =>
+      new Server(requestStream(request(client), appTime), appTime, response).run(List())
+    }
+  }
+
+}

src/main/scala/BlazeClientTest.scala

@@ -1,83 +1,32 @@
-package io.bitrise.apm.symbolicator
-
-import cats.effect.{ExitCode, IO, IOApp}
+import cats.effect.{ConcurrentEffect, ExitCode, IO, IOApp}
 import fs2.Stream
-import org.http4s.{HttpRoutes, _}
+import helpers.requestStream
+import org.http4s._
 import org.http4s.client.Client
 import org.http4s.client.blaze.BlazeClientBuilder
 import org.http4s.dsl.io._
 import org.http4s.implicits._
-import org.http4s.server.blaze.BlazeServerBuilder
 
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
-object BlazeClientTest extends IOApp {
-
-  // Numbers below vary on different computers
-  // In my case, if the request payload size is 65398 or greater
-  //  AND response payload size is 81161 or greater
-  //  then I get an EOF exception in some but not all cases
-  // If however either of these payload sizes is lower then
-  //  EOF exception doesn't occur, even if running for an extended period
-
-  // broken on first request
-  val appTime = 5.seconds
-  val requestPayloadSize = 65398
-  val responsePayloadSize = 81161
-
-  // can hold for 5 seconds, but broken on longer run, like 30 seconds appTime
-  //val appTime = 30.seconds
-  //val requestPayloadSize = 65397
-  //val responsePayloadSize = 81161
-
-  // more stable, can hold up to 30 seconds appTime, seen broken on 120 seconds
-  //val appTime = 30.seconds
-  //val requestPayloadSize = 65398
-  //val responsePayloadSize = 81160
+class BlazeClientTest(appTime: FiniteDuration, requestPayloadSize: Int, responsePayloadSize: Int) extends IOApp {
 
   val uri = uri"http://localhost:8099"
   val body = "x" * requestPayloadSize
   val req = Request[IO](POST, uri).withEntity(body)
   val response = "x" * responsePayloadSize
 
-  var i = 0
   override def run(args: List[String]): IO[ExitCode] = {
-    def requestStream(client: Client[IO]): Stream[IO, Unit] = Stream
-      .fixedRate(0.01.second)
-      .flatMap(_ => {
-        i = i + 1
+    def request(client: Client[IO]): Stream[IO, String] = client.stream(req).flatMap(_.bodyText)
 
-        client.stream(req).flatMap(_.bodyText)
-      })
-      .evalMap(c => IO.delay(println(s"$i ${c.size}")))
-      .interruptAfter(appTime)
+    def simpleClient(implicit c: ConcurrentEffect[IO]): BlazeClientBuilder[IO] =
+      BlazeClientBuilder[IO](ExecutionContext.global)
+        .withRequestTimeout(45.seconds)
+        .withIdleTimeout(1.minute)
+        .withResponseHeaderTimeout(44.seconds)
 
-    server(simpleClient.stream.flatMap(requestStream))
+    new Server(simpleClient.stream.flatMap(c => requestStream(request(c), appTime)), appTime, response).run(List())
   }
 
-  val simpleClient: BlazeClientBuilder[IO] =
-    BlazeClientBuilder[IO](ExecutionContext.global)
-      .withRequestTimeout(45.seconds)
-      .withIdleTimeout(1.minute)
-      .withResponseHeaderTimeout(44.seconds)
-
-  def server(app: Stream[IO, Unit]) =
-    BlazeServerBuilder[IO](ExecutionContext.global)
-      .withIdleTimeout(5.minutes)
-      .bindHttp(8099, "0.0.0.0")
-      .withHttpApp(
-        HttpRoutes
-          .of[IO] {
-            case POST -> Root => Ok(response)
-          }
-          .orNotFound
-      )
-      .serve
-      .concurrently(app)
-      .interruptAfter(appTime + 2.seconds)
-      .compile
-      .drain
-      .as(ExitCode.Success)
-
 }

src/main/scala/EmberClientTest.scala

@@ -1,78 +1,30 @@
-package io.bitrise.apm.symbolicator
-
 import cats.effect.{ExitCode, IO, IOApp, Resource}
 import fs2.Stream
+import helpers.requestStream
+import org.http4s._
 import org.http4s.client.Client
 import org.http4s.dsl.io._
 import org.http4s.ember.client.EmberClientBuilder
 import org.http4s.implicits._
-import org.http4s.server.blaze.BlazeServerBuilder
-import org.http4s.{HttpRoutes, _}
 
-import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
-object EmberClientTest extends IOApp {
-
-  // Numbers below vary on different computers
-  // In my case, if the request payload size is 65337 or greater
-  //  then I get an java.io.IOException on the first request
-  // If however the request payload size is 65336 or lower
-  //  java.io.IOException doesn't occur, even if the response size is 200MB
-  //  while running the test for an extended period
-
-  // broken on first request
-  val appTime = 5.seconds
-  val requestPayloadSize = 65337
-  val responsePayloadSize = 1
-
-  // stable for extended period
-  //val appTime = 120.seconds
-  //val requestPayloadSize = 65336
-  //val responsePayloadSize = 200 * 1000 * 1000
+class EmberClientTest(appTime: FiniteDuration, requestPayloadSize: Int, responsePayloadSize: Int) extends IOApp {
 
   val uri = uri"http://localhost:8099"
   val body = "x" * requestPayloadSize
   val req = Request[IO](POST, uri).withEntity(body)
   val response = "x" * responsePayloadSize
 
-  var i = 0
   override def run(args: List[String]): IO[ExitCode] = {
-    def requestStream(client: Client[IO]): Stream[IO, Unit] =
-      Stream
-        .fixedRate(0.01.second)
-        .flatMap(_ => {
-          i = i + 1
+    def request(client: Client[IO]): Stream[IO, String] = client.stream(req).flatMap(_.bodyText)
 
-          client.stream(req).flatMap(_.bodyText)
-        })
-        .evalMap(c => IO.delay(println(s"$i ${c.size}")))
-        .interruptAfter(appTime)
+    val simpleClient: Resource[IO, Client[IO]] =
+      EmberClientBuilder.default[IO].build
 
     simpleClient.use { client =>
-      server(requestStream(client))
+      new Server(requestStream(request(client), appTime), appTime, response).run(List())
     }
   }
 
-  val simpleClient: Resource[IO, Client[IO]] =
-    EmberClientBuilder.default[IO].build
-
-  def server(app: Stream[IO, Unit]) =
-    BlazeServerBuilder[IO](ExecutionContext.global)
-      .withIdleTimeout(5.minutes)
-      .bindHttp(8099, "0.0.0.0")
-      .withHttpApp(
-        HttpRoutes
-          .of[IO] {
-            case POST -> Root => Ok(response)
-          }
-          .orNotFound
-      )
-      .serve
-      .concurrently(app)
-      .interruptAfter(appTime + 2.seconds)
-      .compile
-      .drain
-      .as(ExitCode.Success)
-
 }

src/main/scala/JdkHttpClientTest.scala

@@ -1,90 +1,29 @@
-package io.bitrise.apm.symbolicator
-
 import cats.effect.{ExitCode, IO, IOApp}
 import fs2.Stream
+import helpers.requestStream
+import org.http4s._
 import org.http4s.client.Client
 import org.http4s.client.jdkhttpclient.JdkHttpClient
 import org.http4s.dsl.io._
 import org.http4s.implicits._
-import org.http4s.server.blaze.BlazeServerBuilder
-import org.http4s.{HttpRoutes, _}
 
-import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
-object JdkHttpClientTest extends IOApp {
-
-  val appTime = 300.seconds
-  // Numbers below may vary on different computers
-  // In my case, if the request payload size is 523329 or greater
-  //  OR response payload size is 65417 or greater
-  //  then I get a java.io.IOException: fixed content-length: 65416, bytes received: 49032
-  //  in some but not all cases.
-  // If however both of these payload sizes are lower then
-  //  fixed content-length exception doesn't occur, even if running for an extended period.
-  //  Yet in some rare cases I get a java.io.IOException: HTTP/1.1 header parser received no bytes
-
-  // stable - at least for content-length
-  val requestPayloadSize = 523328 // it's 8 times 65416 !?
-  val responsePayloadSize = 65416 // it's 8 times 8177 !?
-
-  // broken - yet fairly stable
-  // val requestPayloadSize = 523329
-  // val responsePayloadSize = 65416
-
-  // quite broken
-  // val requestPayloadSize = 523328
-  // val responsePayloadSize = 65417
+class JdkHttpClientTest(appTime: FiniteDuration, requestPayloadSize: Int, responsePayloadSize: Int) extends IOApp {
 
   val uri = uri"http://localhost:8099"
   val body = "x" * requestPayloadSize
   val req = Request[IO](POST, uri).withEntity(body)
   val response = "x" * responsePayloadSize
 
-  var i = 0
   override def run(args: List[String]): IO[ExitCode] = {
-    simpleClient.flatMap(client => {
-      val requestStream: Stream[IO, Unit] = Stream
-        .fixedRate(0.01.second)
-        .flatMap(_ => {
-          i = i + 1
-          val y: Stream[IO, Response[IO]] = client.stream(req)
-          y.flatMap(r => r.bodyText)
-        })
-        .evalMap(c => IO.delay(println(s"$i ${c.size}")))
-        .interruptAfter(appTime)
+    def request(client: Client[IO]): Stream[IO, String] = client.stream(req).flatMap(_.bodyText)
 
-      server(requestStream)
-    })
-  }
-
-  import java.net.http.HttpClient
-  val client0: IO[Client[IO]] = IO {
-    HttpClient
-      .newBuilder()
-      .version(HttpClient.Version.HTTP_2)
-      .build()
-
-  }.map(JdkHttpClient(_))
+    val simpleClient: IO[Client[IO]] = JdkHttpClient.simple[IO]
 
-  val simpleClient: IO[Client[IO]] = JdkHttpClient.simple[IO]
-
-  def server(app: Stream[IO, Unit]) =
-    BlazeServerBuilder[IO](ExecutionContext.global)
-      .withIdleTimeout(5.minutes)
-      .bindHttp(8099, "0.0.0.0")
-      .withHttpApp(
-        HttpRoutes
-          .of[IO] {
-            case POST -> Root => Ok(response)
-          }
-          .orNotFound
-      )
-      .serve
-      .concurrently(app)
-      .interruptAfter(appTime + 2.seconds)
-      .compile
-      .drain
-      .as(ExitCode.Success)
+    simpleClient.flatMap { client =>
+      new Server(requestStream(request(client), appTime), appTime, response).run(List())
+    }
+  }
 
 }

src/main/scala/JettyClientTest.scala

@@ -0,0 +1,31 @@
+import cats.effect.{ExitCode, IO, IOApp, Resource}
+import fs2.Stream
+import helpers.requestStream
+import org.http4s._
+import org.http4s.client.Client
+import org.http4s.client.jetty.JettyClient
+import org.http4s.dsl.io._
+import org.http4s.implicits._
+
+import scala.concurrent.duration._
+
+class JettyClientTest(appTime: FiniteDuration, requestPayloadSize: Int, responsePayloadSize: Int) extends IOApp {
+
+  val uri = uri"http://localhost:8099"
+  val body = "x" * requestPayloadSize
+  val req = Request[IO](POST, uri).withEntity(body)
+  val response = "x" * responsePayloadSize
+
+  override def run(args: List[String]): IO[ExitCode] = {
+    def request(client: Client[IO]): Stream[IO, String] = client.stream(req).flatMap(_.bodyText)
+
+    val simpleClient: Resource[IO, Client[IO]] = {
+      JettyClient.resource[IO]()
+    }
+
+    simpleClient.use { client =>
+      new Server(requestStream(request(client), appTime), appTime, response).run(List())
+    }
+  }
+
+}