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 d1a17ddb PekkoHttpClientUtils: division by zero guard (#819)
d1a17ddb is described below
commit d1a17ddb27228d05c218373455659bf41f8c2f29
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Aug 4 09:42:33 2026 +0100
PekkoHttpClientUtils: division by zero guard (#819)
* concurrency issue with roundRobin var
* Update PekkoHttpClientUtils.scala
* Update PekkoHttpClientUtils.scala
* Update PekkoHttpClientUtils.scala
* Update PekkoHttpClientUtils.scala
---
.../pekko/grpc/internal/PekkoHttpClientUtils.scala | 20 +++++++++++++-------
1 file changed, 13 insertions(+), 7 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 508ba6e3..c9797961 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
@@ -29,7 +29,7 @@ import pekko.http.scaladsl.{ ClientTransport,
ConnectionContext, Http }
import pekko.http.scaladsl.model._
import pekko.http.scaladsl.model.headers.RawHeader
import pekko.http.scaladsl.settings.ClientConnectionSettings
-import pekko.stream.{ Materializer, OverflowStrategy }
+import pekko.stream.{ Materializer, OverflowStrategy, QueueOfferResult }
import pekko.stream.scaladsl.{ Keep, Sink, Source }
import pekko.util.ByteString
import io.grpc.{ CallOptions, MethodDescriptor, Status, StatusRuntimeException
}
@@ -37,7 +37,6 @@ 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._
import scala.jdk.FutureConverters._
import scala.util.{ Failure, Success }
@@ -68,17 +67,20 @@ object PekkoHttpClientUtils {
// https://github.com/akka/akka-grpc/issues/1196
// https://github.com/akka/akka-grpc/issues/1197
- var roundRobin: Int = 0
+ @volatile var roundRobin: Int = 0
val clientConnectionSettings =
ClientConnectionSettings(sys).withTransport(ClientTransport.withCustomResolver((host,
_) => {
settings.overrideAuthority.foreach { authority =>
assert(host == authority)
}
- settings.serviceDiscovery.lookup(settings.serviceName, 10.seconds).map
{ resolved =>
+ settings.serviceDiscovery.lookup(settings.serviceName,
settings.resolveTimeout).map { resolved =>
+ if (resolved.addresses.isEmpty)
+ throw new IllegalStateException(
+ s"Service discovery for '${settings.serviceName}' returned no
addresses")
// quasi-roundrobin is nicer than random selection: somewhat lower
chance of making
// an 'unlucky choice' multiple times in a row.
roundRobin += 1
- val target = resolved.addresses(roundRobin % resolved.addresses.size)
+ val target = resolved.addresses(math.abs(roundRobin %
resolved.addresses.size))
target.address match {
case Some(address) =>
new InetSocketAddress(address,
target.port.getOrElse(settings.defaultPort))
@@ -123,7 +125,11 @@ object PekkoHttpClientUtils {
def singleRequest(request: HttpRequest): Future[HttpResponse] = {
val p = Promise[HttpResponse]()
- queue.offer(request.addAttribute(ResponsePromise.Key,
ResponsePromise(p))).flatMap(_ => p.future)
+ queue.offer(request.addAttribute(ResponsePromise.Key,
ResponsePromise(p))).foreach {
+ case QueueOfferResult.Enqueued => // promise will be completed by the
response sink
+ case _ => p.tryFailure(new
IllegalStateException("Request queue closed"))
+ }
+ p.future
}
implicit def serializerFromMethodDescriptor[I, O](descriptor:
MethodDescriptor[I, O]): ProtobufSerializer[I] =
@@ -221,7 +227,7 @@ object PekkoHttpClientUtils {
case Chunk(data, _) =>
data
case LastChunk(_, trailer) =>
- trailerPromise.success(trailer)
+ trailerPromise.trySuccess(trailer)
ByteString.empty
}
.watchTermination((_, done) =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
