You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by md...@apache.org on 2023/12/31 10:10:04 UTC
(incubator-pekko-http) branch main updated: Replace ExecutionContexts.sameThreadExecutionContext with parasitic
This is an automated email from the ASF dual-hosted git repository.
mdedetrich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-http.git
The following commit(s) were added to refs/heads/main by this push:
new d8d347b66 Replace ExecutionContexts.sameThreadExecutionContext with parasitic
d8d347b66 is described below
commit d8d347b66e5bcd9000ec8e7e76ddf9ff59716d85
Author: Matthew de Detrich <ma...@aiven.io>
AuthorDate: Sun Dec 31 20:23:09 2023 +1100
Replace ExecutionContexts.sameThreadExecutionContext with parasitic
---
.../pekko/http/impl/engine/client/PoolMasterActor.scala | 2 +-
.../http/impl/engine/client/pool/NewHostConnectionPool.scala | 8 ++++----
.../org/apache/pekko/http/impl/engine/http2/Http2.scala | 2 +-
.../pekko/http/impl/engine/server/HttpServerBluePrint.scala | 2 +-
.../scala/org/apache/pekko/http/impl/util/StreamUtils.scala | 2 +-
.../src/main/scala/org/apache/pekko/http/javadsl/Http.scala | 2 +-
.../scala/org/apache/pekko/http/javadsl/ServerBinding.scala | 6 +++---
.../scala/org/apache/pekko/http/javadsl/ServerBuilder.scala | 2 +-
.../src/main/scala/org/apache/pekko/http/scaladsl/Http.scala | 12 ++++++------
.../pekko/http/impl/engine/ws/WebSocketIntegrationSpec.scala | 2 +-
.../apache/pekko/http/javadsl/server/RequestContext.scala | 8 ++++----
.../http/javadsl/server/directives/BasicDirectives.scala | 10 +++++-----
.../http/javadsl/server/directives/RouteDirectives.scala | 2 +-
13 files changed, 30 insertions(+), 30 deletions(-)
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/PoolMasterActor.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/PoolMasterActor.scala
index 0adbfaf9d..e59732f70 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/PoolMasterActor.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/PoolMasterActor.scala
@@ -191,7 +191,7 @@ private[http] final class PoolMasterActor extends Actor with ActorLogging {
// has completed.
val completed = pool.shutdown()(context.dispatcher)
shutdownCompletedPromise.tryCompleteWith(
- completed.map(_ => Done)(ExecutionContexts.sameThreadExecutionContext))
+ completed.map(_ => Done)(ExecutionContexts.parasitic))
statusById += poolId -> PoolInterfaceShuttingDown(shutdownCompletedPromise)
case Some(PoolInterfaceShuttingDown(formerPromise)) =>
// Pool is already shutting down, mirror the existing promise.
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala
index f591e73b0..351f03c6a 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala
@@ -469,7 +469,7 @@ private[client] object NewHostConnectionPool {
entityComplete.onComplete(safely {
case Success(_) => withSlot(_.onRequestEntityCompleted())
case Failure(cause) => withSlot(_.onRequestEntityFailed(cause))
- })(ExecutionContexts.sameThreadExecutionContext)
+ })(ExecutionContexts.parasitic)
request.withEntity(newEntity)
}
@@ -524,9 +524,9 @@ private[client] object NewHostConnectionPool {
ongoingResponseEntity = None
ongoingResponseEntityKillSwitch = None
}
- }(ExecutionContexts.sameThreadExecutionContext)
+ }(ExecutionContexts.parasitic)
case Failure(_) => throw new IllegalStateException("Should never fail")
- })(ExecutionContexts.sameThreadExecutionContext)
+ })(ExecutionContexts.parasitic)
withSlot(_.onResponseReceived(response.withEntity(newEntity)))
}
@@ -609,7 +609,7 @@ private[client] object NewHostConnectionPool {
onConnectionAttemptFailed(currentEmbargoLevel)
sl.onConnectionAttemptFailed(cause)
}
- })(ExecutionContexts.sameThreadExecutionContext)
+ })(ExecutionContexts.parasitic)
slotCon
}
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala
index a58388b86..ea7ede8eb 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala
@@ -125,7 +125,7 @@ private[http] final class Http2Ext(implicit val system: ActorSystem)
// See https://github.com/akka/akka/issues/17992
case NonFatal(ex) =>
Done
- }(ExecutionContexts.sameThreadExecutionContext)
+ }(ExecutionContexts.parasitic)
} catch {
case NonFatal(e) =>
log.error(e, "Could not materialize handling flow for {}", incoming)
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala
index b173544d7..71ee7de31 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala
@@ -503,7 +503,7 @@ private[http] object HttpServerBluePrint {
log.error(ex,
s"Response stream for [${requestStart.debugString}] failed with '${ex.getMessage}'. Aborting connection.")
case _ => // ignore
- }(ExecutionContexts.sameThreadExecutionContext)
+ }(ExecutionContexts.parasitic)
newEntity
}
diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala
index 6b6343524..cf5822c54 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala
@@ -89,7 +89,7 @@ private[http] object StreamUtils {
killResult.future.value match {
case Some(res) => handleKill(res)
case None =>
- killResult.future.onComplete(killCallback.invoke)(ExecutionContexts.sameThreadExecutionContext)
+ killResult.future.onComplete(killCallback.invoke)(ExecutionContexts.parasitic)
}
}
diff --git a/http-core/src/main/scala/org/apache/pekko/http/javadsl/Http.scala b/http-core/src/main/scala/org/apache/pekko/http/javadsl/Http.scala
index 069da2a37..1f50e0178 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/javadsl/Http.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/javadsl/Http.scala
@@ -47,7 +47,7 @@ object Http extends ExtensionId[Http] with ExtensionIdProvider {
}
class Http(system: ExtendedActorSystem) extends pekko.actor.Extension {
- import pekko.dispatch.ExecutionContexts.{ sameThreadExecutionContext => ec }
+ import pekko.dispatch.ExecutionContexts.{ parasitic => ec }
import language.implicitConversions
private implicit def completionStageCovariant[T, U >: T](in: CompletionStage[T]): CompletionStage[U] =
diff --git a/http-core/src/main/scala/org/apache/pekko/http/javadsl/ServerBinding.scala b/http-core/src/main/scala/org/apache/pekko/http/javadsl/ServerBinding.scala
index 6997df443..0a46df496 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/javadsl/ServerBinding.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/javadsl/ServerBinding.scala
@@ -89,7 +89,7 @@ class ServerBinding private[http] (delegate: pekko.http.scaladsl.Http.ServerBind
def terminate(hardDeadline: java.time.Duration): CompletionStage[HttpTerminated] = {
delegate.terminate(FiniteDuration.apply(hardDeadline.toMillis, TimeUnit.MILLISECONDS))
- .map(_.asInstanceOf[HttpTerminated])(ExecutionContexts.sameThreadExecutionContext)
+ .map(_.asInstanceOf[HttpTerminated])(ExecutionContexts.parasitic)
.asJava
}
@@ -103,7 +103,7 @@ class ServerBinding private[http] (delegate: pekko.http.scaladsl.Http.ServerBind
*/
def whenTerminationSignalIssued: CompletionStage[java.time.Duration] =
delegate.whenTerminationSignalIssued
- .map(deadline => deadline.time.asJava)(ExecutionContexts.sameThreadExecutionContext)
+ .map(deadline => deadline.time.asJava)(ExecutionContexts.parasitic)
.asJava
/**
@@ -120,7 +120,7 @@ class ServerBinding private[http] (delegate: pekko.http.scaladsl.Http.ServerBind
*/
def whenTerminated: CompletionStage[HttpTerminated] =
delegate.whenTerminated
- .map(_.asInstanceOf[HttpTerminated])(ExecutionContexts.sameThreadExecutionContext)
+ .map(_.asInstanceOf[HttpTerminated])(ExecutionContexts.parasitic)
.asJava
/**
diff --git a/http-core/src/main/scala/org/apache/pekko/http/javadsl/ServerBuilder.scala b/http-core/src/main/scala/org/apache/pekko/http/javadsl/ServerBuilder.scala
index 05cba140f..861eeb297 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/javadsl/ServerBuilder.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/javadsl/ServerBuilder.scala
@@ -195,6 +195,6 @@ object ServerBuilder {
def connectionSource(): Source[IncomingConnection, CompletionStage[ServerBinding]] =
http.bindImpl(interface, port, context.asScala, settings.asScala, log)
.map(new IncomingConnection(_))
- .mapMaterializedValue(_.map(new ServerBinding(_))(ExecutionContexts.sameThreadExecutionContext).asJava).asJava
+ .mapMaterializedValue(_.map(new ServerBinding(_))(ExecutionContexts.parasitic).asJava).asJava
}
}
diff --git a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/Http.scala b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/Http.scala
index 690ba7396..cbaf8ac86 100644
--- a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/Http.scala
+++ b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/Http.scala
@@ -136,7 +136,7 @@ class HttpExt @InternalStableApi /* constructor signature is hardcoded in Teleme
.watchTermination() { (termWatchBefore, termWatchAfter) =>
// flag termination when the user handler has gotten (or has emitted) termination
// signals in both directions
- termWatchBefore.flatMap(_ => termWatchAfter)(ExecutionContexts.sameThreadExecutionContext)
+ termWatchBefore.flatMap(_ => termWatchAfter)(ExecutionContexts.parasitic)
}
.joinMat(baseFlow)(Keep.both))
@@ -280,7 +280,7 @@ class HttpExt @InternalStableApi /* constructor signature is hardcoded in Teleme
// from the TCP layer through the HTTP layer to the Http.IncomingConnection.flow.
// See https://github.com/akka/akka/issues/17992
case NonFatal(ex) => Done
- }(ExecutionContexts.sameThreadExecutionContext)
+ }(ExecutionContexts.parasitic)
} catch {
case NonFatal(e) =>
log.error(e, "Could not materialize handling flow for {}", incoming)
@@ -833,7 +833,7 @@ class HttpExt @InternalStableApi /* constructor signature is hardcoded in Teleme
val parallelism = settings.pipeliningLimit * settings.maxConnections
Flow[(HttpRequest, T)].mapAsyncUnordered(parallelism) {
case (request, userContext) => poolInterface(request).transform(response => Success(response -> userContext))(
- ExecutionContexts.sameThreadExecutionContext)
+ ExecutionContexts.parasitic)
}
}
@@ -948,7 +948,7 @@ object Http extends ExtensionId[HttpExt] with ExtensionIdProvider {
* Note: rather than unbinding explicitly you can also use [[addToCoordinatedShutdown]] to add this task to Akka's coordinated shutdown.
*/
def unbind(): Future[Done] =
- unbindAction().map(_ => Done)(ExecutionContexts.sameThreadExecutionContext)
+ unbindAction().map(_ => Done)(ExecutionContexts.parasitic)
/**
* Triggers "graceful" termination request being handled on this connection.
@@ -998,7 +998,7 @@ object Http extends ExtensionId[HttpExt] with ExtensionIdProvider {
_whenTerminationSignalIssued.trySuccess(hardDeadline.fromNow)
val terminated =
- unbindAction().flatMap(_ => terminateAction(hardDeadline))(ExecutionContexts.sameThreadExecutionContext)
+ unbindAction().flatMap(_ => terminateAction(hardDeadline))(ExecutionContexts.parasitic)
_whenTerminated.completeWith(terminated)
whenTerminated
}
@@ -1039,7 +1039,7 @@ object Http extends ExtensionId[HttpExt] with ExtensionIdProvider {
unbind()
}
shutdown.addTask(CoordinatedShutdown.PhaseServiceRequestsDone, s"http-terminate-${localAddress}") { () =>
- terminate(hardTerminationDeadline).map(_ => Done)(ExecutionContexts.sameThreadExecutionContext)
+ terminate(hardTerminationDeadline).map(_ => Done)(ExecutionContexts.parasitic)
}
this
}
diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketIntegrationSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketIntegrationSpec.scala
index 9a5a21ee9..fbc62e771 100644
--- a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketIntegrationSpec.scala
+++ b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/ws/WebSocketIntegrationSpec.scala
@@ -100,7 +100,7 @@ class WebSocketIntegrationSpec extends PekkoSpecWithMaterializer(
override def preStart(): Unit = {
promise.future.foreach(_ => getAsyncCallback[Done](_ => complete(shape.out)).invoke(Done))(
- pekko.dispatch.ExecutionContexts.sameThreadExecutionContext)
+ pekko.dispatch.ExecutionContexts.parasitic)
}
setHandlers(shape.in, shape.out, this)
diff --git a/http/src/main/scala/org/apache/pekko/http/javadsl/server/RequestContext.scala b/http/src/main/scala/org/apache/pekko/http/javadsl/server/RequestContext.scala
index 3bfb0e7dd..9caa61140 100644
--- a/http/src/main/scala/org/apache/pekko/http/javadsl/server/RequestContext.scala
+++ b/http/src/main/scala/org/apache/pekko/http/javadsl/server/RequestContext.scala
@@ -60,18 +60,18 @@ class RequestContext private (val delegate: scaladsl.server.RequestContext) {
def complete[T](value: T, marshaller: Marshaller[T, HttpResponse]): CompletionStage[RouteResult] = {
delegate.complete(ToResponseMarshallable(value)(marshaller))
- .fast.map(r => r: RouteResult)(pekko.dispatch.ExecutionContexts.sameThreadExecutionContext).asJava
+ .fast.map(r => r: RouteResult)(pekko.dispatch.ExecutionContexts.parasitic).asJava
}
def completeWith(response: HttpResponse): CompletionStage[RouteResult] = {
delegate.complete(response.asScala)
- .fast.map(r => r: RouteResult)(pekko.dispatch.ExecutionContexts.sameThreadExecutionContext).asJava
+ .fast.map(r => r: RouteResult)(pekko.dispatch.ExecutionContexts.parasitic).asJava
}
@varargs def reject(rejections: Rejection*): CompletionStage[RouteResult] = {
val scalaRejections = rejections.map(_.asScala)
delegate.reject(scalaRejections: _*)
- .fast.map(r => r: RouteResult)(pekko.dispatch.ExecutionContexts.sameThreadExecutionContext).asJava
+ .fast.map(r => r: RouteResult)(pekko.dispatch.ExecutionContexts.parasitic).asJava
}
def redirect(uri: Uri, redirectionType: StatusCode): CompletionStage[RouteResult] = {
@@ -80,7 +80,7 @@ class RequestContext private (val delegate: scaladsl.server.RequestContext) {
def fail(error: Throwable): CompletionStage[RouteResult] =
delegate.fail(error)
- .fast.map(r => r: RouteResult)(pekko.dispatch.ExecutionContexts.sameThreadExecutionContext).asJava
+ .fast.map(r => r: RouteResult)(pekko.dispatch.ExecutionContexts.parasitic).asJava
def withRequest(req: HttpRequest): RequestContext = wrap(delegate.withRequest(req.asScala))
def withExecutionContext(ec: ExecutionContextExecutor): RequestContext = wrap(delegate.withExecutionContext(ec))
diff --git a/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/BasicDirectives.scala b/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/BasicDirectives.scala
index 557104607..eb269c968 100644
--- a/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/BasicDirectives.scala
+++ b/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/BasicDirectives.scala
@@ -99,8 +99,8 @@ abstract class BasicDirectives {
inner: Supplier[Route]): Route = RouteAdapter {
D.mapRouteResultFuture(stage =>
FutureConverters.asScala(
- f(stage.fast.map(_.asJava)(ExecutionContexts.sameThreadExecutionContext).asJava)).fast.map(_.asScala)(
- ExecutionContexts.sameThreadExecutionContext)) {
+ f(stage.fast.map(_.asJava)(ExecutionContexts.parasitic).asJava)).fast.map(_.asScala)(
+ ExecutionContexts.parasitic)) {
inner.get.delegate
}
}
@@ -108,7 +108,7 @@ abstract class BasicDirectives {
def mapRouteResultWith(f: JFunction[RouteResult, CompletionStage[RouteResult]], inner: Supplier[Route]): Route =
RouteAdapter {
D.mapRouteResultWith(r =>
- FutureConverters.asScala(f(r.asJava)).fast.map(_.asScala)(ExecutionContexts.sameThreadExecutionContext)) {
+ FutureConverters.asScala(f(r.asJava)).fast.map(_.asScala)(ExecutionContexts.parasitic)) {
inner.get.delegate
}
}
@@ -116,7 +116,7 @@ abstract class BasicDirectives {
def mapRouteResultWithPF(
f: PartialFunction[RouteResult, CompletionStage[RouteResult]], inner: Supplier[Route]): Route = RouteAdapter {
D.mapRouteResultWith(r =>
- FutureConverters.asScala(f(r.asJava)).fast.map(_.asScala)(ExecutionContexts.sameThreadExecutionContext)) {
+ FutureConverters.asScala(f(r.asJava)).fast.map(_.asScala)(ExecutionContexts.parasitic)) {
inner.get.delegate
}
}
@@ -175,7 +175,7 @@ abstract class BasicDirectives {
f: JFunction[JIterable[Rejection], CompletionStage[RouteResult]], inner: Supplier[Route]): Route = RouteAdapter {
D.recoverRejectionsWith(rs =>
FutureConverters.asScala(f.apply(Util.javaArrayList(rs.map(_.asJava)))).fast.map(_.asScala)(
- ExecutionContexts.sameThreadExecutionContext)) { inner.get.delegate }
+ ExecutionContexts.parasitic)) { inner.get.delegate }
}
/**
diff --git a/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/RouteDirectives.scala b/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/RouteDirectives.scala
index 00ddbf341..38c86e703 100644
--- a/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/RouteDirectives.scala
+++ b/http/src/main/scala/org/apache/pekko/http/javadsl/server/directives/RouteDirectives.scala
@@ -46,7 +46,7 @@ abstract class RouteDirectives extends RespondWithDirectives {
import RoutingJavaMapping.Implicits._
// Don't try this at home – we only use it here for the java -> scala conversions
- private implicit val conversionExecutionContext: ExecutionContext = ExecutionContexts.sameThreadExecutionContext
+ private implicit val conversionExecutionContext: ExecutionContext = ExecutionContexts.parasitic
/**
* Java-specific call added so you can chain together multiple alternate routes using comma,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org