This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 41b142dd23 perf(stream): optimize Source.actorRef with direct-push
fast path and message-only FunctionRef (#3091)
41b142dd23 is described below
commit 41b142dd2355d9106d4588c2a2773a1641610234
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Jun 19 23:23:37 2026 +0800
perf(stream): optimize Source.actorRef with direct-push fast path and
message-only FunctionRef (#3091)
* fix(stream): force-start supervisor on UnstartedCell race in
StageActor.localCell
Motivation:
The first Source.actorRef (or any StageActor-backed stage) materialized
on a fresh ActorSystem can race the async supervisor startup. When the
RepointableActorRef still holds an UnstartedCell, localCell throws
IllegalStateException instead of waiting for the real ActorCell.
Modification:
Add an UnstartedCell case to StageActor.localCell that force-starts the
supervisor via Ask(Identify) — the Identify round-trip ensures the
Supervise message has been processed and the cell has been swapped.
Result:
First materialization on a fresh system no longer throws. The fix is
global — all stages using getStageActor/getEagerStageActor benefit.
Tests:
- sbt "stream-tests / Test / testOnly ...ActorRefSourceSpec" — 14/14 pass
- New regression test creates 20 fresh ActorSystems and materializes
Source.actorRef as the first operation on each.
References:
Refs akkadotnet/akka.net#8263
* perf(stream): add direct-push fast path in ActorRefSource
Motivation:
When the buffer is empty and downstream has demand, ActorRefSource was
enqueuing elements into the buffer and immediately dequeuing them in
tryPush — an unnecessary enqueue/dequeue round-trip on every element.
Modification:
Check buffer.isEmpty && isAvailable(out) before the buffer path; when
both hold, push directly to the outlet, skipping the buffer entirely.
Result:
Each element in the low-latency case (consumer keeps up with producer)
avoids two buffer operations (~5-10ns per element). No behavioral change
— ordering is preserved since the fast path only fires when the buffer
is empty.
Tests:
- sbt "stream-tests / Test / testOnly ...ActorRefSourceSpec" — 15/15 pass
- New test sends elements one at a time with continuous demand and
verifies FIFO ordering through the direct-push path.
References:
Refs akkadotnet/akka.net#8263
* perf(stream): bypass StageActor in ActorRefSource for message-only async
callback
Motivation:
ActorRefSource used getEagerStageActor which creates a (ActorRef, Any)
Tuple2 (~24 bytes) per message via the StageActor FunctionRef lambda.
The sender is never used by ActorRefSource — it is always Actor.noSender
for external tells — so this allocation is pure waste on the hot path.
Modification:
Replace getEagerStageActor with a direct FunctionRef created on the
stream supervisor's ActorCell. The FunctionRef lambda intercepts
PoisonPill/Kill (ignoring them) and delegates to a message-only
AsyncCallback[Any] — no tuple boxing. The supervisor cell is resolved
inline with UnstartedCell race handling. postStop removes the
FunctionRef to prevent leaks.
Result:
Each element avoids ~24 bytes of Tuple2 allocation. At 1M elem/s this
saves ~24 MB/s of GC pressure. PoisonPill/Kill are explicitly ignored
at the FunctionRef level, preventing them from leaking as data when
T = Any.
Tests:
- sbt "stream-tests / Test / testOnly ...ActorRefSourceSpec" — 17/17 pass
- New regression tests: PoisonPill and Kill ignored for T = Any
References:
Refs akkadotnet/akka.net#8263
* test(stream): add JMH benchmark for Source.actorRef push throughput
Motivation:
No existing benchmark measured Source.actorRef throughput specifically.
The StageActorRefBenchmark covers StageActor tell throughput via a
custom sink, but not the Source.actorRef ingress path which is the
target of the recent performance optimizations.
Modification:
Add ActorRefSourceBenchmark that creates a Source.actorRef with a
1024-element buffer and pushes 100k Int elements from a dedicated
sender thread, measuring throughput in elements/second.
Result:
Provides a reproducible benchmark for tracking Source.actorRef
performance across the direct-push fast path and message-only
FunctionRef optimizations.
Tests:
- Not run - benchmark only (compile verified via sbt
bench-jmh/Compile/compile)
References:
Refs akkadotnet/akka.net#8263
* test(stream): redesign ActorRefSource benchmark for
optimization-sensitive measurement
The original buffer=1024 saturated-push benchmark didn't exercise the
direct-push fast path (buffer stays full, so elements always enqueue).
Replace with two targeted benchmarks:
- no_buffer (buffer=0): measures FunctionRef+AsyncCallback path cost
without buffer noise, isolating the tuple boxing elimination benefit
- pingpong (buffer=1, sync send-wait): ensures buffer is always empty
when next element arrives, guaranteeing 100% direct-path hits
Results (10 iter, 5 warmup, same machine):
no_buffer: 18.0M → 19.5M ops/s (+8.3%)
pingpong: 883K → 988K ops/s (+11.9%)
* refactor(stream): address code review findings for ActorRefSource
optimization
- Extract shared resolveSupervisorCell on GraphStageLogic to eliminate
duplicated localCell logic between StageActor and ActorRefSource
- Add warning log for PoisonPill/Kill (matching StageActor behavior),
using systemLog to avoid interpreter initialization dependency
- Wrap removeFunctionRef in postStop with NonFatal catch for robustness
during ActorSystem shutdown
- Add super.postStop() call for forward compatibility
- Log unmatched messages at debug level instead of silent drop
- Add comment explaining direct-push fast path completion safety
- Reduce startup race test from 20 to 5 iterations, verify stream
completion via Done future
- Add post-completion dead letter test
* style(stream): apply scalafmt to changed files
---
.../pekko/stream/ActorRefSourceBenchmark.scala | 107 +++++++++++++
.../pekko/stream/scaladsl/ActorRefSourceSpec.scala | 89 ++++++++++-
.../apache/pekko/stream/impl/ActorRefSource.scala | 172 +++++++++++++--------
.../org/apache/pekko/stream/stage/GraphStage.scala | 37 ++++-
4 files changed, 331 insertions(+), 74 deletions(-)
diff --git
a/bench-jmh/src/main/scala/org/apache/pekko/stream/ActorRefSourceBenchmark.scala
b/bench-jmh/src/main/scala/org/apache/pekko/stream/ActorRefSourceBenchmark.scala
new file mode 100644
index 0000000000..11f9e2ef4e
--- /dev/null
+++
b/bench-jmh/src/main/scala/org/apache/pekko/stream/ActorRefSourceBenchmark.scala
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream
+
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicLong
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+import org.openjdk.jmh.annotations._
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.stream.scaladsl.Keep
+import pekko.stream.scaladsl.Sink
+import pekko.stream.scaladsl.Source
+
+object ActorRefSourceBenchmark {
+ final val OperationsPerInvocation = 100000
+}
+
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@BenchmarkMode(Array(Mode.Throughput))
+class ActorRefSourceBenchmark {
+ import ActorRefSourceBenchmark._
+
+ implicit val system: ActorSystem = ActorSystem("ActorRefSourceBenchmark")
+
+ @Setup
+ def setup(): Unit = {
+ SystemMaterializer(system).materializer
+ }
+
+ @TearDown
+ def shutdown(): Unit = {
+ Await.result(system.terminate(), 5.seconds)
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(OperationsPerInvocation)
+ def actorRef_source_no_buffer_100k(): Unit = {
+ val doneLatch = new CountDownLatch(1)
+ val sourceRef = Source
+ .actorRef[Any](
+ completionMatcher = { case "done" => CompletionStrategy.draining },
+ failureMatcher = PartialFunction.empty,
+ bufferSize = 0,
+ overflowStrategy = OverflowStrategy.dropHead)
+ .toMat(Sink.ignore)(Keep.left)
+ .run()
+
+ val sender = new Thread(() => {
+ var i = 0
+ while (i < OperationsPerInvocation) {
+ sourceRef ! i
+ i += 1
+ }
+ sourceRef ! "done"
+ doneLatch.countDown()
+ })
+ sender.start()
+ if (!doneLatch.await(30, TimeUnit.SECONDS))
+ throw new RuntimeException("ActorRefSource benchmark timed out")
+ sender.join()
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(OperationsPerInvocation)
+ def actorRef_source_pingpong_100k(): Unit = {
+ val counter = new AtomicLong(0)
+ val sourceRef = Source
+ .actorRef[Long](
+ completionMatcher = { case -1L => CompletionStrategy.draining },
+ failureMatcher = PartialFunction.empty,
+ bufferSize = 1,
+ overflowStrategy = OverflowStrategy.dropHead)
+ .toMat(Sink.foreach[Long](_ => counter.incrementAndGet()))(Keep.left)
+ .run()
+
+ var i = 0L
+ while (i < OperationsPerInvocation) {
+ sourceRef ! i
+ val expected = i + 1
+ while (counter.get() < expected) () // spin-wait for consumption
+ i += 1
+ }
+ sourceRef ! -1L
+ }
+}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala
index 255f41e3ad..dda34e17ae 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala
@@ -14,11 +14,12 @@
package org.apache.pekko.stream.scaladsl
import scala.annotation.nowarn
+import scala.concurrent.Await
import scala.concurrent.duration._
import org.apache.pekko
import pekko.Done
-import pekko.actor.{ ActorRef, Status }
+import pekko.actor.{ ActorRef, ActorSystem, Kill, PoisonPill, Status }
import pekko.stream.{ OverflowStrategy, _ }
import pekko.stream.testkit._
import pekko.stream.testkit.Utils._
@@ -227,5 +228,91 @@ class ActorRefSourceSpec extends StreamSpec {
mat.shutdown()
}
}
+
+ "materialize even when the stream supervisor is not yet started" in {
+ // Regression: the first Source.actorRef materialization on a fresh
ActorSystem
+ // can race the async supervisor startup, hitting an UnstartedCell in
StageActor.localCell.
+ // The fix force-starts the supervisor via Ask(Identify) before
returning the cell.
+ (1 to 5).foreach { i =>
+ val sys = ActorSystem(s"ActorRefSource-startup-race-$i")
+ try {
+ val mat = Materializer(sys)
+ val (ref, done) = Source
+ .actorRef[Int]({ case "ok" => CompletionStrategy.draining },
PartialFunction.empty, 8,
+ OverflowStrategy.fail)
+ .toMat(Sink.ignore)(Keep.both)
+ .run()(mat)
+ ref should not be null
+ ref ! "ok"
+ Await.result(done, 5.seconds) should be(Done)
+ } finally {
+ Await.result(sys.terminate(), 10.seconds)
+ }
+ }
+ }
+ "push directly without buffer round-trip when buffer is empty and demand
exists" in {
+ val s = TestSubscriber.manualProbe[Int]()
+ val ref = Source
+ .actorRef({ case "ok" => CompletionStrategy.draining },
PartialFunction.empty, 100, OverflowStrategy.fail)
+ .to(Sink.fromSubscriber(s))
+ .run()
+ val sub = s.expectSubscription()
+ // Request enough demand upfront so every element hits the direct-push
fast path
+ sub.request(10)
+ // Send elements one at a time — each should push directly without
buffering
+ for (n <- 1 to 10) {
+ ref ! n
+ s.expectNext(n)
+ }
+ ref ! "ok"
+ s.expectComplete()
+ }
+
+ "ignore PoisonPill and not emit it as a data element" in {
+ val s = TestSubscriber.manualProbe[Any]()
+ val ref = Source
+ .actorRef[Any]({ case "ok" => CompletionStrategy.draining },
PartialFunction.empty, 10, OverflowStrategy.fail)
+ .to(Sink.fromSubscriber(s))
+ .run()
+ val sub = s.expectSubscription()
+ sub.request(10)
+ ref ! PoisonPill
+ // PoisonPill must not be emitted and must not complete the stream
+ s.expectNoMessage(300.millis)
+ ref ! "real-element"
+ s.expectNext("real-element")
+ ref ! "ok"
+ s.expectComplete()
+ }
+
+ "ignore Kill and not emit it as a data element" in {
+ val s = TestSubscriber.manualProbe[Any]()
+ val ref = Source
+ .actorRef[Any]({ case "ok" => CompletionStrategy.draining },
PartialFunction.empty, 10, OverflowStrategy.fail)
+ .to(Sink.fromSubscriber(s))
+ .run()
+ val sub = s.expectSubscription()
+ sub.request(10)
+ ref ! Kill
+ // Kill must not be emitted and must not complete the stream
+ s.expectNoMessage(300.millis)
+ ref ! "real-element"
+ s.expectNext("real-element")
+ ref ! "ok"
+ s.expectComplete()
+ }
+
+ "not fail when messages are sent after stream completion" in {
+ val (ref, done) = Source
+ .actorRef[Int]({ case "ok" => CompletionStrategy.draining },
PartialFunction.empty, 10, OverflowStrategy.fail)
+ .toMat(Sink.ignore)(Keep.both)
+ .run()
+ ref ! "ok"
+ Await.result(done, 5.seconds) should be(Done)
+ // After completion, FunctionRef should be removed; messages go to dead
letters
+ ref ! 42
+ ref ! 43
+ // No exception should be thrown
+ }
}
}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefSource.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefSource.scala
index 1a76f23bf0..0d73f71231 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/ActorRefSource.scala
@@ -13,8 +13,10 @@
package org.apache.pekko.stream.impl
+import scala.util.control.NonFatal
+
import org.apache.pekko
-import pekko.actor.ActorRef
+import pekko.actor.{ ActorRef, FunctionRef, Kill, PoisonPill }
import pekko.annotation.InternalApi
import pekko.stream._
import pekko.stream.OverflowStrategies._
@@ -62,85 +64,123 @@ private object ActorRefSource {
}
private var isCompleting: Boolean = false
- override protected def stageActorName: String =
inheritedAttributes.nameForActorRef(super.stageActorName)
-
private val name = inheritedAttributes.nameOrDefault(getClass.toString)
- override val ref: ActorRef = getEagerStageActor(eagerMaterializer) {
- case (_, m) if failureMatcher.isDefinedAt(m) =>
- failStage(failureMatcher(m))
- case (_, m) if completionMatcher.isDefinedAt(m) =>
- completionMatcher(m) match {
+
+ private val cell = resolveSupervisorCell(eagerMaterializer)
+
+ private val callback = getAsyncCallback[Any](onMessage)
+
+ private val functionRef: FunctionRef = {
+ val actorName =
inheritedAttributes.nameForActorRef(super.stageActorName)
+ val systemLog = eagerMaterializer.system.log
+ cell.addFunctionRef(
+ (_, msg) =>
+ msg match {
+ case PoisonPill | Kill =>
+ systemLog.warning(
+ "{} message sent to ActorRefSource({}) will be ignored,
since it is not a real Actor. " +
+ "Use a custom message type to communicate with it instead.",
+ msg,
+ functionRef.path)
+ case _ => callback.invoke(msg)
+ },
+ actorName)
+ }
+
+ override val ref: ActorRef = functionRef
+
+ override def postStop(): Unit = {
+ try cell.removeFunctionRef(functionRef)
+ catch { case NonFatal(_) => () }
+ super.postStop()
+ }
+
+ private def onMessage(msg: Any): Unit = {
+ if (failureMatcher.isDefinedAt(msg)) {
+ failStage(failureMatcher(msg))
+ } else if (completionMatcher.isDefinedAt(msg)) {
+ completionMatcher(msg) match {
case CompletionStrategy.Draining =>
isCompleting = true
tryPush()
case CompletionStrategy.Immediately =>
completeStage()
}
- case (_, m: T @unchecked) =>
- buffer match {
- case OptionVal.Some(buf) =>
- if (isCompleting) {
- log.warning(
- "Dropping element because Status.Success received already,
only draining already buffered elements: [{}] (pending: [{}] in stream [{}])",
- m,
- buf.used,
- name)
- } else if (!buf.isFull) {
- buf.enqueue(m)
- tryPush()
- } else
- overflowStrategy match {
- case s: DropHead =>
- log.log(
- s.logLevel,
- "Dropping the head element because buffer is full and
overflowStrategy is: [DropHead] in stream [{}]",
+ } else {
+ msg match {
+ case m: T @unchecked =>
+ buffer match {
+ case OptionVal.Some(buf) =>
+ if (isCompleting) {
+ log.warning(
+ "Dropping element because Status.Success received
already, only draining already buffered elements: [{}] (pending: [{}] in stream
[{}])",
+ m,
+ buf.used,
name)
- buf.dropHead()
+ } else if (buf.isEmpty && isAvailable(out)) {
+ // Direct-push fast path: safe to skip tryPush() because
isCompleting
+ // is checked above, and completion is driven by the
completionMatcher branch.
+ push(out, m)
+ } else if (!buf.isFull) {
buf.enqueue(m)
tryPush()
- case s: DropTail =>
- log.log(
- s.logLevel,
- "Dropping the tail element because buffer is full and
overflowStrategy is: [DropTail] in stream [{}]",
+ } else
+ overflowStrategy match {
+ case s: DropHead =>
+ log.log(
+ s.logLevel,
+ "Dropping the head element because buffer is full
and overflowStrategy is: [DropHead] in stream [{}]",
+ name)
+ buf.dropHead()
+ buf.enqueue(m)
+ tryPush()
+ case s: DropTail =>
+ log.log(
+ s.logLevel,
+ "Dropping the tail element because buffer is full
and overflowStrategy is: [DropTail] in stream [{}]",
+ name)
+ buf.dropTail()
+ buf.enqueue(m)
+ tryPush()
+ case s: DropBuffer =>
+ log.log(
+ s.logLevel,
+ "Dropping all the buffered elements because buffer
is full and overflowStrategy is: [DropBuffer] in stream [{}]",
+ name)
+ buf.clear()
+ buf.enqueue(m)
+ tryPush()
+ case s: Fail =>
+ log.log(
+ s.logLevel,
+ "Failing because buffer is full and overflowStrategy
is: [Fail] in stream [{}]",
+ name)
+ val bufferOverflowException =
+ BufferOverflowException(s"Buffer overflow (max
capacity was: $maxBuffer)!")
+ failStage(bufferOverflowException)
+ case _: Backpressure =>
+ failStage(new IllegalStateException("Backpressure is
not supported"))
+ }
+ case _ =>
+ if (isCompleting) {
+ log.warning(
+ "Dropping element because Status.Success received
already: [{}] in stream [{}]",
+ m,
name)
- buf.dropTail()
- buf.enqueue(m)
- tryPush()
- case s: DropBuffer =>
- log.log(
- s.logLevel,
- "Dropping all the buffered elements because buffer is
full and overflowStrategy is: [DropBuffer] in stream [{}]",
- name)
- buf.clear()
- buf.enqueue(m)
- tryPush()
- case s: Fail =>
- log.log(
- s.logLevel,
- "Failing because buffer is full and overflowStrategy is:
[Fail] in stream [{}]",
+ } else if (isAvailable(out)) {
+ push(out, m)
+ } else {
+ log.debug(
+ "Dropping element because there is no downstream demand
and no buffer: [{}] in stream [{}]",
+ m,
name)
- val bufferOverflowException =
- BufferOverflowException(s"Buffer overflow (max capacity
was: $maxBuffer)!")
- failStage(bufferOverflowException)
- case _: Backpressure =>
- // there is a precondition check in Source.actorRefSource
factory method to not allow backpressure as strategy
- failStage(new IllegalStateException("Backpressure is not
supported"))
- }
- case _ =>
- if (isCompleting) {
- log.warning("Dropping element because Status.Success received
already: [{}] in stream [{}]", m, name)
- } else if (isAvailable(out)) {
- push(out, m)
- } else {
- log.debug(
- "Dropping element because there is no downstream demand and
no buffer: [{}] in stream [{}]",
- m,
- name)
+ }
}
+ case unexpected =>
+ log.debug("Dropping unexpected non-data message [{}] in stream
[{}]", unexpected, name)
}
-
- case _ => throw new IllegalArgumentException() // won't happen,
compiler exhaustiveness check pleaser
- }.ref
+ }
+ }
private def tryPush(): Unit = {
if (isAvailable(out) && buffer.isDefined && buffer.get.nonEmpty) {
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
index dcd6faab5f..5ea8c444a9 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
@@ -21,13 +21,15 @@ import java.util.concurrent.atomic.AtomicReference
import scala.annotation.nowarn
import scala.annotation.tailrec
import scala.collection.{ immutable, mutable }
-import scala.concurrent.{ Future, Promise }
+import scala.concurrent.{ Await, Future, Promise }
import scala.concurrent.duration.FiniteDuration
import org.apache.pekko
import pekko.{ Done, NotUsed }
import pekko.actor._
import pekko.annotation.InternalApi
+import pekko.pattern.ask
+import pekko.util.Timeout
import pekko.dispatch.AbstractNodeQueue
import pekko.japi.function.{ Effect, Procedure }
import pekko.stream._
@@ -312,12 +314,22 @@ object GraphStageLogic {
private object StageActor {
def localCell(ref: ActorRef, description: String): ActorCell =
ref match {
- case ref: LocalActorRef => ref.underlying
- case ref: RepointableActorRef =>
- ref.underlying match {
- case cell: ActorCell => cell
- case unknown =>
- throw new IllegalStateException(s"$description must be a local
actor, was [${unknown.getClass.getName}]")
+ case ref: LocalActorRef => ref.underlying
+ case r: RepointableActorRef =>
+ r.underlying match {
+ case cell: ActorCell => cell
+ case _: UnstartedCell =>
+ implicit val timeout: Timeout = r.system.settings.CreationTimeout
+ Await.result(r ? Identify(null), timeout.duration)
+ r.underlying match {
+ case cell: ActorCell => cell
+ case unknown =>
+ throw new IllegalStateException(
+ s"$description must be a local actor, was
[${unknown.getClass.getName}]")
+ }
+ case unknown =>
+ throw new IllegalStateException(
+ s"$description must be a local actor, was
[${unknown.getClass.getName}]")
}
case unknown =>
throw new IllegalStateException(s"$description must be a local
actor, was [${unknown.getClass.getName}]")
@@ -1554,6 +1566,17 @@ abstract class GraphStageLogic private[stream] (val
inCount: Int, val outCount:
existing
}
+ /**
+ * INTERNAL API
+ *
+ * Resolves the stream supervisor's [[ActorCell]] so a caller can host a
FunctionRef on it.
+ * If the supervisor is a [[RepointableActorRef]] that has not yet been
started (its cell is still
+ * an [[UnstartedCell]]), it is force-started via Ask(Identify) before the
cell is returned.
+ */
+ @InternalApi
+ private[pekko] def resolveSupervisorCell(materializer: Materializer):
ActorCell =
+ StageActor.localCell(materializer.supervisor, "Stream supervisor")
+
/**
* Override and return a name to be given to the StageActor of this operator.
*
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
