This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-grpc.git
The following commit(s) were added to refs/heads/main by this push:
new 5d493211 apply deadline on pekko-http backend (#824)
5d493211 is described below
commit 5d4932115c7e5f861c3ef9e02c447ad00ef9c9b3
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Aug 8 08:37:45 2026 +0100
apply deadline on pekko-http backend (#824)
* apply deadline on pekko-http backend
* Refactor applyDeadline call for clarity
* Update PekkoHttpClientUtils.scala
---
.../pekko/grpc/internal/PekkoHttpClientUtils.scala | 32 +++++++++++++++++++-
.../grpc/internal/PekkoHttpClientUtilsSpec.scala | 35 +++++++++++++++++++++-
2 files changed, 65 insertions(+), 2 deletions(-)
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
index c4548ed4..4cda5ef2 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
@@ -37,6 +37,7 @@ import io.grpc.{ CallOptions, MethodDescriptor, Status,
StatusRuntimeException }
import javax.net.ssl.{ KeyManager, SSLContext, TrustManager }
import scala.collection.immutable
import scala.concurrent.{ ExecutionContext, Future, Promise }
+import scala.concurrent.duration.DurationLong
import scala.jdk.FutureConverters._
import scala.util.{ Failure, Success }
@@ -207,7 +208,36 @@ object PekkoHttpClientUtils {
descriptor.getFullMethodName),
GrpcEntityHelpers.metadataHeaders(headers.entries),
source)
- responseToSource(singleRequest(httpRequest), deserializer)
+ applyDeadline(responseToSource(singleRequest(httpRequest),
deserializer), options)
+ }
+ }
+ }
+
+ /**
+ * INTERNAL API
+ */
+ @InternalApi
+ private[internal] def applyDeadline[O](source: Source[O,
Future[GrpcResponseMetadata]], options: CallOptions)
+ : Source[O, Future[GrpcResponseMetadata]] = {
+ val deadline = options.getDeadline
+ if (deadline == null) source
+ else {
+ val remaining =
deadline.timeRemaining(java.util.concurrent.TimeUnit.MILLISECONDS)
+ if (remaining <= 0) {
+ val ex = Status.DEADLINE_EXCEEDED
+ .withDescription(s"Deadline expired ${-remaining} ms ago")
+ .asRuntimeException()
+ Source.failed(ex).mapMaterializedValue(_ => Future.failed(ex))
+ } else {
+ source.completionTimeout(remaining.millis).recoverWithRetries(
+ 1,
+ {
+ case _: pekko.stream.CompletionTimeoutException =>
+ val ex = Status.DEADLINE_EXCEEDED
+ .withDescription(s"Deadline of ${remaining}ms exceeded")
+ .asRuntimeException()
+ Source.failed(ex)
+ })
}
}
}
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala
index b67a59ed..36db8f6a 100644
---
a/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala
+++
b/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala
@@ -18,13 +18,15 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.actor.ActorSystem
+import pekko.grpc.GrpcResponseMetadata
import pekko.http.scaladsl.model.HttpEntity.Strict
import pekko.http.scaladsl.model._
import pekko.http.scaladsl.model.StatusCodes._
import pekko.http.scaladsl.model.headers.RawHeader
+import pekko.stream.scaladsl.{ Sink, Source }
import pekko.testkit.TestKit
import pekko.util.ByteString
-import io.grpc.{ Metadata, Status, StatusRuntimeException }
+import io.grpc.{ CallOptions, Deadline, Metadata, Status,
StatusRuntimeException }
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.matchers.should.Matchers
import org.scalatest.time.Span
@@ -86,4 +88,35 @@ class PekkoHttpClientUtilsSpec extends
TestKit(ActorSystem()) with AnyWordSpecLi
lazy val key = Metadata.Key.of("custom-key",
Metadata.ASCII_STRING_MARSHALLER)
lazy val keyBin = Metadata.Key.of("custom-key-bin",
Metadata.BINARY_BYTE_MARSHALLER)
}
+
+ "applyDeadline" should {
+ type Resp = Future[GrpcResponseMetadata]
+
+ "pass source through unchanged when no deadline is set" in {
+ val source = Source.single(ByteString(1, 2, 3))
+ val options = CallOptions.DEFAULT
+ val result =
PekkoHttpClientUtils.applyDeadline(source.asInstanceOf[Source[Any, Resp]],
options)
+ val value = result.runWith(Sink.head).futureValue
+ value shouldBe ByteString(1, 2, 3)
+ }
+
+ "fail immediately when deadline is already expired" in {
+ val source = Source.single(ByteString(1, 2, 3))
+ val options = CallOptions.DEFAULT.withDeadline(Deadline.after(0,
java.util.concurrent.TimeUnit.MILLISECONDS))
+ val result =
PekkoHttpClientUtils.applyDeadline(source.asInstanceOf[Source[Any, Resp]],
options)
+ val failure = result.runWith(Sink.head).failed.futureValue
+ failure shouldBe a[StatusRuntimeException]
+ failure.asInstanceOf[StatusRuntimeException].getStatus.getCode shouldBe
Status.Code.DEADLINE_EXCEEDED
+ }
+
+ "fail with DEADLINE_EXCEEDED when source does not complete in time" in {
+ // Source that never completes
+ val source = Source.maybe[ByteString]
+ val options = CallOptions.DEFAULT.withDeadline(Deadline.after(100,
java.util.concurrent.TimeUnit.MILLISECONDS))
+ val result =
PekkoHttpClientUtils.applyDeadline(source.asInstanceOf[Source[Any, Resp]],
options)
+ val failure = result.runWith(Sink.head).failed.futureValue
+ failure shouldBe a[StatusRuntimeException]
+ failure.asInstanceOf[StatusRuntimeException].getStatus.getCode shouldBe
Status.Code.DEADLINE_EXCEEDED
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
