def fromSinkAndSource[I, O](sink: Graph[SinkShape[I], _], source: Graph[SourceShape[O], _]): Flow[I, O, NotUsed] =fromSinkAndSourceMat(sink, source)(Keep.none)

/*** Allows coupling termination (cancellation, completion, erroring) of Sinks and Sources while creating a Flow them them.* Similar to `Flow.fromSinkAndSource` however that API does not connect the completion signals of the wrapped stages.*/
object CoupledTerminationFlow {@deprecated("Use `Flow.fromSinkAndSourceCoupledMat(..., ...)(Keep.both)` instead", "2.5.2")def fromSinkAndSource[I, O, M1, M2](in: Sink[I, M1], out: Source[O, M2]): Flow[I, O, (M1, M2)] =Flow.fromSinkAndSourceCoupledMat(in, out)(Keep.both)
 

val nestedFlow =Flow[Int].filter(_ != 0) // an atomic processing stage.map(_ - 2) // another atomic processing stage.named("nestedFlow") // wraps up the Flow, and gives it a name

val nestedSink =nestedFlow.to(Sink.fold(0)(_ + _)) // wire an atomic sink to the nestedFlow.named("nestedSink") // wrap it up// Create a RunnableGraph
val runnableGraph = nestedSource.to(nestedSink)

import akka.actor._
import akka.stream._
import akka.stream.scaladsl._import scala.collection.immutableobject GraphModules {def someProcess[I, O]: I => O = i => i.asInstanceOf[O]case class TwoThreeShape[I, I2, O, O2, O3](in1: Inlet[I],in2: Inlet[I2],out1: Outlet[O],out2: Outlet[O2],out3: Outlet[O3]) extends Shape {override def inlets: immutable.Seq[Inlet[_]] = in1 :: in2 :: Niloverride def outlets: immutable.Seq[Outlet[_]] = out1 :: out2 :: out3 :: Niloverride def deepCopy(): Shape = TwoThreeShape(in1.carbonCopy(),in2.carbonCopy(),out1.carbonCopy(),out2.carbonCopy(),out3.carbonCopy())}
//a functional module with 2 input 3 outputdef TwoThreeGraph[I, I2, O, O2, O3] = GraphDSL.create() { implicit builder =>val balancer = builder.add(Balance[I](2))val flow = builder.add(Flow[I2].map(someProcess[I2, O2]))TwoThreeShape(balancer.in, flow.in, balancer.out(0), balancer.out(1), flow.out)}val closedGraph = GraphDSL.create() {implicit builder =>import GraphDSL.Implicits._val inp1 = builder.add(Source(List(1,2,3))).outval inp2 = builder.add(Source(List(10,20,30))).outval merge = builder.add(Merge[Int](2))val mod23 = builder.add(TwoThreeGraph[Int,Int,Int,Int,Int])inp1 ~> mod23.in1inp2 ~> mod23.in2mod23.out1 ~> merge.in(0)mod23.out2 ~> merge.in(1)mod23.out3 ~> Sink.foreach(println)merge ~> Sink.foreach(println)ClosedShape}
}object TailorGraph extends App {import GraphModules._implicit val sys = ActorSystem("streamSys")implicit val ec = sys.dispatcherimplicit val mat = ActorMaterializer()RunnableGraph.fromGraph(closedGraph).run()scala.io.StdIn.readLine()sys.terminate()}

      def ~>[Out](junction: UniformFanInShape[T, Out])(implicit b: Builder[_]): PortOps[Out] = {...}def ~>[Out](junction: UniformFanOutShape[T, Out])(implicit b: Builder[_]): PortOps[Out] = {...}def ~>[Out](flow: FlowShape[T, Out])(implicit b: Builder[_]): PortOps[Out] = {...}def ~>(to: Graph[SinkShape[T], _])(implicit b: Builder[_]): Unit =b.addEdge(importAndGetPort(b), b.add(to).in)def ~>(to: SinkShape[T])(implicit b: Builder[_]): Unit =b.addEdge(importAndGetPort(b), to.in)
...

   def ~>[U >: T](to: Inlet[U])(implicit b: Builder[_]): Unit =b.addEdge(importAndGetPort(b), to)

  import GraphDSL.Implicits._RunnableGraph.fromGraph(GraphDSL.create() { implicit builder =>val A: Outlet[Int]                  = builder.add(Source.single(0)).outval B: UniformFanOutShape[Int, Int] = builder.add(Broadcast[Int](2))val C: UniformFanInShape[Int, Int]  = builder.add(Merge[Int](2))val D: FlowShape[Int, Int]          = builder.add(Flow[Int].map(_ + 1))val E: UniformFanOutShape[Int, Int] = builder.add(Balance[Int](2))val F: UniformFanInShape[Int, Int]  = builder.add(Merge[Int](2))val G: Inlet[Any]                   = builder.add(Sink.foreach(println)).inC     <~      FA  ~>  B  ~>  C     ~>      FB  ~>  D  ~>  E  ~>  FE  ~>  GClosedShape})

RunnableGraph.fromGraph(GraphDSL.create() { implicit builder =>val B = builder.add(Broadcast[Int](2))val C = builder.add(Merge[Int](2))val E = builder.add(Balance[Int](2))val F = builder.add(Merge[Int](2))Source.single(0) ~> B.in; B.out(0) ~> C.in(1); C.out ~> F.in(0)C.in(0) <~ F.outB.out(1).map(_ + 1) ~> E.in; E.out(0) ~> F.in(1)E.out(1) ~> Sink.foreach(println)ClosedShape
})

val partial = GraphDSL.create() { implicit builder =>val B = builder.add(Broadcast[Int](2))val C = builder.add(Merge[Int](2))val E = builder.add(Balance[Int](2))val F = builder.add(Merge[Int](2))C  <~  FB  ~>                            C  ~>  FB  ~>  Flow[Int].map(_ + 1)  ~>  E  ~>  FFlowShape(B.in, E.out(1))}.named("partial")