He-Pin commented on code in PR #3035:
URL: https://github.com/apache/pekko/pull/3035#discussion_r3343517196
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -206,29 +207,61 @@ object GraphStageLogic {
*
* Not for user instantiation, use [[GraphStageLogic.getStageActor]].
*/
- final class StageActor @InternalApi() private[pekko] (
+ final class StageActor @InternalApi() private (
materializer: Materializer,
- getAsyncCallback: StageActorRef.Receive => AsyncCallback[(ActorRef,
Any)],
initialReceive: StageActorRef.Receive,
- name: String) {
+ name: String,
+ cell: ActorCell,
+ buildDispatch: StageActorRef.Receive => ((ActorRef, Any)) => Unit) {
+
+ @InternalApi private[pekko] def this(
+ materializer: Materializer,
+ getAsyncCallback: StageActorRef.Receive => AsyncCallback[(ActorRef,
Any)],
+ initialReceive: StageActorRef.Receive,
+ name: String) =
+ this(
+ materializer,
+ initialReceive,
+ name,
+ StageActor.localCell(materializer.supervisor, "Stream supervisor"),
+ receive => getAsyncCallback(receive).invoke)
+
+ @InternalApi private[pekko] def this(
+ materializer: Materializer,
+ interpreter: GraphInterpreter,
+ logic: GraphStageLogic,
+ initialReceive: StageActorRef.Receive,
+ name: String) =
+ this(
+ materializer,
+ initialReceive,
+ name,
+ StageActor.localCell(materializer.supervisor, "Stream supervisor"),
+ // Coalesce per-tell mailbox traffic: N tells produce 1 AsyncInput
envelope (amortized).
+ receive =>
+ new StageActor.LazyDispatch(
+ interpreter,
+ logic,
+ receive.asInstanceOf[Any => Unit],
+ StageActor.drainBatchSize(materializer)))
+
+ // Monomorphic Function1 captured once; JIT can inline the apply at the
FunctionRef call site.
+ private val dispatch: ((ActorRef, Any)) => Unit =
buildDispatch(internalReceive)
- private val callback = getAsyncCallback(internalReceive)
-
- private def cell = materializer.supervisor match {
- case ref: LocalActorRef => ref.underlying
- case unknown =>
- throw new IllegalStateException(s"Stream supervisor must be a local
actor, was [${unknown.getClass.getName}]")
- }
private val functionRef: FunctionRef = {
- val f: (ActorRef, Any) => Unit = {
- case (_, m @ (PoisonPill | Kill)) =>
- materializer.logger.warning(
- "{} message sent to StageActor({}) will be ignored, since it is
not a real Actor." +
- "Use a custom message type to communicate with it instead.",
- m,
- functionRef.path)
- case pair => callback.invoke(pair)
- }
+ // Explicit (sender, msg) lambda (not a pattern-match Function2 literal)
so the PoisonPill / Kill
+ // branch matches on `msg` directly and does not allocate a Tuple2. The
regular branch still
+ // constructs one tuple per tell, as required by the `((ActorRef, Any))
=> Unit` public Receive type.
+ val f: (ActorRef, Any) => Unit = (sender, msg) =>
+ msg match {
+ case PoisonPill | Kill =>
+ materializer.logger.warning(
+ "{} message sent to StageActor({}) will be ignored, since it is
not a real Actor." +
+ "Use a custom message type to communicate with it instead.",
Review Comment:
Fixed in da487e3c74: added trailing space so the concatenation reads `...not
a real Actor. Use a custom message type...`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
