Replace ExecutionContexts.sameThreadExecutionContext with parasitic

apache · Dec 31, 2023 · d8d347b · d8d347b
1 parent 7336f7c
commit d8d347b
Show file tree

Hide file tree

Showing 13 changed files with 30 additions and 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
@@ -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/.../src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala b/.../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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -99,24 +99,24 @@ 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
     }
   }
 
   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
       }
     }
 
   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
@@ -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,