BDOG-1512 Adds HttpClient2 to support streams, withProxy etc.

CHANGELOG.md

@@ -1,3 +1,23 @@
+## Version 13.13.0
+
+### Auditing max body length
+
+Payloads will be truncated in audit logs if they exceed the max supported (as configured by `http-verbs.auditing.maxBodyLength`).
+
+### WSProxyConfiguration
+
+`WSProxyConfiguration.apply` has been deprecated, use `WSProxyConfiguration.buildWsProxyServer` instead.
+
+There are some differences with `WSProxyConfiguration.buildWsProxyServer`:
+  * configPrefix is fixed to `proxy`.
+  * `proxy.proxyRequiredForThisEnvironment` has been replaced with `proxy.enabled`, but note, it defaults to false (rather than true). This is appropriate for development and tests, but will need explicitly enabling when deployed.
+
+
+### Adds HttpClient2
+This is in addition to `HttpClient` (for now), so can be optionally used instead.
+
+See [README](/README.md) for details.
+
 ## Version 13.12.0
 
 ### Supported Play Versions

README.md

@@ -33,19 +33,97 @@ Where `play-xx` is your version of Play (e.g. `play-28`).
 
 ## Usage
 
+There are two HttpClients available.
+
+### uk.gov.hmrc.http.HttpClient
+
 Examples can be found [here](https://github.com/hmrc/http-verbs/blob/master/http-verbs-test-common/src/test/scala/uk/gov/hmrc/http/examples/Examples.scala)
 
-### URLs
+URLs can be supplied as either `java.net.URL` or `String`. We recommend supplying `java.net.URL` and using the provided [URL interpolator](#url-interpolator) for correct escaping of query and path parameters.
+
+
+### uk.gov.hmrc.http.client2.HttpClient2
+
+This client follows the same patterns as `HttpClient` - that is, it also requires a `HeaderCarrier` to represent the context of the caller, and an `HttpReads` to process the http response.
+
+In addition, it:
+- Supports streaming
+- Exposes the underlying `play.api.libs.ws.WSRequest` with `transform`, making it easier to customise the request.
+- Only accepts the URL as `java.net.URL`; you can make use of the provided [URL interpolator](#url-interpolator).
+
+Examples can be found in [here](/http-verbs-common/src/test/scala/uk/gov/hmrc/http/client2/HttpClient2Spec.scala)
+
+To migrate:
+
+```scala
+httpClient.GET[ResponseType](url)
+```
+
+becomes
+
+```scala
+httpClient2.get(url"$url").execute[ResponseType]
+```
+
+and
+
+```scala
+httpClient.POST[ResponseType](url, payload, headers)
+```
+
+becomes
+
+```scala
+httpClient2.post(url"$url").withBody(Json.toJson(payload)).addHeaders(headers).execute[ResponseType]
+```
+
 
-URLs can be supplied as either `java.net.URL` or `String`. We recommend supplying `java.net.URL` for correct escaping of query and path parameters. A [URL interpolator](https://sttp.softwaremill.com/en/latest/model/uri.html) has been provided for convenience.
+#### Header manipulation
 
+With `HttpClient`, replacing a header can require providing a customised client implementation (e.g. to replace the user-agent header), or updating the `HeaderCarrier` (e.g. to replace the authorisation header). This can now all be done with the `replaceHeader` on `HttpClient2` per call. e.g.
+
+```scala
+httpClient2.get(url"$url").replaceHeader("User-Agent" -> userAgent).replaceHeader("Authorization" -> authorization).execute[ResponseType]
+```
+
+#### Using proxy
+
+With `HttpClient`, to use a proxy requires creating a new instance of HttpClient to mix in `WSProxy` and configure. With `HttpClient2` this can be done with the same client, calling `withProxy` per call. e.g.
+
+```scala
+httpClient2.get(url"$url").withProxy.execute[ResponseType]
+```
+
+* It uses `WSProxyConfiguration.buildWsProxyServer` which needs enabling with `proxy.enabled` in configuration, which by default is `false`, for development. See [WSProxyConfiguration](CHANGELOG.md#wsproxyconfiguration) for configuration changes.
+
+#### Streaming
+
+Streaming is supported with `HttpClient2`, and will be audited in the same way as `HttpClient`. Note that payloads will be truncated in audit logs if they exceed the max supported (as configured by `http-verbs.auditing.maxBodyLength`).
+
+Streamed requests can simply be passed to `withBody`:
+
+```scala
+val reqStream: Source[ByteString, _] = ???
+httpClient2.post(url"$url").withBody(reqStream).execute[ResponseType]
+```
+
+For streamed responses, use `stream` rather than `execute`:
+
+```scala
+httpClient2.get(url"$url").stream[Source[ByteString, _]]
+```
+
+### URL interpolator
+
+A [URL interpolator](https://sttp.softwaremill.com/en/latest/model/uri.html) has been provided to help with escaping query and parameters correctly.
 
 ```scala
 import uk.gov.hmrc.http.StringContextOps
 
 url"http://localhost:8080/users/${user.id}?email=${user.email}"
 ```
 
+
 ### Headers
 
 #### Creating HeaderCarrier
@@ -120,7 +198,7 @@ class MyConnectorSpec extends WireMockSupport with GuiceOneAppPerSuite {
 }
 ```
 
-The `HttpClientSupport` trait can provide an instance of HttpClient as an alternative to instanciating the application:
+The `HttpClientSupport` trait can provide an instance of `HttpClient` as an alternative to instanciating the application:
 ```scala
 class MyConnectorSpec extends WireMockSupport with HttpClientSupport {
   private val connector = new MyConnector(
@@ -130,6 +208,8 @@ class MyConnectorSpec extends WireMockSupport with HttpClientSupport {
 }
 ```
 
+Similarly `HttpClient2Support` can be used to provide an instance of `HttpClient2`.
+
 The `ExternalWireMockSupport` trait is an alternative to `WireMockSupport` which uses `127.0.0.1` instead of `localhost` for the hostname which is treated as an external host for header forwarding rules. This should be used for tests of connectors which call endpoints external to the platform. The variable `externalWireMockHost` (or `externalWireMockUrl`) should be used to provide the hostname in configuration.
 
 Both `WireMockSupport` and `ExternalWireMockSupport` can be used together for integration tests if required.

build.sbt

@@ -59,6 +59,7 @@ lazy val sharedSources =
   shareSources("http-verbs-common")
 
 lazy val httpVerbsPlay28 = Project("http-verbs-play-28", file("http-verbs-play-28"))
+  .enablePlugins(BuildInfoPlugin)
   .settings(
     commonSettings,
     sharedSources,
@@ -70,6 +71,10 @@ lazy val httpVerbsPlay28 = Project("http-verbs-play-28", file("http-verbs-play-2
       AppDependencies.coreTestPlay28,
     Test / fork := true // akka is not unloaded properly, which can affect other tests
   )
+  .settings( // https://github.com/sbt/sbt-buildinfo
+    buildInfoKeys := Seq[BuildInfoKey](version),
+    buildInfoPackage := "uk.gov.hmrc.http"
+   )
   .dependsOn(httpVerbs)
 
 lazy val sharedTestSources =

http-verbs-common/src/main/resources/reference.conf

@@ -14,7 +14,11 @@
 
 internalServiceHostPatterns = [ "^.*\\.service$", "^.*\\.mdtp$", "^localhost$" ]
 bootstrap.http.headersAllowlist = []
-http-verbs.retries {
-  intervals = [ "500.millis", "1.second", "2.seconds", "4.seconds", "8.seconds" ]
-  ssl-engine-closed-already.enabled = false
+http-verbs {
+  retries {
+    intervals = [ "500.millis", "1.second", "2.seconds", "4.seconds", "8.seconds" ]
+    ssl-engine-closed-already.enabled = false
+  }
+  auditing.maxBodyLength = 32665
 }
+proxy.enabled = false

http-verbs-common/src/main/scala/uk/gov/hmrc/http/HttpDelete.scala

@@ -34,8 +34,8 @@ trait HttpDelete
 
   override def DELETE[O](url: String, headers: Seq[(String, String)] = Seq.empty)(implicit rds: HttpReads[O], hc: HeaderCarrier, ec: ExecutionContext): Future[O] =
     withTracing(DELETE_VERB, url) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(DELETE_VERB, url)(doDelete(url, allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(DELETE_VERB, url)(doDelete(url, allHeaders))
       executeHooks(DELETE_VERB, url"$url", allHeaders, None, httpResponse)
       mapErrors(DELETE_VERB, url, httpResponse).map(rds.read(DELETE_VERB, url, _))
     }

http-verbs-common/src/main/scala/uk/gov/hmrc/http/HttpGet.scala

@@ -45,8 +45,8 @@ trait HttpGet extends CoreGet with GetHttpTransport with HttpVerb with Connectio
     val urlWithQuery = url + makeQueryString(queryParams)
 
     withTracing(GET_VERB, urlWithQuery) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(GET_VERB, urlWithQuery)(doGet(urlWithQuery, headers = allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(GET_VERB, urlWithQuery)(doGet(urlWithQuery, headers = allHeaders))
       executeHooks(GET_VERB, url"$url", allHeaders, None, httpResponse)
       mapErrors(GET_VERB, urlWithQuery, httpResponse).map(response => rds.read(GET_VERB, urlWithQuery, response))
     }

http-verbs-common/src/main/scala/uk/gov/hmrc/http/HttpPatch.scala

@@ -42,8 +42,8 @@ trait HttpPatch
       hc: HeaderCarrier,
       ec: ExecutionContext): Future[O] =
     withTracing(PATCH_VERB, url) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(PATCH_VERB, url)(doPatch(url, body, allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(PATCH_VERB, url)(doPatch(url, body, allHeaders))
       executeHooks(PATCH_VERB, url"$url", allHeaders, Option(HookData.FromString(Json.stringify(wts.writes(body)))), httpResponse)
       mapErrors(PATCH_VERB, url, httpResponse).map(response => rds.read(PATCH_VERB, url, response))
     }

http-verbs-common/src/main/scala/uk/gov/hmrc/http/HttpPost.scala

@@ -42,8 +42,8 @@ trait HttpPost
       hc: HeaderCarrier,
       ec: ExecutionContext): Future[O] =
     withTracing(POST_VERB, url) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(POST_VERB, url)(doPost(url, body, allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(POST_VERB, url)(doPost(url, body, allHeaders))
       executeHooks(POST_VERB, url"$url", allHeaders, Option(HookData.FromString(Json.stringify(wts.writes(body)))), httpResponse)
       mapErrors(POST_VERB, url, httpResponse).map(rds.read(POST_VERB, url, _))
     }
@@ -56,8 +56,8 @@ trait HttpPost
       hc: HeaderCarrier,
       ec: ExecutionContext): Future[O] =
     withTracing(POST_VERB, url) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(POST_VERB, url)(doPostString(url, body, allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(POST_VERB, url)(doPostString(url, body, allHeaders))
       executeHooks(POST_VERB, url"$url", allHeaders, Option(HookData.FromString(body)), httpResponse)
       mapErrors(POST_VERB, url, httpResponse).map(rds.read(POST_VERB, url, _))
     }
@@ -70,8 +70,8 @@ trait HttpPost
       hc: HeaderCarrier,
       ec: ExecutionContext): Future[O] =
     withTracing(POST_VERB, url) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(POST_VERB, url)(doFormPost(url, body, allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(POST_VERB, url)(doFormPost(url, body, allHeaders))
       executeHooks(POST_VERB, url"$url", allHeaders, Option(HookData.FromMap(body)), httpResponse)
       mapErrors(POST_VERB, url, httpResponse).map(rds.read(POST_VERB, url, _))
     }
@@ -83,8 +83,8 @@ trait HttpPost
       hc: HeaderCarrier,
       ec: ExecutionContext): Future[O] =
     withTracing(POST_VERB, url) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(POST_VERB, url)(doEmptyPost(url, allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(POST_VERB, url)(doEmptyPost(url, allHeaders))
       executeHooks(POST_VERB, url"$url", allHeaders, None, httpResponse)
       mapErrors(POST_VERB, url, httpResponse).map(rds.read(POST_VERB, url, _))
     }

http-verbs-common/src/main/scala/uk/gov/hmrc/http/HttpPut.scala

@@ -36,8 +36,8 @@ trait HttpPut extends CorePut with PutHttpTransport with HttpVerb with Connectio
       hc: HeaderCarrier,
       ec: ExecutionContext): Future[O] =
     withTracing(PUT_VERB, url) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(PUT_VERB, url)(doPut(url, body, allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(PUT_VERB, url)(doPut(url, body, allHeaders))
       executeHooks(PUT_VERB, url"$url", allHeaders, Option(HookData.FromString(Json.stringify(wts.writes(body)))), httpResponse)
       mapErrors(PUT_VERB, url, httpResponse).map(response => rds.read(PUT_VERB, url, response))
     }
@@ -50,8 +50,8 @@ trait HttpPut extends CorePut with PutHttpTransport with HttpVerb with Connectio
       hc: HeaderCarrier,
       ec: ExecutionContext): Future[O] =
     withTracing(PUT_VERB, url) {
-      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers)
-      val httpResponse = retry(PUT_VERB, url)(doPutString(url, body, allHeaders))
+      val allHeaders = HeaderCarrier.headersForUrl(hcConfig, url, headers) :+ "Http-Client-Version" -> BuildInfo.version
+      val httpResponse = retryOnSslEngineClosed(PUT_VERB, url)(doPutString(url, body, allHeaders))
       executeHooks(PUT_VERB, url"$url", allHeaders, Option(HookData.FromString(body)), httpResponse)
       mapErrors(PUT_VERB, url, httpResponse).map(rds.read(PUT_VERB, url, _))
     }

http-verbs-common/src/main/scala/uk/gov/hmrc/http/HttpReadsLegacyInstances.scala

@@ -17,8 +17,7 @@
 package uk.gov.hmrc.http
 
 import com.github.ghik.silencer.silent
-import play.api.libs.json
-import play.api.libs.json.{JsNull, JsValue}
+import play.api.libs.json.{JsNull, JsValue, Reads}
 
 
 trait HttpReadsLegacyInstances extends HttpReadsLegacyOption with HttpReadsLegacyJson
@@ -45,21 +44,21 @@ trait HttpReadsLegacyOption extends HttpErrorFunctions {
 
 trait HttpReadsLegacyJson extends HttpErrorFunctions {
   @deprecated("Use uk.gov.hmrc.http.HttpReads.Implicits instead. See README for differences.", "11.0.0")
-  implicit def readFromJson[O](implicit rds: json.Reads[O], mf: Manifest[O]): HttpReads[O] = new HttpReads[O] {
+  implicit def readFromJson[O](implicit rds: Reads[O], mf: Manifest[O]): HttpReads[O] = new HttpReads[O] {
     def read(method: String, url: String, response: HttpResponse) =
       readJson(method, url, handleResponse(method, url)(response).json)
   }
 
   @deprecated("Use uk.gov.hmrc.http.HttpReads.Implicits instead. See README for differences.", "11.0.0")
-  def readSeqFromJsonProperty[O](name: String)(implicit rds: json.Reads[O], mf: Manifest[O]) = new HttpReads[Seq[O]] {
+  def readSeqFromJsonProperty[O](name: String)(implicit rds: Reads[O], mf: Manifest[O]) = new HttpReads[Seq[O]] {
     def read(method: String, url: String, response: HttpResponse) = response.status match {
       case 204 | 404 => Seq.empty
       case _ =>
         readJson[Seq[O]](method, url, (handleResponse(method, url)(response).json \ name).getOrElse(JsNull)) //Added JsNull here to force validate to fail - replicates existing behaviour
     }
   }
 
-  private def readJson[A](method: String, url: String, jsValue: JsValue)(implicit rds: json.Reads[A], mf: Manifest[A]) =
+  private def readJson[A](method: String, url: String, jsValue: JsValue)(implicit rds: Reads[A], mf: Manifest[A]) =
     jsValue
       .validate[A]
       .fold(

http-verbs-common/src/main/scala/uk/gov/hmrc/http/HttpResponse.scala

@@ -16,6 +16,8 @@
 
 package uk.gov.hmrc.http
 
+import akka.stream.scaladsl.Source
+import akka.util.ByteString
 import com.github.ghik.silencer.silent
 import play.api.libs.json.{JsValue, Json}
 
@@ -44,6 +46,9 @@ trait HttpResponse {
   def json: JsValue =
     Json.parse(body)
 
+  def bodyAsSource: Source[ByteString, _] =
+    Source.single(ByteString(body))
+
   def header(key: String): Option[String] =
     headers.get(key).flatMap(_.headOption)
 
@@ -106,6 +111,22 @@ object HttpResponse {
     }
   }
 
+  def apply(
+    status      : Int,
+    bodyAsSource: Source[ByteString, _],
+    headers     : Map[String, Seq[String]]
+  ): HttpResponse = {
+    val pStatus       = status
+    val pBodyAsSource = bodyAsSource
+    val pHeaders      = headers
+    new HttpResponse {
+      override def status      : Int                      = pStatus
+      override def body        : String                   = sys.error(s"This is a streamed response, please use `bodyAsSource`")
+      override def bodyAsSource: Source[ByteString, _]    = pBodyAsSource
+      override def allHeaders  : Map[String, Seq[String]] = pHeaders
+    }
+  }
+
   def unapply(that: HttpResponse): Option[(Int, String, Map[String, Seq[String]])] =
     Some((that.status, that.body, that.headers))
 }

http-verbs-common/src/main/scala/uk/gov/hmrc/http/Retries.scala

@@ -35,19 +35,32 @@ trait Retries {
 
   private val logger = LoggerFactory.getLogger("application")
 
-  def retry[A](verb: String, url: String)(block: => Future[A])(implicit ec: ExecutionContext): Future[A] = {
+  def retryOnSslEngineClosed[A](verb: String, url: String)(block: => Future[A])(implicit ec: ExecutionContext): Future[A] =
+    retryFor(s"$verb $url") { case ex @ `sslEngineClosedMatcher`() => true }(block)
+
+  @deprecated("Use retryOnSslEngineClosed instead", "14.0.0")
+  def retry[A](verb: String, url: String)(block: => Future[A])(implicit ec: ExecutionContext): Future[A] =
+    retryOnSslEngineClosed(verb, url)(block)
+
+  def retryFor[A](
+     label    : String
+  )( condition: PartialFunction[Exception, Boolean]
+  )( block    : => Future[A]
+  )(implicit
+    ec: ExecutionContext
+  ): Future[A] = {
     def loop(remainingIntervals: Seq[FiniteDuration]): Future[A] = {
       // scheduling will loose MDC data. Here we explicitly ensure it is available on block.
       block
         .recoverWith {
-          case ex @ `sslEngineClosedMatcher`() if remainingIntervals.nonEmpty =>
+          case ex: Exception if condition.lift(ex).getOrElse(false) && remainingIntervals.nonEmpty =>
             val delay = remainingIntervals.head
-            logger.warn(s"Retrying $verb $url in $delay due to '${ex.getMessage}' error")
-              val mdcData = Mdc.mdcData
-              after(delay, actorSystem.scheduler){
-                Mdc.putMdc(mdcData)
-                loop(remainingIntervals.tail)
-              }
+            logger.warn(s"Retrying $label in $delay due to error: ${ex.getMessage}")
+            val mdcData = Mdc.mdcData
+            after(delay, actorSystem.scheduler){
+              Mdc.putMdc(mdcData)
+              loop(remainingIntervals.tail)
+            }
         }
       }
     loop(intervals)