Version 1.1.0: Add support for custom logging context

Author: JannikArndt

README.md

@@ -10,11 +10,12 @@ This is a wrapper around the akka-http-client that adds
 * AWS request signing
 
 [![CircleCI](https://circleci.com/gh/moia-dev/scala-http-client/tree/master.svg?style=svg)](https://circleci.com/gh/moia-dev/scala-http-client/tree/master)
+[![Scala 2.13](https://img.shields.io/maven-central/v/io.moia/scala-http-client_2.13.svg)](https://search.maven.org/search?q=scala-http-client_2.13)
 
 ## Usage
 
 ```sbt
-libraryDependencies += "io.moia" % "scala-http-client" % "1.0.0"
+libraryDependencies += "io.moia" % "scala-http-client" % "1.1.0"
 ```
 
 ```scala
@@ -23,7 +24,9 @@ implicit val executionContext: ExecutionContext = system.dispatcher
 
 val httpClientConfig: HttpClientConfig = HttpClientConfig("http", isSecureConnection = false, "127.0.0.1", 8888)
 val clock: Clock                       = Clock.systemUTC()
-val httpMetrics: HttpMetrics           = (_: HttpMethod, _: Uri.Path, _: HttpResponse) => ()
+val httpMetrics: HttpMetrics[String]   = new HttpMetrics[String] {
+  override def meterResponse(method: HttpMethod, path: Uri.Path, response: HttpResponse)(implicit ctx: String): Unit = ()
+}
 val retryConfig: RetryConfig =
     RetryConfig(
       retriesTooManyRequests = 2,
@@ -55,4 +58,37 @@ response.flatMap {
   case DomainError(content)       => Unmarshal(content).to[DomainErrorObject].map(Left(_))
   case failure: HttpClientFailure => throw GatewayException(failure.toString)
 }
-```
+```
+
+## Custom Logging
+
+To use a custom logger (for correlation ids etc), you can use the typed `LoggingHttpClient`:
+
+```scala
+// create a context-class
+case class LoggingContext(context: String)
+
+// create a logger
+implicit val canLogString: CanLog[LoggingContext]            = new CanLog[LoggingContext] // override logMessage here!
+implicit val theLogger: LoggerTakingImplicit[LoggingContext] = Logger.takingImplicit(LoggerFactory.getLogger(getClass.getName))
+
+// create the client
+new LoggingHttpClient[LoggingContext](httpClientConfig, "TestGateway", typedHttpMetrics, retryConfig, clock, None)
+
+// create an implicit logging context
+implicit val ctx: LoggingContext                             = LoggingContext("Logging Context")
+
+// make a request
+testHttpClient.request(HttpMethods.POST, HttpEntity.Empty, "/test", immutable.Seq.empty, Deadline.now + 10.seconds)
+```
+
+## Publishing
+
+You need an account on https://oss.sonatype.org that can [access](https://issues.sonatype.org/browse/OSSRH-52948) the `io.moia` group.
+
+Add your credentials to `~/.sbt/sonatype_credential` and run
+```sbt
+sbt:scala-http-client> +publishSigned
+```
+
+Then head to https://oss.sonatype.org/#stagingRepositories, select the repository, `Close` and then `Release` it.

build.sbt

@@ -1,9 +1,8 @@
-
 lazy val root = (project in file("."))
   .settings(
     name := "scala-http-client",
     organization := "io.moia",
-    version := "1.0.0-akka2.5",
+    version := "1.1.0-akka2.5",
     licenses += ("Apache-2.0", url("http://www.apache.org/licenses/LICENSE-2.0")),
     scmInfo := Some(ScmInfo(url("https://github.com/moia-dev/scala-http-client"), "scm:git@github.com:moia-dev/scala-http-client.git")),
     homepage := Some(url("https://github.com/moia-dev/scala-http-client")),
@@ -17,7 +16,8 @@ lazy val root = (project in file("."))
       }
     },
     libraryDependencies ++= akkaDependencies ++ awsDependencies ++ testDependencies ++ loggingDependencies ++ otherDependencies
-  ).settings(sonatypeSettings: _*)
+  )
+  .settings(sonatypeSettings: _*)
 
 val akkaVersion     = "2.5.29"
 val akkaHttpVersion = "10.1.11"
@@ -36,8 +36,9 @@ lazy val awsDependencies = Seq(
 )
 
 lazy val testDependencies = Seq(
-  "org.scalatest" %% "scalatest"     % "3.1.0"  % Test,
-  "org.mockito"   %% "mockito-scala" % "1.11.2" % Test
+  "org.scalatest"   %% "scalatest"       % "3.1.0"  % Test,
+  "org.mockito"     %% "mockito-scala"   % "1.11.2" % Test,
+  "org.mock-server" % "mockserver-netty" % "5.9.0"  % Test
 )
 
 lazy val loggingDependencies = Seq(
@@ -90,4 +91,4 @@ lazy val sonatypeSettings = {
     sonatypeProjectHosting := Some(GitHubHosting("moia-dev", "scala-http-client", "oss-support@moia.io")),
     credentials += Credentials(Path.userHome / ".sbt" / "sonatype_credential")
   )
-}
+}

src/main/scala/io/moia/scalaHttpClient/HttpClient.scala

@@ -7,7 +7,8 @@ import akka.http.scaladsl.Http
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.model.headers.{`Retry-After`, RetryAfterDateTime, RetryAfterDuration}
 import akka.stream.ActorMaterializer
-import com.typesafe.scalalogging.StrictLogging
+import com.typesafe.scalalogging.{Logger, LoggerTakingImplicit}
+import org.slf4j.LoggerFactory
 
 import scala.collection.immutable
 import scala.concurrent.duration._
@@ -18,25 +19,50 @@ import scala.util.{Failure, Success, Try}
 class HttpClient(
     config: HttpClientConfig,
     gatewayType: String,
-    httpMetrics: HttpMetrics,
+    httpMetrics: HttpMetrics[String],
     retryConfig: RetryConfig,
     clock: Clock,
     awsRequestSigner: Option[AwsRequestSigner]
-)(implicit system: ActorSystem)
+)(
+    implicit system: ActorSystem
+) extends LoggingHttpClient[String](config, gatewayType, httpMetrics, retryConfig, clock, awsRequestSigner)(
+      system,
+      Logger.takingImplicit(LoggerFactory.getLogger(getClass.getName))((msg: String, _: String) => msg)
+    ) {
+  override def request(
+      method: HttpMethod,
+      entity: MessageEntity,
+      path: String,
+      headers: immutable.Seq[HttpHeader],
+      deadline: Deadline,
+      queryString: Option[String]
+  )(implicit executionContext: ExecutionContext, ctx: String = ""): Future[HttpClientResponse] =
+    super.request(method, entity, path, headers, deadline, queryString)
+}
+
+class LoggingHttpClient[LoggingContext](
+    config: HttpClientConfig,
+    gatewayType: String,
+    httpMetrics: HttpMetrics[LoggingContext],
+    retryConfig: RetryConfig,
+    clock: Clock,
+    awsRequestSigner: Option[AwsRequestSigner]
+)(implicit system: ActorSystem, logger: LoggerTakingImplicit[LoggingContext])
     extends HttpLayer(config, gatewayType, httpMetrics, retryConfig, clock, awsRequestSigner) {
   override protected def sendRequest: HttpRequest => Future[HttpResponse] = Http().singleRequest(_)
 }
 
-abstract class HttpLayer(
+abstract class HttpLayer[LoggingContext](
     config: HttpClientConfig,
     gatewayType: String,
-    httpMetrics: HttpMetrics,
+    httpMetrics: HttpMetrics[LoggingContext],
     retryConfig: RetryConfig,
     clock: Clock,
     awsRequestSigner: Option[AwsRequestSigner] = None
 )(
-    implicit system: ActorSystem
-) extends StrictLogging {
+    implicit system: ActorSystem,
+    logger: LoggerTakingImplicit[LoggingContext]
+) {
 
   protected def sendRequest: HttpRequest => Future[HttpResponse]
 
@@ -48,7 +74,8 @@ abstract class HttpLayer(
       deadline: Deadline,
       queryString: Option[String] = None
   )(
-      implicit executionContext: ExecutionContext
+      implicit executionContext: ExecutionContext,
+      ctx: LoggingContext
   ): Future[HttpClientResponse] =
     if (deadline.isOverdue()) {
       Future.successful(DeadlineExpired())
@@ -73,7 +100,8 @@ abstract class HttpLayer(
   }
 
   private[this] def executeRequest(request: HttpRequest, tryNumber: Int, deadline: Deadline)(
-      implicit ec: ExecutionContext
+      implicit ec: ExecutionContext,
+      ctx: LoggingContext
   ): Future[HttpClientResponse] =
     Future
       .successful(request)
@@ -87,7 +115,7 @@ abstract class HttpLayer(
 
   private[this] def handleResponse(tryNumber: Int, deadline: Deadline, httpRequest: HttpRequest)(
       response: HttpResponse
-  )(implicit ec: ExecutionContext): Future[HttpClientResponse] = response match {
+  )(implicit ec: ExecutionContext, ctx: LoggingContext): Future[HttpClientResponse] = response match {
     case response @ HttpResponse(StatusCodes.Success(_), _, _, _)                => Future.successful(HttpClientSuccess(response))
     case response @ HttpResponse(StatusCodes.BadRequest, _, HttpEntity.Empty, _) => Future.successful(HttpClientError(response))
     case response @ HttpResponse(StatusCodes.BadRequest, _, _, _)                => Future.successful(DomainError(response))
@@ -113,7 +141,8 @@ abstract class HttpLayer(
   }
 
   private[this] def retryWithConfig(tryNum: Int, request: HttpRequest, response: HttpResponse, deadline: Deadline)(
-      implicit ec: ExecutionContext
+      implicit ec: ExecutionContext,
+      ctx: LoggingContext
   ): Future[HttpClientResponse] =
     if (deadline.isOverdue()) {
       logger.info(s"[$gatewayType] Try #$tryNum: Deadline has expired.")
@@ -145,7 +174,8 @@ abstract class HttpLayer(
   }
 
   private[this] def handleErrors(tryNum: Int, deadline: Deadline, request: HttpRequest)(
-      implicit ec: ExecutionContext
+      implicit ec: ExecutionContext,
+      ctx: LoggingContext
   ): PartialFunction[Throwable, Future[HttpClientResponse]] = {
     case NonFatal(e) if tryNum <= retryConfig.retriesException =>
       val delay: FiniteDuration = calculateDelay(None, tryNum)
@@ -156,16 +186,16 @@ abstract class HttpLayer(
       Future.successful(DeadlineExpired(None))
   }
 
-  private[this] def logRequest[T]: PartialFunction[Try[HttpRequest], Unit] = {
+  private[this] def logRequest[T](implicit ctx: LoggingContext): PartialFunction[Try[HttpRequest], Unit] = {
     case Success(request) => logger.debug(s"[$gatewayType] Sending request to ${request.method.value} ${request.uri}.")
   }
 
-  private[this] def logRetryAfter: PartialFunction[Try[HttpResponse], Unit] = {
+  private[this] def logRetryAfter(implicit ctx: LoggingContext): PartialFunction[Try[HttpResponse], Unit] = {
     case Success(response) if response.header[`Retry-After`].isDefined =>
       logger.info(s"[$gatewayType] Received retry-after header with value ${response.header[`Retry-After`]}")
   }
 
-  private[this] def logResponse(request: HttpRequest): PartialFunction[Try[HttpResponse], Unit] = {
+  private[this] def logResponse(request: HttpRequest)(implicit ctx: LoggingContext): PartialFunction[Try[HttpResponse], Unit] = {
     case Success(response) =>
       httpMetrics.meterResponse(request.method, request.uri.path, response)
       logger.debug(s"[$gatewayType] Received response ${response.status} from ${request.method.value} ${request.uri}.")

src/main/scala/io/moia/scalaHttpClient/HttpMetrics.scala

@@ -2,6 +2,6 @@ package io.moia.scalaHttpClient
 
 import akka.http.scaladsl.model.{HttpMethod, HttpResponse, Uri}
 
-trait HttpMetrics {
-  def meterResponse(method: HttpMethod, path: Uri.Path, response: HttpResponse): Unit
+trait HttpMetrics[LoggingContext] {
+  def meterResponse(method: HttpMethod, path: Uri.Path, response: HttpResponse)(implicit ctx: LoggingContext): Unit
 }

src/test/scala/io/moia/scalaHttpClient/HttpClientTest.scala

@@ -1,39 +1,17 @@
 package io.moia.scalaHttpClient
 
-import java.time.Clock
-
-import akka.actor.ActorSystem
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.model.headers.{ModeledCustomHeader, ModeledCustomHeaderCompanion}
-import akka.stream.ActorMaterializer
+import com.typesafe.scalalogging.StrictLogging
 import io.moia.scalaHttpClient.AwsRequestSigner.AwsRequestSignerConfig
-import org.scalatest.matchers.should.Matchers
-import org.scalatest.wordspec.AnyWordSpecLike
 import org.scalatest.{Inside, Inspectors}
 
 import scala.collection.immutable
 import scala.concurrent.duration._
-import scala.concurrent.{ExecutionContext, Future, Promise}
+import scala.concurrent.{Future, Promise}
 import scala.util.Try
 
-class HttpClientTest extends AnyWordSpecLike with Matchers with FutureValues with Inside {
-
-  private implicit val system: ActorSystem                = ActorSystem("test")
-  private implicit val mat: ActorMaterializer             = ActorMaterializer()
-  private implicit val executionContext: ExecutionContext = system.dispatcher
-  private val httpClientConfig: HttpClientConfig          = HttpClientConfig("http", isSecureConnection = false, "127.0.0.1", 8888)
-  private val clock: Clock                                = Clock.systemUTC()
-  private val httpMetrics: HttpMetrics                    = (_: HttpMethod, _: Uri.Path, _: HttpResponse) => ()
-  private val retryConfig: RetryConfig =
-    RetryConfig(
-      retriesTooManyRequests = 2,
-      retriesServiceUnavailable = 0,
-      retriesRequestTimeout = 0,
-      retriesServerError = 0,
-      retriesException = 3,
-      initialBackoff = 10.millis,
-      strictifyResponseTimeout = 1.second
-    )
+class HttpClientTest extends TestSetup with Inside with StrictLogging {
 
   classOf[HttpClient].getSimpleName should {
     "sign requests" in {

src/test/scala/io/moia/scalaHttpClient/LoggingHttpClientTest.scala

@@ -0,0 +1,54 @@
+package io.moia.scalaHttpClient
+
+import akka.http.scaladsl.model._
+import com.typesafe.scalalogging.{CanLog, Logger, LoggerTakingImplicit}
+import org.slf4j.{LoggerFactory, MDC}
+
+import scala.collection.immutable
+import scala.concurrent.Future
+import scala.concurrent.duration._
+
+class LoggingHttpClientTest extends TestSetup {
+
+  val typedHttpMetrics: HttpMetrics[LoggingContext] = new HttpMetrics[LoggingContext] {
+    override def meterResponse(method: HttpMethod, path: Uri.Path, response: HttpResponse)(implicit ctx: LoggingContext): Unit = ()
+  }
+
+  case class LoggingContext(context: String)
+
+  private implicit val canLogString: CanLog[LoggingContext] = new CanLog[LoggingContext] {
+    val prevContext: String = MDC.get("context")
+
+    override def logMessage(originalMsg: String, ctx: LoggingContext): String = {
+      // you can put things to the MDC here
+      MDC.put("context", ctx.context)
+
+      // …or influence the string that is going to be logged:
+      s"$originalMsg for (${ctx.context})"
+    }
+    override def afterLog(ctx: LoggingContext): Unit =
+      if (prevContext != null) {
+        MDC.put("context", prevContext)
+      } else MDC.remove("context")
+  }
+
+  private implicit val theLogger: LoggerTakingImplicit[LoggingContext] = Logger.takingImplicit(LoggerFactory.getLogger(getClass.getName))
+  private implicit val ctx: LoggingContext                             = LoggingContext("Logging Context")
+
+  classOf[LoggingHttpClient[LoggingContext]].getSimpleName should {
+    "take a customer logger" in {
+      // given
+      val testHttpClient =
+        new LoggingHttpClient[LoggingContext](httpClientConfig, "TestGateway", typedHttpMetrics, retryConfig, clock, None) {
+          override def sendRequest: HttpRequest => Future[HttpResponse] = (_: HttpRequest) => Future.successful(HttpResponse())
+        }
+
+      // when
+      val _ =
+        testHttpClient.request(HttpMethods.POST, HttpEntity.Empty, "/test", immutable.Seq.empty, Deadline.now + 10.seconds).futureValue
+
+      // then succeed if it compiles
+      succeed
+    }
+  }
+}

src/test/scala/io/moia/scalaHttpClient/MockServer.scala

@@ -0,0 +1,44 @@
+package io.moia.scalaHttpClient
+
+import org.mockserver.integration.ClientAndServer
+import org.mockserver.integration.ClientAndServer.startClientAndServer
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, Suite}
+
+trait MockServer extends BeforeAndAfterEach with BeforeAndAfterAll { self: Suite =>
+
+  val mockServerPort = 38080
+
+  val mockServerHttpClientConfig: HttpClientConfig =
+    HttpClientConfig(
+      scheme = "http",
+      isSecureConnection = false,
+      host = "127.0.0.1",
+      port = mockServerPort
+    )
+
+  private var clientAndServer: Option[ClientAndServer] = None
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    clientAndServer.getOrElse {
+      val newServerMock = startClientAndServer(mockServerPort)
+      clientAndServer = Some(newServerMock)
+      newServerMock
+    }
+    ()
+  }
+
+  override def afterEach(): Unit = {
+    super.afterEach()
+    clientAndServer.foreach(_.reset())
+  }
+
+  override def afterAll(): Unit = {
+    super.afterAll()
+    clientAndServer.foreach(_.stop())
+    clientAndServer = None
+  }
+
+  def getClientAndServer: ClientAndServer =
+    clientAndServer.getOrElse(throw new IllegalStateException("Mock not started yet!"))
+}