import akka.actor._ import akka.stream._ import akka.stream.scaladsl._ import akka.stream.stage._ import scala.concurrent.duration._ //external system
object Injector { var callback: AsyncCallback[String] = null def inject(m: String) = { if (callback != null) callback.invoke(m) } } class InjectControl(injector: Injector.type) 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 = { val callback = getAsyncCallback[String] { m =>
          if (m.length > 0) m match { case "Stop" => completeStage() case s: String => extMessage = s } } injector.callback = callback } 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) }) } } object GetAsyncCallbackDemo extends App { implicit val sys = ActorSystem("demoSys") implicit val ec = sys.dispatcher implicit val mat = ActorMaterializer( ActorMaterializerSettings(sys) .withInputBuffer(initialSize = 16, maxSize = 16) ) val source = Source(Stream.from(1)).map(_.toString).delay(1.second,DelayOverflowStrategy.backpressure) val graph = new InjectControl(Injector) val flow = Flow.fromGraph(graph) source.via(flow).to(Sink.foreach(println)).run() Thread.sleep(2000) Injector.inject("hello") Thread.sleep(2000) Injector.inject("world!") Thread.sleep(2000) Injector.inject("Stop") scala.io.StdIn.readLine() sys.terminate() }