pekko/akka-docs-dev/rst/scala/code/docs/Migrations.scala

package docs

import akka.stream.scaladsl._
import akka.stream._
import akka.stream.stage.{ OutHandler, InHandler, GraphStageLogic, GraphStage }
import akka.stream.testkit.AkkaSpec

import scala.concurrent.{ Future, ExecutionContext, Promise }
import scala.concurrent.duration._
import scala.util.{ Failure, Success, Try }

class Migrations extends AkkaSpec {

  "Examples in migration guide" must {
    "compile" in {
      val flow1 = Flow[Int]
      val flow2 = Flow[Int]

      def inlet: Inlet[Int] = ???
      def outlet: Outlet[Int] = ???

      def inlet1: Inlet[Int] = ???
      def outlet1: Outlet[Int] = ???
      def inlet2: Inlet[Int] = ???
      def outlet2: Outlet[Int] = ???

      lazy val dontExecuteMe = {
        //#flow-wrap
        val graphSource: Graph[SourceShape[Int], Unit] = ???
        val source: Source[Int, Unit] = Source.fromGraph(graphSource)

        val graphSink: Graph[SinkShape[Int], Unit] = ???
        val sink: Sink[Int, Unit] = Sink.fromGraph(graphSink)

        val graphFlow: Graph[FlowShape[Int, Int], Unit] = ???
        val flow: Flow[Int, Int, Unit] = Flow.fromGraph(graphFlow)

        Flow.fromSinkAndSource(Sink.head[Int], Source.single(0))
        //#flow-wrap

        //#bidiflow-wrap
        val bidiGraph: Graph[BidiShape[Int, Int, Int, Int], Unit] = ???
        val bidi: BidiFlow[Int, Int, Int, Int, Unit] = BidiFlow.fromGraph(bidiGraph)

        BidiFlow.fromFlows(flow1, flow2)

        BidiFlow.fromFunctions((x: Int) => x + 1, (y: Int) => y * 3)
        //#bidiflow-wrap

        //#graph-create
        // Replaces FlowGraph.closed()
        FlowGraph.create() { builder =>
          //...
          ClosedShape
        }

        // Replaces FlowGraph.partial()
        FlowGraph.create() { builder =>
          //...
          FlowShape(inlet, outlet)
        }
        //#graph-create

        //#graph-create-2
        Source.fromGraph(
          FlowGraph.create() { builder =>
            //...
            SourceShape(outlet)
          })

        Sink.fromGraph(
          FlowGraph.create() { builder =>
            //...
            SinkShape(inlet)
          })

        Flow.fromGraph(
          FlowGraph.create() { builder =>
            //...
            FlowShape(inlet, outlet)
          })

        BidiFlow.fromGraph(
          FlowGraph.create() { builder =>
            //...
            BidiShape(inlet1, outlet1, inlet2, outlet2)
          })
        //#graph-create-2

        //#graph-edges
        RunnableGraph.fromGraph(
          FlowGraph.create() { implicit builder =>
            import FlowGraph.Implicits._
            outlet ~> inlet
            outlet ~> flow ~> inlet
            //...
            ClosedShape
          })
        //#graph-edges

        val promise = Promise[Unit]()

        //#source-creators
        val src: Source[Int, Promise[Option[Int]]] = Source.maybe[Int]
        //...
        // This finishes the stream without emitting anything, just like Source.lazyEmpty did
        promise.trySuccess(Some(()))

        // FIXME: After https://github.com/akka/akka/pull/18792 merged
        val ticks = Source(1.second, 3.seconds, "tick")
        //#source-creators

        //#flatten
        // Please note that the parenthesis is mandatory due to implicit parameters
        Flow[Source[Int, Any]].flattenConcat()
        //#flatten

        //#port-async
        class MapAsyncOne[In, Out](f: In ⇒ Future[Out])(implicit ec: ExecutionContext)
          extends GraphStage[FlowShape[In, Out]] {
          val in: Inlet[In] = Inlet("MapAsyncOne.in")
          val out: Outlet[Out] = Outlet("MapAsyncOne.out")
          override val shape: FlowShape[In, Out] = FlowShape(in, out)

          // The actual logic is encapsulated in a GraphStageLogic now
          override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
            new GraphStageLogic(shape) {

              // All of the state *must* be encapsulated in the GraphStageLogic,
              // not in the GraphStage
              private var elemInFlight: Out = _

              val callback = getAsyncCallback(onAsyncInput)
              var holdingUpstream = false

              // All upstream related events now are handled in an InHandler instance
              setHandler(in, new InHandler {
                // No context or element parameter for onPush
                override def onPush(): Unit = {
                  // The element is not passed as an argument but needs to be dequeued explicitly
                  val elem = grab(in)
                  val future = f(elem)
                  future.onComplete(callback.invoke)
                  // ctx.holdUpstream is no longer needed, but we need to track the state
                  holdingUpstream = true
                }

                // No context parameter
                override def onUpstreamFinish(): Unit = {
                  if (holdingUpstream) absorbTermination()
                  else completeStage() // ctx.finish turns into completeStage()
                }
              })

              setHandler(out, new OutHandler {
                override def onPull(): Unit = {
                  if (elemInFlight != null) {
                    val e = elemInFlight
                    elemInFlight = null.asInstanceOf[Out]
                    pushIt(e)
                  } // holdDownstream is no longer needed
                }
              })

              // absorbTermination turns into the code below.
              // This emulates the behavior of the AsyncStage stage.
              private def absorbTermination(): Unit =
                if (isAvailable(shape.outlet)) getHandler(out).onPull()

              // The line below emulates the behavior of the AsyncStage holdingDownstream
              private def holdingDownstream(): Boolean =
                !(isClosed(in) || hasBeenPulled(in))

              // Any method can be used as a callback, we chose the previous name for
              // easier comparison with the original code
              private def onAsyncInput(input: Try[Out]) =
                input match {
                  case Failure(ex)                       ⇒ failStage(ex)
                  case Success(e) if holdingDownstream() ⇒ pushIt(e)
                  case Success(e) ⇒
                    elemInFlight = e
                  // ctx.ignore is no longer needed
                }

              private def pushIt(elem: Out): Unit = {
                // ctx.isFinishing turns into isClosed(in)
                if (isClosed(in)) {
                  // pushAndFinish is now two actions
                  push(out, elem)
                  completeStage()
                } else {
                  // pushAndPull is now two actions
                  push(out, elem)
                  pull(in)
                  holdingUpstream = false
                }
              }
            }

        }

        //#port-async
      }
    }
  }

}