This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 3cb1ed8c4 Allow using SdkHttpConfigurationOption over default
pekko-http connection settings (#827)
3cb1ed8c4 is described below
commit 3cb1ed8c479405cc94a758bee3edd694fb8ec45b
Author: João Ferreira <[email protected]>
AuthorDate: Thu Sep 26 14:02:39 2024 +0100
Allow using SdkHttpConfigurationOption over default pekko-http connection
settings (#827)
* withConnectionPoolSettingsBuilderFromAttributeMap
* scalafmt tests
* remove comment
* reorganize imports
* DurationConverters scala 2.12
---
.../stream/connectors/awsspi/PekkoHttpClient.scala | 44 +++++++++--
.../connectors/awsspi/PekkoHttpClientSpec.scala | 88 ++++++++++++++++++++++
2 files changed, 124 insertions(+), 8 deletions(-)
diff --git
a/aws-spi-pekko-http/src/main/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClient.scala
b/aws-spi-pekko-http/src/main/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClient.scala
index 9aacfd2b5..df57812e2 100644
---
a/aws-spi-pekko-http/src/main/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClient.scala
+++
b/aws-spi-pekko-http/src/main/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClient.scala
@@ -32,17 +32,19 @@ import pekko.http.scaladsl.settings.ConnectionPoolSettings
import pekko.stream.scaladsl.Source
import pekko.stream.{ Materializer, SystemMaterializer }
import pekko.util.ByteString
-import pekko.util.OptionConverters
+import pekko.util.OptionConverters._
+import pekko.util.JavaDurationConverters._
import org.slf4j.LoggerFactory
import software.amazon.awssdk.http.async._
-import software.amazon.awssdk.http.SdkHttpRequest
+import software.amazon.awssdk.http.{ SdkHttpConfigurationOption,
SdkHttpRequest }
import software.amazon.awssdk.utils.AttributeMap
import scala.collection.immutable
import scala.concurrent.duration.Duration
import scala.concurrent.{ Await, ExecutionContext }
-class PekkoHttpClient(shutdownHandle: () => Unit, connectionSettings:
ConnectionPoolSettings)(implicit
+class PekkoHttpClient(shutdownHandle: () => Unit, private[awsspi] val
connectionSettings: ConnectionPoolSettings)(
+ implicit
actorSystem: ActorSystem,
ec: ExecutionContext,
mat: Materializer) extends SdkAsyncHttpClient {
@@ -84,8 +86,7 @@ object PekkoHttpClient {
contentType: ContentType,
contentPublisher: SdkHttpContentPublisher): RequestEntity =
method.requestEntityAcceptance match {
- case Expected =>
- OptionConverters.toScala(contentPublisher.contentLength()) match {
+ case Expected => contentPublisher.contentLength().toScala match {
case Some(length) =>
HttpEntity(contentType, length,
Source.fromPublisher(contentPublisher).map(ByteString(_)))
case None => HttpEntity(contentType,
Source.fromPublisher(contentPublisher).map(ByteString(_)))
@@ -151,18 +152,39 @@ object PekkoHttpClient {
else throw new RuntimeException(s"Could not parse custom content type
'$contentTypeStr'.")
}
+ private[awsspi] def buildConnectionPoolSettings(
+ base: ConnectionPoolSettings, attributeMap: AttributeMap):
ConnectionPoolSettings = {
+ def zeroToInfinite(duration: java.time.Duration):
scala.concurrent.duration.Duration =
+ if (duration.isZero) scala.concurrent.duration.Duration.Inf
+ else duration.asScala
+
+ base
+ .withUpdatedConnectionSettings(s =>
+
s.withConnectingTimeout(attributeMap.get(SdkHttpConfigurationOption.CONNECTION_TIMEOUT).asScala)
+
.withIdleTimeout(attributeMap.get(SdkHttpConfigurationOption.CONNECTION_MAX_IDLE_TIMEOUT).asScala))
+
.withMaxConnections(attributeMap.get(SdkHttpConfigurationOption.MAX_CONNECTIONS).intValue())
+
.withMaxConnectionLifetime(zeroToInfinite(attributeMap.get(SdkHttpConfigurationOption.CONNECTION_TIME_TO_LIVE)))
+ }
+
def builder() = PekkoHttpClientBuilder()
case class PekkoHttpClientBuilder(private val actorSystem:
Option[ActorSystem] = None,
private val executionContext: Option[ExecutionContext] = None,
- private val connectionPoolSettings: Option[ConnectionPoolSettings] =
None)
+ private val connectionPoolSettings: Option[ConnectionPoolSettings] =
None,
+ private val connectionPoolSettingsBuilder: (ConnectionPoolSettings,
AttributeMap) => ConnectionPoolSettings =
+ (c, _) => c)
extends SdkAsyncHttpClient.Builder[PekkoHttpClientBuilder] {
- def buildWithDefaults(attributeMap: AttributeMap): SdkAsyncHttpClient = {
+ def buildWithDefaults(serviceDefaults: AttributeMap): SdkAsyncHttpClient =
{
implicit val as = actorSystem.getOrElse(ActorSystem("aws-pekko-http"))
implicit val ec = executionContext.getOrElse(as.dispatcher)
val mat: Materializer = SystemMaterializer(as).materializer
- val cps = connectionPoolSettings.getOrElse(ConnectionPoolSettings(as))
+ val resolvedOptions =
serviceDefaults.merge(SdkHttpConfigurationOption.GLOBAL_HTTP_DEFAULTS);
+
+ val cps = connectionPoolSettingsBuilder(
+ connectionPoolSettings.getOrElse(ConnectionPoolSettings(as)),
+ resolvedOptions
+ )
val shutdownhandleF = () => {
if (actorSystem.isEmpty) {
Await.result(Http().shutdownAllConnectionPools().flatMap(_ =>
as.terminate()),
@@ -179,6 +201,12 @@ object PekkoHttpClient {
copy(executionContext = Some(executionContext))
def withConnectionPoolSettings(connectionPoolSettings:
ConnectionPoolSettings): PekkoHttpClientBuilder =
copy(connectionPoolSettings = Some(connectionPoolSettings))
+ def withConnectionPoolSettingsBuilder(
+ connectionPoolSettingsBuilder: (ConnectionPoolSettings, AttributeMap)
=> ConnectionPoolSettings
+ ): PekkoHttpClientBuilder =
+ copy(connectionPoolSettingsBuilder = connectionPoolSettingsBuilder)
+ def withConnectionPoolSettingsBuilderFromAttributeMap():
PekkoHttpClientBuilder =
+ copy(connectionPoolSettingsBuilder = buildConnectionPoolSettings)
}
lazy val xAmzJson = ContentType(MediaType.customBinary("application",
"x-amz-json-1.0", Compressible))
diff --git
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClientSpec.scala
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClientSpec.scala
index 6046aa1b6..3ae36ed21 100644
---
a/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClientSpec.scala
+++
b/aws-spi-pekko-http/src/test/scala/org/apache/pekko/stream/connectors/awsspi/PekkoHttpClientSpec.scala
@@ -18,13 +18,20 @@
package org.apache.pekko.stream.connectors.awsspi
import java.util.Collections
+import com.typesafe.config.ConfigFactory
import org.apache.pekko
import pekko.http.scaladsl.model.headers.`Content-Type`
import pekko.http.scaladsl.model.MediaTypes
+import pekko.http.scaladsl.settings.{ ClientConnectionSettings,
ConnectionPoolSettings }
+import pekko.util.JavaDurationConverters._
import org.scalatest.OptionValues
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec
+import software.amazon.awssdk.http.SdkHttpConfigurationOption
+import software.amazon.awssdk.utils.AttributeMap
+
+import scala.concurrent.duration._
class PekkoHttpClientSpec extends AnyWordSpec with Matchers with OptionValues {
@@ -47,5 +54,86 @@ class PekkoHttpClientSpec extends AnyWordSpec with Matchers
with OptionValues {
contentTypeHeader.value.lowercaseName() shouldBe
`Content-Type`.lowercaseName
reqHeaders should have size 1
}
+ "build() should use default ConnectionPoolSettings" in {
+ val pekkoClient: PekkoHttpClient = new
PekkoHttpAsyncHttpService().createAsyncHttpClientFactory()
+ .build()
+ .asInstanceOf[PekkoHttpClient]
+
+ pekkoClient.connectionSettings shouldBe
ConnectionPoolSettings(ConfigFactory.load())
+ }
+
+ "withConnectionPoolSettingsBuilderFromAttributeMap().buildWithDefaults()
should propagate configuration options" in {
+ val attributeMap = AttributeMap.builder()
+ .put(SdkHttpConfigurationOption.CONNECTION_TIMEOUT, 1.second.asJava)
+ .put(SdkHttpConfigurationOption.CONNECTION_MAX_IDLE_TIMEOUT,
2.second.asJava)
+ .put(SdkHttpConfigurationOption.MAX_CONNECTIONS, Integer.valueOf(3))
+ .put(SdkHttpConfigurationOption.CONNECTION_TIME_TO_LIVE,
4.second.asJava)
+ .build()
+ val pekkoClient: PekkoHttpClient = new
PekkoHttpAsyncHttpService().createAsyncHttpClientFactory()
+ .withConnectionPoolSettingsBuilderFromAttributeMap()
+ .buildWithDefaults(attributeMap)
+ .asInstanceOf[PekkoHttpClient]
+
+ pekkoClient.connectionSettings.connectionSettings.connectingTimeout
shouldBe 1.second
+ pekkoClient.connectionSettings.connectionSettings.idleTimeout shouldBe
2.seconds
+ pekkoClient.connectionSettings.maxConnections shouldBe 3
+ pekkoClient.connectionSettings.maxConnectionLifetime shouldBe 4.seconds
+ }
+
+ "withConnectionPoolSettingsBuilderFromAttributeMap().build() should
fallback to GLOBAL_HTTP_DEFAULTS" in {
+ val pekkoClient: PekkoHttpClient = new
PekkoHttpAsyncHttpService().createAsyncHttpClientFactory()
+ .withConnectionPoolSettingsBuilderFromAttributeMap()
+ .build()
+ .asInstanceOf[PekkoHttpClient]
+
+ pekkoClient.connectionSettings.connectionSettings.connectingTimeout
shouldBe
+
SdkHttpConfigurationOption.GLOBAL_HTTP_DEFAULTS.get(SdkHttpConfigurationOption.CONNECTION_TIMEOUT).asScala
+ pekkoClient.connectionSettings.connectionSettings.idleTimeout shouldBe
+
SdkHttpConfigurationOption.GLOBAL_HTTP_DEFAULTS.get(SdkHttpConfigurationOption.CONNECTION_MAX_IDLE_TIMEOUT).asScala
+ pekkoClient.connectionSettings.maxConnections shouldBe
+
SdkHttpConfigurationOption.GLOBAL_HTTP_DEFAULTS.get(SdkHttpConfigurationOption.MAX_CONNECTIONS).intValue()
+ infiniteToZero(pekkoClient.connectionSettings.maxConnectionLifetime)
shouldBe
+
SdkHttpConfigurationOption.GLOBAL_HTTP_DEFAULTS.get(SdkHttpConfigurationOption.CONNECTION_TIME_TO_LIVE)
+ }
+
+ "withConnectionPoolSettingsBuilder().build() should use passed
connectionPoolSettings builder" in {
+ val connectionPoolSettings = ConnectionPoolSettings(ConfigFactory.load())
+ .withConnectionSettings(
+ ClientConnectionSettings(ConfigFactory.load())
+ .withConnectingTimeout(1.second)
+ .withIdleTimeout(2.seconds)
+ )
+ .withMaxConnections(3)
+ .withMaxConnectionLifetime(4.seconds)
+
+ val pekkoClient: PekkoHttpClient = new
PekkoHttpAsyncHttpService().createAsyncHttpClientFactory()
+ .withConnectionPoolSettingsBuilder((_, _) => connectionPoolSettings)
+ .build()
+ .asInstanceOf[PekkoHttpClient]
+
+ pekkoClient.connectionSettings shouldBe connectionPoolSettings
+ }
+
+ "withConnectionPoolSettings().build() should use passed
ConnectionPoolSettings" in {
+ val connectionPoolSettings = ConnectionPoolSettings(ConfigFactory.load())
+ .withConnectionSettings(
+ ClientConnectionSettings(ConfigFactory.load())
+ .withConnectingTimeout(1.second)
+ .withIdleTimeout(2.seconds)
+ )
+ .withMaxConnections(3)
+ .withMaxConnectionLifetime(4.seconds)
+ val pekkoClient: PekkoHttpClient = new
PekkoHttpAsyncHttpService().createAsyncHttpClientFactory()
+ .withConnectionPoolSettings(connectionPoolSettings)
+ .build()
+ .asInstanceOf[PekkoHttpClient]
+
+ pekkoClient.connectionSettings shouldBe connectionPoolSettings
+ }
+ }
+
+ private def infiniteToZero(duration: scala.concurrent.duration.Duration):
java.time.Duration = duration match {
+ case _: scala.concurrent.duration.Duration.Infinite =>
java.time.Duration.ZERO
+ case duration: FiniteDuration => duration.asJava
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
