Rework deprecated methods

kamon-io · Feb 28, 2023 · b4d0320 · b4d0320
1 parent 9c03703
commit b4d0320
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 12 deletions.
diff --git a/.../kamon-pekko-http/src/main/scala/kamon/instrumentation/pekko/http/ServerFlowWrapper.scala b/.../kamon-pekko-http/src/main/scala/kamon/instrumentation/pekko/http/ServerFlowWrapper.scala
@@ -108,8 +108,8 @@ object ServerFlowWrapper {
         override def onPull(): Unit =
           pull(requestIn)
 
-        override def onDownstreamFinish(): Unit =
-          cancel(requestIn)
+        override def onDownstreamFinish(t: Throwable): Unit =
+          cancel(requestIn, t)
       })
 
       setHandler(responseIn, new InHandler {
@@ -177,8 +177,8 @@ object ServerFlowWrapper {
         override def onPull(): Unit =
           pull(responseIn)
 
-        override def onDownstreamFinish(): Unit =
-          cancel(responseIn)
+        override def onDownstreamFinish(t: Throwable): Unit =
+          cancel(responseIn, t)
       })
 
       override def preStart(): Unit =

diff --git a/...ntation/kamon-pekko-http/src/test/scala/kamon/pekko/http/PekkoHttpServerTracingSpec.scala b/...ntation/kamon-pekko-http/src/test/scala/kamon/pekko/http/PekkoHttpServerTracingSpec.scala
@@ -31,6 +31,7 @@ import java.util.UUID
 import javax.net.ssl.{HostnameVerifier, SSLSession}
 import scala.concurrent.duration._
 import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
 
 class PekkoHttpServerTracingSpec extends AnyWordSpecLike with Matchers with ScalaFutures with Inside with InitAndStopKamonAfterAll
     with MetricInspection.Syntax with Reconfigure with TestWebServer with Eventually with OptionValues with TestSpanReporter {
@@ -39,7 +40,6 @@ class PekkoHttpServerTracingSpec extends AnyWordSpecLike with Matchers with Scal
 
   implicit private val system = ActorSystem("http-server-instrumentation-spec")
   implicit private val executor = system.dispatcher
-  implicit private val materializer = ActorMaterializer()
 
   val (sslSocketFactory, trustManager) = clientSSL()
   val okHttp = new OkHttpClient.Builder()
@@ -228,8 +228,12 @@ class PekkoHttpServerTracingSpec extends AnyWordSpecLike with Matchers with Scal
 
       "correctly time entity transfer timings" in {
         val target = s"$protocol://$interface:$port/$stream"
-        client.newCall(new Request.Builder().url(target).build()).execute()
 
+        try {
+          client.newCall(new Request.Builder().url(target).build()).execute()
+        } catch {
+          case NonFatal(_) => // call failed..
+        }
         val span = eventually(timeout(10 seconds)) {
           val span = testSpanReporter().nextSpan().value
           span.operationName shouldBe "/stream"

diff --git a/instrumentation/kamon-pekko-http/src/test/scala/kamon/testkit/TestWebServer.scala b/instrumentation/kamon-pekko-http/src/test/scala/kamon/testkit/TestWebServer.scala
@@ -18,9 +18,8 @@ package kamon.testkit
 
 import java.security.cert.{Certificate, CertificateFactory}
 import java.security.{KeyStore, SecureRandom}
-
 import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.http.scaladsl.{Http, HttpsConnectionContext}
+import org.apache.pekko.http.scaladsl.{ConnectionContext, Http, HttpsConnectionContext}
 import org.apache.pekko.http.scaladsl.model.{ContentTypes, HttpEntity, HttpResponse}
 import com.github.pjfanning.pekkohttpjson4s.Json4sSupport
 import org.apache.pekko.http.scaladsl.model.StatusCodes.{BadRequest, InternalServerError, OK}
@@ -30,6 +29,7 @@ import org.apache.pekko.http.scaladsl.server.{RequestContext, Route}
 import org.apache.pekko.stream.ActorMaterializer
 import org.apache.pekko.stream.scaladsl.Source
 import org.apache.pekko.util.ByteString
+
 import javax.net.ssl.{KeyManagerFactory, SSLContext, SSLSocketFactory, TrustManagerFactory, X509TrustManager}
 import kamon.Kamon
 import kamon.instrumentation.pekko.http.TracingDirectives
@@ -47,7 +47,6 @@ trait TestWebServer extends TracingDirectives {
     import Endpoints._
 
     implicit val ec: ExecutionContext = system.dispatcher
-    implicit val materializer = ActorMaterializer()
 
     val routes = logRequest("routing-request") {
       get {
@@ -180,9 +179,9 @@ trait TestWebServer extends TracingDirectives {
     }
 
     if(https)
-      new WebServer(interface, port, "https", Http().bindAndHandleAsync(Route.asyncHandler(routes), interface, port, httpContext()))
+      new WebServer(interface, port, "https", Http().newServerAt(interface, port).enableHttps(httpContext()).bind(Route.toFunction(routes)))
     else
-      new WebServer(interface, port, "http", Http().bindAndHandle(routes, interface, port))
+      new WebServer(interface, port, "http", Http().newServerAt(interface, port).bindFlow(routes))
   }
 
   def httpContext() = {
@@ -196,7 +195,7 @@ trait TestWebServer extends TracingDirectives {
     val context = SSLContext.getInstance("TLS")
     context.init(keyManagerFactory.getKeyManagers, null, new SecureRandom)
 
-    new HttpsConnectionContext(context)
+    ConnectionContext.httpsServer(context)
   }
 
   def clientSSL(): (SSLSocketFactory, X509TrustManager) = {