Akka streams partition either

EitherSourceExtension.scala

  implicit class EitherSourceExtension[L, R, Mat](source: Source[Either[L, R], Mat]) {

    def partition(left: Sink[L, NotUsed], right: Sink[R, NotUsed]): Graph[ClosedShape, NotUsed] = {
      GraphDSL.create() { implicit builder =>
        import akka.stream.scaladsl.GraphDSL.Implicits._
        val partition = builder.add(Partition[Either[L, R]](2, element => if (element.isLeft) 0 else 1))
        source ~> partition.in
        partition.out(0).collect {
          case Left(value) => value
        } ~> left
        partition.out(1).collect {
          case Right(value) => value
        } ~> right
        ClosedShape
      }
    }
  }