1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
|
package xyz.driver.core.rest
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.headers.`User-Agent`
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, Uri}
import akka.http.scaladsl.settings.{ClientConnectionSettings, ConnectionPoolSettings}
import akka.stream.scaladsl.{Keep, Sink, Source}
import akka.stream.{ActorMaterializer, OverflowStrategy, QueueOfferResult, ThrottleMode}
import xyz.driver.core.Name
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.concurrent.duration._
import scala.util.{Failure, Success}
class PooledHttpClient(
baseUri: Uri,
applicationName: Name[App],
applicationVersion: String,
requestRateLimit: Int = 64,
requestQueueSize: Int = 1024)(implicit actorSystem: ActorSystem, executionContext: ExecutionContext)
extends HttpClient {
private val host = baseUri.authority.host.toString()
private val port = baseUri.effectivePort
private val scheme = baseUri.scheme
protected implicit val materializer: ActorMaterializer = ActorMaterializer()(actorSystem)
private val clientConnectionSettings: ClientConnectionSettings =
ClientConnectionSettings(actorSystem).withUserAgentHeader(
Option(`User-Agent`(applicationName.value + "/" + applicationVersion)))
private val connectionPoolSettings: ConnectionPoolSettings = ConnectionPoolSettings(actorSystem)
.withConnectionSettings(clientConnectionSettings)
private val pool = if (scheme.equalsIgnoreCase("https")) {
Http().cachedHostConnectionPoolHttps[Promise[HttpResponse]](host, port, settings = connectionPoolSettings)
} else {
Http().cachedHostConnectionPool[Promise[HttpResponse]](host, port, settings = connectionPoolSettings)
}
private val queue = Source
.queue[(HttpRequest, Promise[HttpResponse])](requestQueueSize, OverflowStrategy.dropNew)
.via(pool)
.throttle(requestRateLimit, 1.second, maximumBurst = requestRateLimit, ThrottleMode.shaping)
.toMat(Sink.foreach({
case ((Success(resp), p)) => p.success(resp)
case ((Failure(e), p)) => p.failure(e)
}))(Keep.left)
.run
def makeRequest(request: HttpRequest): Future[HttpResponse] = {
val responsePromise = Promise[HttpResponse]()
queue.offer(request -> responsePromise).flatMap {
case QueueOfferResult.Enqueued =>
responsePromise.future
case QueueOfferResult.Dropped =>
Future.failed(new Exception(s"Request queue to the host $host is overflown"))
case QueueOfferResult.Failure(ex) =>
Future.failed(ex)
case QueueOfferResult.QueueClosed =>
Future.failed(new Exception("Queue was closed (pool shut down) while running the request"))
}
}
}
|
