/** * Initialize a [[StageActorRef]] which can be used to interact with from the outside world "as-if" an [[Actor]]. * The messages are looped through the [[getAsyncCallback]] mechanism of [[GraphStage]] so they are safe to modify * internal state of this stage. * * This method must (the earliest) be called after the [[GraphStageLogic]] constructor has finished running, * for example from the [[preStart]] callback the graph stage logic provides. * * Created [[StageActorRef]] to get messages and watch other actors in synchronous way. * * The [[StageActorRef]]'s lifecycle is bound to the Stage, in other words when the Stage is finished, * the Actor will be terminated as well. The entity backing the [[StageActorRef]] is not a real Actor, * but the [[GraphStageLogic]] itself, therefore it does not react to [[PoisonPill]]. * * @param receive callback that will be called upon receiving of a message by this special Actor * @return minimal actor with watch method */
  // FIXME: I don't like the Pair allocation :(
 @ApiMayChange final protected def getStageActor(receive: ((ActorRef, Any)) ? Unit): StageActor = { _stageActor match { case null ? val actorMaterializer = ActorMaterializerHelper.downcast(interpreter.materializer) _stageActor = new StageActor(actorMaterializer, getAsyncCallback, receive) _stageActor case existing ? existing.become(receive) existing } } ... /** * The ActorRef by which this StageActor can be contacted from the outside. * This is a full-fledged ActorRef that supports watching and being watched * as well as location transparent (remote) communication. */ def ref: ActorRef = functionRef

      def behavior(m:(ActorRef,Any)): Unit = { val (sender, msg) = m msg.asInstanceOf[String] match { case "Stop" => completeStage() case s@ _ => extMessage = s } }

class StageAsActor(extActor: ActorRef) extends GraphStage[FlowShape[String,String]] { val inport = Inlet[String]("input") val outport = Outlet[String]("output") val shape = FlowShape.of(inport,outport) override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) { var extMessage = ""
      override def preStart(): Unit = { extActor ! getStageActor(behavior).ref } def behavior(m:(ActorRef,Any)): Unit = { val (sender, msg) = m msg.asInstanceOf[String] match { case "Stop" => completeStage() case s@ _ => extMessage = s } } setHandler(inport, new InHandler { override def onPush(): Unit =
          if (extMessage.length > 0) { push(outport,extMessage) extMessage="" } else push(outport, grab(inport)) }) setHandler(outport, new OutHandler { override def onPull(): Unit = pull(inport) }) } }